1mod store;
2
3use crate::{
4 auth,
5 db::{self, ChannelId, MessageId, User, UserId},
6 AppState, Result,
7};
8use anyhow::anyhow;
9use async_tungstenite::tungstenite::{
10 protocol::CloseFrame as TungsteniteCloseFrame, Message as TungsteniteMessage,
11};
12use axum::{
13 body::Body,
14 extract::{
15 ws::{CloseFrame as AxumCloseFrame, Message as AxumMessage},
16 ConnectInfo, WebSocketUpgrade,
17 },
18 headers::{Header, HeaderName},
19 http::StatusCode,
20 middleware,
21 response::IntoResponse,
22 routing::get,
23 Extension, Router, TypedHeader,
24};
25use collections::HashMap;
26use futures::{channel::mpsc, future::BoxFuture, FutureExt, SinkExt, StreamExt, TryStreamExt};
27use lazy_static::lazy_static;
28use rpc::{
29 proto::{self, AnyTypedEnvelope, EntityMessage, EnvelopedMessage, RequestMessage},
30 Connection, ConnectionId, Peer, Receipt, TypedEnvelope,
31};
32use std::{
33 any::TypeId,
34 future::Future,
35 marker::PhantomData,
36 net::SocketAddr,
37 ops::{Deref, DerefMut},
38 rc::Rc,
39 sync::{
40 atomic::{AtomicBool, Ordering::SeqCst},
41 Arc,
42 },
43 time::Duration,
44};
45use store::{Store, Worktree};
46use time::OffsetDateTime;
47use tokio::{
48 sync::{RwLock, RwLockReadGuard, RwLockWriteGuard},
49 time::Sleep,
50};
51use tower::ServiceBuilder;
52use tracing::{info_span, instrument, Instrument};
53
54type MessageHandler =
55 Box<dyn Send + Sync + Fn(Arc<Server>, Box<dyn AnyTypedEnvelope>) -> BoxFuture<'static, ()>>;
56
57struct Response<R> {
58 server: Arc<Server>,
59 receipt: Receipt<R>,
60 responded: Arc<AtomicBool>,
61}
62
63impl<R: RequestMessage> Response<R> {
64 fn send(self, payload: R::Response) -> Result<()> {
65 self.responded.store(true, SeqCst);
66 self.server.peer.respond(self.receipt, payload)?;
67 Ok(())
68 }
69
70 fn into_receipt(self) -> Receipt<R> {
71 self.responded.store(true, SeqCst);
72 self.receipt
73 }
74}
75
76pub struct Server {
77 peer: Arc<Peer>,
78 store: RwLock<Store>,
79 app_state: Arc<AppState>,
80 handlers: HashMap<TypeId, MessageHandler>,
81 notifications: Option<mpsc::UnboundedSender<()>>,
82}
83
84pub trait Executor: Send + Clone {
85 type Sleep: Send + Future;
86 fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F);
87 fn sleep(&self, duration: Duration) -> Self::Sleep;
88}
89
90#[derive(Clone)]
91pub struct RealExecutor;
92
93const MESSAGE_COUNT_PER_PAGE: usize = 100;
94const MAX_MESSAGE_LEN: usize = 1024;
95
96struct StoreReadGuard<'a> {
97 guard: RwLockReadGuard<'a, Store>,
98 _not_send: PhantomData<Rc<()>>,
99}
100
101struct StoreWriteGuard<'a> {
102 guard: RwLockWriteGuard<'a, Store>,
103 _not_send: PhantomData<Rc<()>>,
104}
105
106impl Server {
107 pub fn new(
108 app_state: Arc<AppState>,
109 notifications: Option<mpsc::UnboundedSender<()>>,
110 ) -> Arc<Self> {
111 let mut server = Self {
112 peer: Peer::new(),
113 app_state,
114 store: Default::default(),
115 handlers: Default::default(),
116 notifications,
117 };
118
119 server
120 .add_request_handler(Server::ping)
121 .add_request_handler(Server::register_project)
122 .add_message_handler(Server::unregister_project)
123 .add_request_handler(Server::join_project)
124 .add_message_handler(Server::leave_project)
125 .add_message_handler(Server::respond_to_join_project_request)
126 .add_request_handler(Server::register_worktree)
127 .add_message_handler(Server::unregister_worktree)
128 .add_request_handler(Server::update_worktree)
129 .add_message_handler(Server::start_language_server)
130 .add_message_handler(Server::update_language_server)
131 .add_message_handler(Server::update_diagnostic_summary)
132 .add_request_handler(Server::forward_project_request::<proto::GetDefinition>)
133 .add_request_handler(Server::forward_project_request::<proto::GetReferences>)
134 .add_request_handler(Server::forward_project_request::<proto::SearchProject>)
135 .add_request_handler(Server::forward_project_request::<proto::GetDocumentHighlights>)
136 .add_request_handler(Server::forward_project_request::<proto::GetProjectSymbols>)
137 .add_request_handler(Server::forward_project_request::<proto::OpenBufferForSymbol>)
138 .add_request_handler(Server::forward_project_request::<proto::OpenBufferById>)
139 .add_request_handler(Server::forward_project_request::<proto::OpenBufferByPath>)
140 .add_request_handler(Server::forward_project_request::<proto::GetCompletions>)
141 .add_request_handler(
142 Server::forward_project_request::<proto::ApplyCompletionAdditionalEdits>,
143 )
144 .add_request_handler(Server::forward_project_request::<proto::GetCodeActions>)
145 .add_request_handler(Server::forward_project_request::<proto::ApplyCodeAction>)
146 .add_request_handler(Server::forward_project_request::<proto::PrepareRename>)
147 .add_request_handler(Server::forward_project_request::<proto::PerformRename>)
148 .add_request_handler(Server::forward_project_request::<proto::ReloadBuffers>)
149 .add_request_handler(Server::forward_project_request::<proto::FormatBuffers>)
150 .add_request_handler(Server::forward_project_request::<proto::CreateProjectEntry>)
151 .add_request_handler(Server::forward_project_request::<proto::RenameProjectEntry>)
152 .add_request_handler(Server::forward_project_request::<proto::DeleteProjectEntry>)
153 .add_request_handler(Server::update_buffer)
154 .add_message_handler(Server::update_buffer_file)
155 .add_message_handler(Server::buffer_reloaded)
156 .add_message_handler(Server::buffer_saved)
157 .add_request_handler(Server::save_buffer)
158 .add_request_handler(Server::get_channels)
159 .add_request_handler(Server::get_users)
160 .add_request_handler(Server::fuzzy_search_users)
161 .add_request_handler(Server::request_contact)
162 .add_request_handler(Server::remove_contact)
163 .add_request_handler(Server::respond_to_contact_request)
164 .add_request_handler(Server::join_channel)
165 .add_message_handler(Server::leave_channel)
166 .add_request_handler(Server::send_channel_message)
167 .add_request_handler(Server::follow)
168 .add_message_handler(Server::unfollow)
169 .add_message_handler(Server::update_followers)
170 .add_request_handler(Server::get_channel_messages);
171
172 Arc::new(server)
173 }
174
175 fn add_message_handler<F, Fut, M>(&mut self, handler: F) -> &mut Self
176 where
177 F: 'static + Send + Sync + Fn(Arc<Self>, TypedEnvelope<M>) -> Fut,
178 Fut: 'static + Send + Future<Output = Result<()>>,
179 M: EnvelopedMessage,
180 {
181 let prev_handler = self.handlers.insert(
182 TypeId::of::<M>(),
183 Box::new(move |server, envelope| {
184 let envelope = envelope.into_any().downcast::<TypedEnvelope<M>>().unwrap();
185 let span = info_span!(
186 "handle message",
187 payload_type = envelope.payload_type_name(),
188 payload = format!("{:?}", envelope.payload).as_str(),
189 );
190 let future = (handler)(server, *envelope);
191 async move {
192 if let Err(error) = future.await {
193 tracing::error!(%error, "error handling message");
194 }
195 }
196 .instrument(span)
197 .boxed()
198 }),
199 );
200 if prev_handler.is_some() {
201 panic!("registered a handler for the same message twice");
202 }
203 self
204 }
205
206 /// Handle a request while holding a lock to the store. This is useful when we're registering
207 /// a connection but we want to respond on the connection before anybody else can send on it.
208 fn add_request_handler<F, Fut, M>(&mut self, handler: F) -> &mut Self
209 where
210 F: 'static + Send + Sync + Fn(Arc<Self>, TypedEnvelope<M>, Response<M>) -> Fut,
211 Fut: Send + Future<Output = Result<()>>,
212 M: RequestMessage,
213 {
214 let handler = Arc::new(handler);
215 self.add_message_handler(move |server, envelope| {
216 let receipt = envelope.receipt();
217 let handler = handler.clone();
218 async move {
219 let responded = Arc::new(AtomicBool::default());
220 let response = Response {
221 server: server.clone(),
222 responded: responded.clone(),
223 receipt: envelope.receipt(),
224 };
225 match (handler)(server.clone(), envelope, response).await {
226 Ok(()) => {
227 if responded.load(std::sync::atomic::Ordering::SeqCst) {
228 Ok(())
229 } else {
230 Err(anyhow!("handler did not send a response"))?
231 }
232 }
233 Err(error) => {
234 server.peer.respond_with_error(
235 receipt,
236 proto::Error {
237 message: error.to_string(),
238 },
239 )?;
240 Err(error)
241 }
242 }
243 }
244 })
245 }
246
247 pub fn handle_connection<E: Executor>(
248 self: &Arc<Self>,
249 connection: Connection,
250 address: String,
251 user: User,
252 mut send_connection_id: Option<mpsc::Sender<ConnectionId>>,
253 executor: E,
254 ) -> impl Future<Output = Result<()>> {
255 let mut this = self.clone();
256 let user_id = user.id;
257 let login = user.github_login;
258 let span = info_span!("handle connection", %user_id, %login, %address);
259 async move {
260 let (connection_id, handle_io, mut incoming_rx) = this
261 .peer
262 .add_connection(connection, {
263 let executor = executor.clone();
264 move |duration| {
265 let timer = executor.sleep(duration);
266 async move {
267 timer.await;
268 }
269 }
270 })
271 .await;
272
273 tracing::info!(%user_id, %login, %connection_id, %address, "connection opened");
274
275 if let Some(send_connection_id) = send_connection_id.as_mut() {
276 let _ = send_connection_id.send(connection_id).await;
277 }
278
279 let contacts = this.app_state.db.get_contacts(user_id).await?;
280
281 {
282 let mut store = this.store_mut().await;
283 store.add_connection(connection_id, user_id);
284 this.peer.send(connection_id, store.build_initial_contacts_update(contacts))?;
285 }
286 this.update_user_contacts(user_id).await?;
287
288 let handle_io = handle_io.fuse();
289 futures::pin_mut!(handle_io);
290 loop {
291 let next_message = incoming_rx.next().fuse();
292 futures::pin_mut!(next_message);
293 futures::select_biased! {
294 result = handle_io => {
295 if let Err(error) = result {
296 tracing::error!(?error, %user_id, %login, %connection_id, %address, "error handling I/O");
297 }
298 break;
299 }
300 message = next_message => {
301 if let Some(message) = message {
302 let type_name = message.payload_type_name();
303 let span = tracing::info_span!("receive message", %user_id, %login, %connection_id, %address, type_name);
304 async {
305 if let Some(handler) = this.handlers.get(&message.payload_type_id()) {
306 let notifications = this.notifications.clone();
307 let is_background = message.is_background();
308 let handle_message = (handler)(this.clone(), message);
309 let handle_message = async move {
310 handle_message.await;
311 if let Some(mut notifications) = notifications {
312 let _ = notifications.send(()).await;
313 }
314 };
315 if is_background {
316 executor.spawn_detached(handle_message);
317 } else {
318 handle_message.await;
319 }
320 } else {
321 tracing::error!(%user_id, %login, %connection_id, %address, "no message handler");
322 }
323 }.instrument(span).await;
324 } else {
325 tracing::info!(%user_id, %login, %connection_id, %address, "connection closed");
326 break;
327 }
328 }
329 }
330 }
331
332 tracing::info!(%user_id, %login, %connection_id, %address, "signing out");
333 if let Err(error) = this.sign_out(connection_id).await {
334 tracing::error!(%user_id, %login, %connection_id, %address, ?error, "error signing out");
335 }
336
337 Ok(())
338 }.instrument(span)
339 }
340
341 #[instrument(skip(self), err)]
342 async fn sign_out(self: &mut Arc<Self>, connection_id: ConnectionId) -> Result<()> {
343 self.peer.disconnect(connection_id);
344
345 let removed_user_id = {
346 let mut store = self.store_mut().await;
347 let removed_connection = store.remove_connection(connection_id)?;
348
349 for (project_id, project) in removed_connection.hosted_projects {
350 broadcast(connection_id, project.guests.keys().copied(), |conn_id| {
351 self.peer
352 .send(conn_id, proto::UnregisterProject { project_id })
353 });
354
355 for (_, receipts) in project.join_requests {
356 for receipt in receipts {
357 self.peer.respond(
358 receipt,
359 proto::JoinProjectResponse {
360 variant: Some(proto::join_project_response::Variant::Decline(
361 proto::join_project_response::Decline {
362 reason: proto::join_project_response::decline::Reason::WentOffline as i32
363 },
364 )),
365 },
366 )?;
367 }
368 }
369 }
370
371 for project_id in removed_connection.guest_project_ids {
372 if let Some(project) = store.project(project_id).trace_err() {
373 broadcast(connection_id, project.connection_ids(), |conn_id| {
374 self.peer.send(
375 conn_id,
376 proto::RemoveProjectCollaborator {
377 project_id,
378 peer_id: connection_id.0,
379 },
380 )
381 });
382 if project.guests.is_empty() {
383 self.peer
384 .send(
385 project.host_connection_id,
386 proto::ProjectUnshared { project_id },
387 )
388 .trace_err();
389 }
390 }
391 }
392
393 removed_connection.user_id
394 };
395
396 self.update_user_contacts(removed_user_id).await?;
397
398 Ok(())
399 }
400
401 async fn ping(
402 self: Arc<Server>,
403 _: TypedEnvelope<proto::Ping>,
404 response: Response<proto::Ping>,
405 ) -> Result<()> {
406 response.send(proto::Ack {})?;
407 Ok(())
408 }
409
410 async fn register_project(
411 self: Arc<Server>,
412 request: TypedEnvelope<proto::RegisterProject>,
413 response: Response<proto::RegisterProject>,
414 ) -> Result<()> {
415 let user_id;
416 let project_id;
417 {
418 let mut state = self.store_mut().await;
419 user_id = state.user_id_for_connection(request.sender_id)?;
420 project_id = state.register_project(request.sender_id, user_id);
421 };
422 self.update_user_contacts(user_id).await?;
423 response.send(proto::RegisterProjectResponse { project_id })?;
424 Ok(())
425 }
426
427 async fn unregister_project(
428 self: Arc<Server>,
429 request: TypedEnvelope<proto::UnregisterProject>,
430 ) -> Result<()> {
431 let (user_id, project) = {
432 let mut state = self.store_mut().await;
433 let project =
434 state.unregister_project(request.payload.project_id, request.sender_id)?;
435 (state.user_id_for_connection(request.sender_id)?, project)
436 };
437 for (_, receipts) in project.join_requests {
438 for receipt in receipts {
439 self.peer.respond(
440 receipt,
441 proto::JoinProjectResponse {
442 variant: Some(proto::join_project_response::Variant::Decline(
443 proto::join_project_response::Decline {
444 reason: proto::join_project_response::decline::Reason::Closed
445 as i32,
446 },
447 )),
448 },
449 )?;
450 }
451 }
452
453 self.update_user_contacts(user_id).await?;
454 Ok(())
455 }
456
457 async fn update_user_contacts(self: &Arc<Server>, user_id: UserId) -> Result<()> {
458 let contacts = self.app_state.db.get_contacts(user_id).await?;
459 let store = self.store().await;
460 let updated_contact = store.contact_for_user(user_id, false);
461 for contact in contacts {
462 if let db::Contact::Accepted {
463 user_id: contact_user_id,
464 ..
465 } = contact
466 {
467 for contact_conn_id in store.connection_ids_for_user(contact_user_id) {
468 self.peer
469 .send(
470 contact_conn_id,
471 proto::UpdateContacts {
472 contacts: vec![updated_contact.clone()],
473 remove_contacts: Default::default(),
474 incoming_requests: Default::default(),
475 remove_incoming_requests: Default::default(),
476 outgoing_requests: Default::default(),
477 remove_outgoing_requests: Default::default(),
478 },
479 )
480 .trace_err();
481 }
482 }
483 }
484 Ok(())
485 }
486
487 async fn join_project(
488 self: Arc<Server>,
489 request: TypedEnvelope<proto::JoinProject>,
490 response: Response<proto::JoinProject>,
491 ) -> Result<()> {
492 let project_id = request.payload.project_id;
493 let host_user_id;
494 let guest_user_id;
495 let host_connection_id;
496 {
497 let state = self.store().await;
498 let project = state.project(project_id)?;
499 host_user_id = project.host_user_id;
500 host_connection_id = project.host_connection_id;
501 guest_user_id = state.user_id_for_connection(request.sender_id)?;
502 };
503
504 let has_contact = self
505 .app_state
506 .db
507 .has_contact(guest_user_id, host_user_id)
508 .await?;
509 if !has_contact {
510 return Err(anyhow!("no such project"))?;
511 }
512
513 self.store_mut().await.request_join_project(
514 guest_user_id,
515 project_id,
516 response.into_receipt(),
517 )?;
518 self.peer.send(
519 host_connection_id,
520 proto::RequestJoinProject {
521 project_id,
522 requester_id: guest_user_id.to_proto(),
523 },
524 )?;
525 Ok(())
526 }
527
528 async fn respond_to_join_project_request(
529 self: Arc<Server>,
530 request: TypedEnvelope<proto::RespondToJoinProjectRequest>,
531 ) -> Result<()> {
532 let host_user_id;
533
534 {
535 let mut state = self.store_mut().await;
536 let project_id = request.payload.project_id;
537 let project = state.project(project_id)?;
538 if project.host_connection_id != request.sender_id {
539 Err(anyhow!("no such connection"))?;
540 }
541
542 host_user_id = project.host_user_id;
543 let guest_user_id = UserId::from_proto(request.payload.requester_id);
544
545 if !request.payload.allow {
546 let receipts = state
547 .deny_join_project_request(request.sender_id, guest_user_id, project_id)
548 .ok_or_else(|| anyhow!("no such request"))?;
549 for receipt in receipts {
550 self.peer.respond(
551 receipt,
552 proto::JoinProjectResponse {
553 variant: Some(proto::join_project_response::Variant::Decline(
554 proto::join_project_response::Decline {
555 reason: proto::join_project_response::decline::Reason::Declined
556 as i32,
557 },
558 )),
559 },
560 )?;
561 }
562 return Ok(());
563 }
564
565 let (receipts_with_replica_ids, project) = state
566 .accept_join_project_request(request.sender_id, guest_user_id, project_id)
567 .ok_or_else(|| anyhow!("no such request"))?;
568
569 let peer_count = project.guests.len();
570 let mut collaborators = Vec::with_capacity(peer_count);
571 collaborators.push(proto::Collaborator {
572 peer_id: project.host_connection_id.0,
573 replica_id: 0,
574 user_id: project.host_user_id.to_proto(),
575 });
576 let worktrees = project
577 .worktrees
578 .iter()
579 .filter_map(|(id, shared_worktree)| {
580 let worktree = project.worktrees.get(&id)?;
581 Some(proto::Worktree {
582 id: *id,
583 root_name: worktree.root_name.clone(),
584 entries: shared_worktree.entries.values().cloned().collect(),
585 diagnostic_summaries: shared_worktree
586 .diagnostic_summaries
587 .values()
588 .cloned()
589 .collect(),
590 visible: worktree.visible,
591 scan_id: shared_worktree.scan_id,
592 })
593 })
594 .collect::<Vec<_>>();
595
596 // Add all guests other than the requesting user's own connections as collaborators
597 for (peer_conn_id, (peer_replica_id, peer_user_id)) in &project.guests {
598 if receipts_with_replica_ids
599 .iter()
600 .all(|(receipt, _)| receipt.sender_id != *peer_conn_id)
601 {
602 collaborators.push(proto::Collaborator {
603 peer_id: peer_conn_id.0,
604 replica_id: *peer_replica_id as u32,
605 user_id: peer_user_id.to_proto(),
606 });
607 }
608 }
609
610 for conn_id in project.connection_ids() {
611 for (receipt, replica_id) in &receipts_with_replica_ids {
612 if conn_id != receipt.sender_id {
613 self.peer.send(
614 conn_id,
615 proto::AddProjectCollaborator {
616 project_id,
617 collaborator: Some(proto::Collaborator {
618 peer_id: receipt.sender_id.0,
619 replica_id: *replica_id as u32,
620 user_id: guest_user_id.to_proto(),
621 }),
622 },
623 )?;
624 }
625 }
626 }
627
628 for (receipt, replica_id) in receipts_with_replica_ids {
629 self.peer.respond(
630 receipt,
631 proto::JoinProjectResponse {
632 variant: Some(proto::join_project_response::Variant::Accept(
633 proto::join_project_response::Accept {
634 worktrees: worktrees.clone(),
635 replica_id: replica_id as u32,
636 collaborators: collaborators.clone(),
637 language_servers: project.language_servers.clone(),
638 },
639 )),
640 },
641 )?;
642 }
643 }
644
645 self.update_user_contacts(host_user_id).await?;
646 Ok(())
647 }
648
649 async fn leave_project(
650 self: Arc<Server>,
651 request: TypedEnvelope<proto::LeaveProject>,
652 ) -> Result<()> {
653 let sender_id = request.sender_id;
654 let project_id = request.payload.project_id;
655 let project;
656 {
657 let mut store = self.store_mut().await;
658 project = store.leave_project(sender_id, project_id)?;
659
660 if project.remove_collaborator {
661 broadcast(sender_id, project.connection_ids, |conn_id| {
662 self.peer.send(
663 conn_id,
664 proto::RemoveProjectCollaborator {
665 project_id,
666 peer_id: sender_id.0,
667 },
668 )
669 });
670 }
671
672 if let Some(requester_id) = project.cancel_request {
673 self.peer.send(
674 project.host_connection_id,
675 proto::JoinProjectRequestCancelled {
676 project_id,
677 requester_id: requester_id.to_proto(),
678 },
679 )?;
680 }
681
682 if project.unshare {
683 self.peer.send(
684 project.host_connection_id,
685 proto::ProjectUnshared { project_id },
686 )?;
687 }
688 }
689 self.update_user_contacts(project.host_user_id).await?;
690 Ok(())
691 }
692
693 async fn register_worktree(
694 self: Arc<Server>,
695 request: TypedEnvelope<proto::RegisterWorktree>,
696 response: Response<proto::RegisterWorktree>,
697 ) -> Result<()> {
698 let host_user_id;
699 {
700 let mut state = self.store_mut().await;
701 host_user_id = state.user_id_for_connection(request.sender_id)?;
702
703 let guest_connection_ids = state
704 .read_project(request.payload.project_id, request.sender_id)?
705 .guest_connection_ids();
706 state.register_worktree(
707 request.payload.project_id,
708 request.payload.worktree_id,
709 request.sender_id,
710 Worktree {
711 root_name: request.payload.root_name.clone(),
712 visible: request.payload.visible,
713 ..Default::default()
714 },
715 )?;
716
717 broadcast(request.sender_id, guest_connection_ids, |connection_id| {
718 self.peer
719 .forward_send(request.sender_id, connection_id, request.payload.clone())
720 });
721 }
722 self.update_user_contacts(host_user_id).await?;
723 response.send(proto::Ack {})?;
724 Ok(())
725 }
726
727 async fn unregister_worktree(
728 self: Arc<Server>,
729 request: TypedEnvelope<proto::UnregisterWorktree>,
730 ) -> Result<()> {
731 let host_user_id;
732 let project_id = request.payload.project_id;
733 let worktree_id = request.payload.worktree_id;
734 {
735 let mut state = self.store_mut().await;
736 let (_, guest_connection_ids) =
737 state.unregister_worktree(project_id, worktree_id, request.sender_id)?;
738 host_user_id = state.user_id_for_connection(request.sender_id)?;
739 broadcast(request.sender_id, guest_connection_ids, |conn_id| {
740 self.peer.send(
741 conn_id,
742 proto::UnregisterWorktree {
743 project_id,
744 worktree_id,
745 },
746 )
747 });
748 }
749 self.update_user_contacts(host_user_id).await?;
750 Ok(())
751 }
752
753 async fn update_worktree(
754 self: Arc<Server>,
755 request: TypedEnvelope<proto::UpdateWorktree>,
756 response: Response<proto::UpdateWorktree>,
757 ) -> Result<()> {
758 let connection_ids = self.store_mut().await.update_worktree(
759 request.sender_id,
760 request.payload.project_id,
761 request.payload.worktree_id,
762 &request.payload.removed_entries,
763 &request.payload.updated_entries,
764 request.payload.scan_id,
765 )?;
766
767 broadcast(request.sender_id, connection_ids, |connection_id| {
768 self.peer
769 .forward_send(request.sender_id, connection_id, request.payload.clone())
770 });
771 response.send(proto::Ack {})?;
772 Ok(())
773 }
774
775 async fn update_diagnostic_summary(
776 self: Arc<Server>,
777 request: TypedEnvelope<proto::UpdateDiagnosticSummary>,
778 ) -> Result<()> {
779 let summary = request
780 .payload
781 .summary
782 .clone()
783 .ok_or_else(|| anyhow!("invalid summary"))?;
784 let receiver_ids = self.store_mut().await.update_diagnostic_summary(
785 request.payload.project_id,
786 request.payload.worktree_id,
787 request.sender_id,
788 summary,
789 )?;
790
791 broadcast(request.sender_id, receiver_ids, |connection_id| {
792 self.peer
793 .forward_send(request.sender_id, connection_id, request.payload.clone())
794 });
795 Ok(())
796 }
797
798 async fn start_language_server(
799 self: Arc<Server>,
800 request: TypedEnvelope<proto::StartLanguageServer>,
801 ) -> Result<()> {
802 let receiver_ids = self.store_mut().await.start_language_server(
803 request.payload.project_id,
804 request.sender_id,
805 request
806 .payload
807 .server
808 .clone()
809 .ok_or_else(|| anyhow!("invalid language server"))?,
810 )?;
811 broadcast(request.sender_id, receiver_ids, |connection_id| {
812 self.peer
813 .forward_send(request.sender_id, connection_id, request.payload.clone())
814 });
815 Ok(())
816 }
817
818 async fn update_language_server(
819 self: Arc<Server>,
820 request: TypedEnvelope<proto::UpdateLanguageServer>,
821 ) -> Result<()> {
822 let receiver_ids = self
823 .store()
824 .await
825 .project_connection_ids(request.payload.project_id, request.sender_id)?;
826 broadcast(request.sender_id, receiver_ids, |connection_id| {
827 self.peer
828 .forward_send(request.sender_id, connection_id, request.payload.clone())
829 });
830 Ok(())
831 }
832
833 async fn forward_project_request<T>(
834 self: Arc<Server>,
835 request: TypedEnvelope<T>,
836 response: Response<T>,
837 ) -> Result<()>
838 where
839 T: EntityMessage + RequestMessage,
840 {
841 let host_connection_id = self
842 .store()
843 .await
844 .read_project(request.payload.remote_entity_id(), request.sender_id)?
845 .host_connection_id;
846
847 response.send(
848 self.peer
849 .forward_request(request.sender_id, host_connection_id, request.payload)
850 .await?,
851 )?;
852 Ok(())
853 }
854
855 async fn save_buffer(
856 self: Arc<Server>,
857 request: TypedEnvelope<proto::SaveBuffer>,
858 response: Response<proto::SaveBuffer>,
859 ) -> Result<()> {
860 let host = self
861 .store()
862 .await
863 .read_project(request.payload.project_id, request.sender_id)?
864 .host_connection_id;
865 let response_payload = self
866 .peer
867 .forward_request(request.sender_id, host, request.payload.clone())
868 .await?;
869
870 let mut guests = self
871 .store()
872 .await
873 .read_project(request.payload.project_id, request.sender_id)?
874 .connection_ids();
875 guests.retain(|guest_connection_id| *guest_connection_id != request.sender_id);
876 broadcast(host, guests, |conn_id| {
877 self.peer
878 .forward_send(host, conn_id, response_payload.clone())
879 });
880 response.send(response_payload)?;
881 Ok(())
882 }
883
884 async fn update_buffer(
885 self: Arc<Server>,
886 request: TypedEnvelope<proto::UpdateBuffer>,
887 response: Response<proto::UpdateBuffer>,
888 ) -> Result<()> {
889 let receiver_ids = self
890 .store()
891 .await
892 .project_connection_ids(request.payload.project_id, request.sender_id)?;
893 broadcast(request.sender_id, receiver_ids, |connection_id| {
894 self.peer
895 .forward_send(request.sender_id, connection_id, request.payload.clone())
896 });
897 response.send(proto::Ack {})?;
898 Ok(())
899 }
900
901 async fn update_buffer_file(
902 self: Arc<Server>,
903 request: TypedEnvelope<proto::UpdateBufferFile>,
904 ) -> Result<()> {
905 let receiver_ids = self
906 .store()
907 .await
908 .project_connection_ids(request.payload.project_id, request.sender_id)?;
909 broadcast(request.sender_id, receiver_ids, |connection_id| {
910 self.peer
911 .forward_send(request.sender_id, connection_id, request.payload.clone())
912 });
913 Ok(())
914 }
915
916 async fn buffer_reloaded(
917 self: Arc<Server>,
918 request: TypedEnvelope<proto::BufferReloaded>,
919 ) -> Result<()> {
920 let receiver_ids = self
921 .store()
922 .await
923 .project_connection_ids(request.payload.project_id, request.sender_id)?;
924 broadcast(request.sender_id, receiver_ids, |connection_id| {
925 self.peer
926 .forward_send(request.sender_id, connection_id, request.payload.clone())
927 });
928 Ok(())
929 }
930
931 async fn buffer_saved(
932 self: Arc<Server>,
933 request: TypedEnvelope<proto::BufferSaved>,
934 ) -> Result<()> {
935 let receiver_ids = self
936 .store()
937 .await
938 .project_connection_ids(request.payload.project_id, request.sender_id)?;
939 broadcast(request.sender_id, receiver_ids, |connection_id| {
940 self.peer
941 .forward_send(request.sender_id, connection_id, request.payload.clone())
942 });
943 Ok(())
944 }
945
946 async fn follow(
947 self: Arc<Self>,
948 request: TypedEnvelope<proto::Follow>,
949 response: Response<proto::Follow>,
950 ) -> Result<()> {
951 let leader_id = ConnectionId(request.payload.leader_id);
952 let follower_id = request.sender_id;
953 if !self
954 .store()
955 .await
956 .project_connection_ids(request.payload.project_id, follower_id)?
957 .contains(&leader_id)
958 {
959 Err(anyhow!("no such peer"))?;
960 }
961 let mut response_payload = self
962 .peer
963 .forward_request(request.sender_id, leader_id, request.payload)
964 .await?;
965 response_payload
966 .views
967 .retain(|view| view.leader_id != Some(follower_id.0));
968 response.send(response_payload)?;
969 Ok(())
970 }
971
972 async fn unfollow(self: Arc<Self>, request: TypedEnvelope<proto::Unfollow>) -> Result<()> {
973 let leader_id = ConnectionId(request.payload.leader_id);
974 if !self
975 .store()
976 .await
977 .project_connection_ids(request.payload.project_id, request.sender_id)?
978 .contains(&leader_id)
979 {
980 Err(anyhow!("no such peer"))?;
981 }
982 self.peer
983 .forward_send(request.sender_id, leader_id, request.payload)?;
984 Ok(())
985 }
986
987 async fn update_followers(
988 self: Arc<Self>,
989 request: TypedEnvelope<proto::UpdateFollowers>,
990 ) -> Result<()> {
991 let connection_ids = self
992 .store()
993 .await
994 .project_connection_ids(request.payload.project_id, request.sender_id)?;
995 let leader_id = request
996 .payload
997 .variant
998 .as_ref()
999 .and_then(|variant| match variant {
1000 proto::update_followers::Variant::CreateView(payload) => payload.leader_id,
1001 proto::update_followers::Variant::UpdateView(payload) => payload.leader_id,
1002 proto::update_followers::Variant::UpdateActiveView(payload) => payload.leader_id,
1003 });
1004 for follower_id in &request.payload.follower_ids {
1005 let follower_id = ConnectionId(*follower_id);
1006 if connection_ids.contains(&follower_id) && Some(follower_id.0) != leader_id {
1007 self.peer
1008 .forward_send(request.sender_id, follower_id, request.payload.clone())?;
1009 }
1010 }
1011 Ok(())
1012 }
1013
1014 async fn get_channels(
1015 self: Arc<Server>,
1016 request: TypedEnvelope<proto::GetChannels>,
1017 response: Response<proto::GetChannels>,
1018 ) -> Result<()> {
1019 let user_id = self
1020 .store()
1021 .await
1022 .user_id_for_connection(request.sender_id)?;
1023 let channels = self.app_state.db.get_accessible_channels(user_id).await?;
1024 response.send(proto::GetChannelsResponse {
1025 channels: channels
1026 .into_iter()
1027 .map(|chan| proto::Channel {
1028 id: chan.id.to_proto(),
1029 name: chan.name,
1030 })
1031 .collect(),
1032 })?;
1033 Ok(())
1034 }
1035
1036 async fn get_users(
1037 self: Arc<Server>,
1038 request: TypedEnvelope<proto::GetUsers>,
1039 response: Response<proto::GetUsers>,
1040 ) -> Result<()> {
1041 let user_ids = request
1042 .payload
1043 .user_ids
1044 .into_iter()
1045 .map(UserId::from_proto)
1046 .collect();
1047 let users = self
1048 .app_state
1049 .db
1050 .get_users_by_ids(user_ids)
1051 .await?
1052 .into_iter()
1053 .map(|user| proto::User {
1054 id: user.id.to_proto(),
1055 avatar_url: format!("https://github.com/{}.png?size=128", user.github_login),
1056 github_login: user.github_login,
1057 })
1058 .collect();
1059 response.send(proto::UsersResponse { users })?;
1060 Ok(())
1061 }
1062
1063 async fn fuzzy_search_users(
1064 self: Arc<Server>,
1065 request: TypedEnvelope<proto::FuzzySearchUsers>,
1066 response: Response<proto::FuzzySearchUsers>,
1067 ) -> Result<()> {
1068 let user_id = self
1069 .store()
1070 .await
1071 .user_id_for_connection(request.sender_id)?;
1072 let query = request.payload.query;
1073 let db = &self.app_state.db;
1074 let users = match query.len() {
1075 0 => vec![],
1076 1 | 2 => db
1077 .get_user_by_github_login(&query)
1078 .await?
1079 .into_iter()
1080 .collect(),
1081 _ => db.fuzzy_search_users(&query, 10).await?,
1082 };
1083 let users = users
1084 .into_iter()
1085 .filter(|user| user.id != user_id)
1086 .map(|user| proto::User {
1087 id: user.id.to_proto(),
1088 avatar_url: format!("https://github.com/{}.png?size=128", user.github_login),
1089 github_login: user.github_login,
1090 })
1091 .collect();
1092 response.send(proto::UsersResponse { users })?;
1093 Ok(())
1094 }
1095
1096 async fn request_contact(
1097 self: Arc<Server>,
1098 request: TypedEnvelope<proto::RequestContact>,
1099 response: Response<proto::RequestContact>,
1100 ) -> Result<()> {
1101 let requester_id = self
1102 .store()
1103 .await
1104 .user_id_for_connection(request.sender_id)?;
1105 let responder_id = UserId::from_proto(request.payload.responder_id);
1106 if requester_id == responder_id {
1107 return Err(anyhow!("cannot add yourself as a contact"))?;
1108 }
1109
1110 self.app_state
1111 .db
1112 .send_contact_request(requester_id, responder_id)
1113 .await?;
1114
1115 // Update outgoing contact requests of requester
1116 let mut update = proto::UpdateContacts::default();
1117 update.outgoing_requests.push(responder_id.to_proto());
1118 for connection_id in self.store().await.connection_ids_for_user(requester_id) {
1119 self.peer.send(connection_id, update.clone())?;
1120 }
1121
1122 // Update incoming contact requests of responder
1123 let mut update = proto::UpdateContacts::default();
1124 update
1125 .incoming_requests
1126 .push(proto::IncomingContactRequest {
1127 requester_id: requester_id.to_proto(),
1128 should_notify: true,
1129 });
1130 for connection_id in self.store().await.connection_ids_for_user(responder_id) {
1131 self.peer.send(connection_id, update.clone())?;
1132 }
1133
1134 response.send(proto::Ack {})?;
1135 Ok(())
1136 }
1137
1138 async fn respond_to_contact_request(
1139 self: Arc<Server>,
1140 request: TypedEnvelope<proto::RespondToContactRequest>,
1141 response: Response<proto::RespondToContactRequest>,
1142 ) -> Result<()> {
1143 let responder_id = self
1144 .store()
1145 .await
1146 .user_id_for_connection(request.sender_id)?;
1147 let requester_id = UserId::from_proto(request.payload.requester_id);
1148 if request.payload.response == proto::ContactRequestResponse::Dismiss as i32 {
1149 self.app_state
1150 .db
1151 .dismiss_contact_notification(responder_id, requester_id)
1152 .await?;
1153 } else {
1154 let accept = request.payload.response == proto::ContactRequestResponse::Accept as i32;
1155 self.app_state
1156 .db
1157 .respond_to_contact_request(responder_id, requester_id, accept)
1158 .await?;
1159
1160 let store = self.store().await;
1161 // Update responder with new contact
1162 let mut update = proto::UpdateContacts::default();
1163 if accept {
1164 update
1165 .contacts
1166 .push(store.contact_for_user(requester_id, false));
1167 }
1168 update
1169 .remove_incoming_requests
1170 .push(requester_id.to_proto());
1171 for connection_id in store.connection_ids_for_user(responder_id) {
1172 self.peer.send(connection_id, update.clone())?;
1173 }
1174
1175 // Update requester with new contact
1176 let mut update = proto::UpdateContacts::default();
1177 if accept {
1178 update
1179 .contacts
1180 .push(store.contact_for_user(responder_id, true));
1181 }
1182 update
1183 .remove_outgoing_requests
1184 .push(responder_id.to_proto());
1185 for connection_id in store.connection_ids_for_user(requester_id) {
1186 self.peer.send(connection_id, update.clone())?;
1187 }
1188 }
1189
1190 response.send(proto::Ack {})?;
1191 Ok(())
1192 }
1193
1194 async fn remove_contact(
1195 self: Arc<Server>,
1196 request: TypedEnvelope<proto::RemoveContact>,
1197 response: Response<proto::RemoveContact>,
1198 ) -> Result<()> {
1199 let requester_id = self
1200 .store()
1201 .await
1202 .user_id_for_connection(request.sender_id)?;
1203 let responder_id = UserId::from_proto(request.payload.user_id);
1204 self.app_state
1205 .db
1206 .remove_contact(requester_id, responder_id)
1207 .await?;
1208
1209 // Update outgoing contact requests of requester
1210 let mut update = proto::UpdateContacts::default();
1211 update
1212 .remove_outgoing_requests
1213 .push(responder_id.to_proto());
1214 for connection_id in self.store().await.connection_ids_for_user(requester_id) {
1215 self.peer.send(connection_id, update.clone())?;
1216 }
1217
1218 // Update incoming contact requests of responder
1219 let mut update = proto::UpdateContacts::default();
1220 update
1221 .remove_incoming_requests
1222 .push(requester_id.to_proto());
1223 for connection_id in self.store().await.connection_ids_for_user(responder_id) {
1224 self.peer.send(connection_id, update.clone())?;
1225 }
1226
1227 response.send(proto::Ack {})?;
1228 Ok(())
1229 }
1230
1231 async fn join_channel(
1232 self: Arc<Self>,
1233 request: TypedEnvelope<proto::JoinChannel>,
1234 response: Response<proto::JoinChannel>,
1235 ) -> Result<()> {
1236 let user_id = self
1237 .store()
1238 .await
1239 .user_id_for_connection(request.sender_id)?;
1240 let channel_id = ChannelId::from_proto(request.payload.channel_id);
1241 if !self
1242 .app_state
1243 .db
1244 .can_user_access_channel(user_id, channel_id)
1245 .await?
1246 {
1247 Err(anyhow!("access denied"))?;
1248 }
1249
1250 self.store_mut()
1251 .await
1252 .join_channel(request.sender_id, channel_id);
1253 let messages = self
1254 .app_state
1255 .db
1256 .get_channel_messages(channel_id, MESSAGE_COUNT_PER_PAGE, None)
1257 .await?
1258 .into_iter()
1259 .map(|msg| proto::ChannelMessage {
1260 id: msg.id.to_proto(),
1261 body: msg.body,
1262 timestamp: msg.sent_at.unix_timestamp() as u64,
1263 sender_id: msg.sender_id.to_proto(),
1264 nonce: Some(msg.nonce.as_u128().into()),
1265 })
1266 .collect::<Vec<_>>();
1267 response.send(proto::JoinChannelResponse {
1268 done: messages.len() < MESSAGE_COUNT_PER_PAGE,
1269 messages,
1270 })?;
1271 Ok(())
1272 }
1273
1274 async fn leave_channel(
1275 self: Arc<Self>,
1276 request: TypedEnvelope<proto::LeaveChannel>,
1277 ) -> Result<()> {
1278 let user_id = self
1279 .store()
1280 .await
1281 .user_id_for_connection(request.sender_id)?;
1282 let channel_id = ChannelId::from_proto(request.payload.channel_id);
1283 if !self
1284 .app_state
1285 .db
1286 .can_user_access_channel(user_id, channel_id)
1287 .await?
1288 {
1289 Err(anyhow!("access denied"))?;
1290 }
1291
1292 self.store_mut()
1293 .await
1294 .leave_channel(request.sender_id, channel_id);
1295
1296 Ok(())
1297 }
1298
1299 async fn send_channel_message(
1300 self: Arc<Self>,
1301 request: TypedEnvelope<proto::SendChannelMessage>,
1302 response: Response<proto::SendChannelMessage>,
1303 ) -> Result<()> {
1304 let channel_id = ChannelId::from_proto(request.payload.channel_id);
1305 let user_id;
1306 let connection_ids;
1307 {
1308 let state = self.store().await;
1309 user_id = state.user_id_for_connection(request.sender_id)?;
1310 connection_ids = state.channel_connection_ids(channel_id)?;
1311 }
1312
1313 // Validate the message body.
1314 let body = request.payload.body.trim().to_string();
1315 if body.len() > MAX_MESSAGE_LEN {
1316 return Err(anyhow!("message is too long"))?;
1317 }
1318 if body.is_empty() {
1319 return Err(anyhow!("message can't be blank"))?;
1320 }
1321
1322 let timestamp = OffsetDateTime::now_utc();
1323 let nonce = request
1324 .payload
1325 .nonce
1326 .ok_or_else(|| anyhow!("nonce can't be blank"))?;
1327
1328 let message_id = self
1329 .app_state
1330 .db
1331 .create_channel_message(channel_id, user_id, &body, timestamp, nonce.clone().into())
1332 .await?
1333 .to_proto();
1334 let message = proto::ChannelMessage {
1335 sender_id: user_id.to_proto(),
1336 id: message_id,
1337 body,
1338 timestamp: timestamp.unix_timestamp() as u64,
1339 nonce: Some(nonce),
1340 };
1341 broadcast(request.sender_id, connection_ids, |conn_id| {
1342 self.peer.send(
1343 conn_id,
1344 proto::ChannelMessageSent {
1345 channel_id: channel_id.to_proto(),
1346 message: Some(message.clone()),
1347 },
1348 )
1349 });
1350 response.send(proto::SendChannelMessageResponse {
1351 message: Some(message),
1352 })?;
1353 Ok(())
1354 }
1355
1356 async fn get_channel_messages(
1357 self: Arc<Self>,
1358 request: TypedEnvelope<proto::GetChannelMessages>,
1359 response: Response<proto::GetChannelMessages>,
1360 ) -> Result<()> {
1361 let user_id = self
1362 .store()
1363 .await
1364 .user_id_for_connection(request.sender_id)?;
1365 let channel_id = ChannelId::from_proto(request.payload.channel_id);
1366 if !self
1367 .app_state
1368 .db
1369 .can_user_access_channel(user_id, channel_id)
1370 .await?
1371 {
1372 Err(anyhow!("access denied"))?;
1373 }
1374
1375 let messages = self
1376 .app_state
1377 .db
1378 .get_channel_messages(
1379 channel_id,
1380 MESSAGE_COUNT_PER_PAGE,
1381 Some(MessageId::from_proto(request.payload.before_message_id)),
1382 )
1383 .await?
1384 .into_iter()
1385 .map(|msg| proto::ChannelMessage {
1386 id: msg.id.to_proto(),
1387 body: msg.body,
1388 timestamp: msg.sent_at.unix_timestamp() as u64,
1389 sender_id: msg.sender_id.to_proto(),
1390 nonce: Some(msg.nonce.as_u128().into()),
1391 })
1392 .collect::<Vec<_>>();
1393 response.send(proto::GetChannelMessagesResponse {
1394 done: messages.len() < MESSAGE_COUNT_PER_PAGE,
1395 messages,
1396 })?;
1397 Ok(())
1398 }
1399
1400 async fn store<'a>(self: &'a Arc<Self>) -> StoreReadGuard<'a> {
1401 #[cfg(test)]
1402 tokio::task::yield_now().await;
1403 let guard = self.store.read().await;
1404 #[cfg(test)]
1405 tokio::task::yield_now().await;
1406 StoreReadGuard {
1407 guard,
1408 _not_send: PhantomData,
1409 }
1410 }
1411
1412 async fn store_mut<'a>(self: &'a Arc<Self>) -> StoreWriteGuard<'a> {
1413 #[cfg(test)]
1414 tokio::task::yield_now().await;
1415 let guard = self.store.write().await;
1416 #[cfg(test)]
1417 tokio::task::yield_now().await;
1418 StoreWriteGuard {
1419 guard,
1420 _not_send: PhantomData,
1421 }
1422 }
1423}
1424
1425impl<'a> Deref for StoreReadGuard<'a> {
1426 type Target = Store;
1427
1428 fn deref(&self) -> &Self::Target {
1429 &*self.guard
1430 }
1431}
1432
1433impl<'a> Deref for StoreWriteGuard<'a> {
1434 type Target = Store;
1435
1436 fn deref(&self) -> &Self::Target {
1437 &*self.guard
1438 }
1439}
1440
1441impl<'a> DerefMut for StoreWriteGuard<'a> {
1442 fn deref_mut(&mut self) -> &mut Self::Target {
1443 &mut *self.guard
1444 }
1445}
1446
1447impl<'a> Drop for StoreWriteGuard<'a> {
1448 fn drop(&mut self) {
1449 #[cfg(test)]
1450 self.check_invariants();
1451
1452 let metrics = self.metrics();
1453 tracing::info!(
1454 connections = metrics.connections,
1455 registered_projects = metrics.registered_projects,
1456 shared_projects = metrics.shared_projects,
1457 "metrics"
1458 );
1459 }
1460}
1461
1462impl Executor for RealExecutor {
1463 type Sleep = Sleep;
1464
1465 fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F) {
1466 tokio::task::spawn(future);
1467 }
1468
1469 fn sleep(&self, duration: Duration) -> Self::Sleep {
1470 tokio::time::sleep(duration)
1471 }
1472}
1473
1474fn broadcast<F>(
1475 sender_id: ConnectionId,
1476 receiver_ids: impl IntoIterator<Item = ConnectionId>,
1477 mut f: F,
1478) where
1479 F: FnMut(ConnectionId) -> anyhow::Result<()>,
1480{
1481 for receiver_id in receiver_ids {
1482 if receiver_id != sender_id {
1483 f(receiver_id).trace_err();
1484 }
1485 }
1486}
1487
1488lazy_static! {
1489 static ref ZED_PROTOCOL_VERSION: HeaderName = HeaderName::from_static("x-zed-protocol-version");
1490}
1491
1492pub struct ProtocolVersion(u32);
1493
1494impl Header for ProtocolVersion {
1495 fn name() -> &'static HeaderName {
1496 &ZED_PROTOCOL_VERSION
1497 }
1498
1499 fn decode<'i, I>(values: &mut I) -> Result<Self, axum::headers::Error>
1500 where
1501 Self: Sized,
1502 I: Iterator<Item = &'i axum::http::HeaderValue>,
1503 {
1504 let version = values
1505 .next()
1506 .ok_or_else(|| axum::headers::Error::invalid())?
1507 .to_str()
1508 .map_err(|_| axum::headers::Error::invalid())?
1509 .parse()
1510 .map_err(|_| axum::headers::Error::invalid())?;
1511 Ok(Self(version))
1512 }
1513
1514 fn encode<E: Extend<axum::http::HeaderValue>>(&self, values: &mut E) {
1515 values.extend([self.0.to_string().parse().unwrap()]);
1516 }
1517}
1518
1519pub fn routes(app_state: Arc<AppState>) -> Router<Body> {
1520 let server = Server::new(app_state.clone(), None);
1521 Router::new()
1522 .route("/rpc", get(handle_websocket_request))
1523 .layer(
1524 ServiceBuilder::new()
1525 .layer(Extension(app_state))
1526 .layer(middleware::from_fn(auth::validate_header))
1527 .layer(Extension(server)),
1528 )
1529}
1530
1531pub async fn handle_websocket_request(
1532 TypedHeader(ProtocolVersion(protocol_version)): TypedHeader<ProtocolVersion>,
1533 ConnectInfo(socket_address): ConnectInfo<SocketAddr>,
1534 Extension(server): Extension<Arc<Server>>,
1535 Extension(user): Extension<User>,
1536 ws: WebSocketUpgrade,
1537) -> axum::response::Response {
1538 if protocol_version != rpc::PROTOCOL_VERSION {
1539 return (
1540 StatusCode::UPGRADE_REQUIRED,
1541 "client must be upgraded".to_string(),
1542 )
1543 .into_response();
1544 }
1545 let socket_address = socket_address.to_string();
1546 ws.on_upgrade(move |socket| {
1547 use util::ResultExt;
1548 let socket = socket
1549 .map_ok(to_tungstenite_message)
1550 .err_into()
1551 .with(|message| async move { Ok(to_axum_message(message)) });
1552 let connection = Connection::new(Box::pin(socket));
1553 async move {
1554 server
1555 .handle_connection(connection, socket_address, user, None, RealExecutor)
1556 .await
1557 .log_err();
1558 }
1559 })
1560}
1561
1562fn to_axum_message(message: TungsteniteMessage) -> AxumMessage {
1563 match message {
1564 TungsteniteMessage::Text(payload) => AxumMessage::Text(payload),
1565 TungsteniteMessage::Binary(payload) => AxumMessage::Binary(payload),
1566 TungsteniteMessage::Ping(payload) => AxumMessage::Ping(payload),
1567 TungsteniteMessage::Pong(payload) => AxumMessage::Pong(payload),
1568 TungsteniteMessage::Close(frame) => AxumMessage::Close(frame.map(|frame| AxumCloseFrame {
1569 code: frame.code.into(),
1570 reason: frame.reason,
1571 })),
1572 }
1573}
1574
1575fn to_tungstenite_message(message: AxumMessage) -> TungsteniteMessage {
1576 match message {
1577 AxumMessage::Text(payload) => TungsteniteMessage::Text(payload),
1578 AxumMessage::Binary(payload) => TungsteniteMessage::Binary(payload),
1579 AxumMessage::Ping(payload) => TungsteniteMessage::Ping(payload),
1580 AxumMessage::Pong(payload) => TungsteniteMessage::Pong(payload),
1581 AxumMessage::Close(frame) => {
1582 TungsteniteMessage::Close(frame.map(|frame| TungsteniteCloseFrame {
1583 code: frame.code.into(),
1584 reason: frame.reason,
1585 }))
1586 }
1587 }
1588}
1589
1590pub trait ResultExt {
1591 type Ok;
1592
1593 fn trace_err(self) -> Option<Self::Ok>;
1594}
1595
1596impl<T, E> ResultExt for Result<T, E>
1597where
1598 E: std::fmt::Debug,
1599{
1600 type Ok = T;
1601
1602 fn trace_err(self) -> Option<T> {
1603 match self {
1604 Ok(value) => Some(value),
1605 Err(error) => {
1606 tracing::error!("{:?}", error);
1607 None
1608 }
1609 }
1610 }
1611}
1612
1613#[cfg(test)]
1614mod tests {
1615 use super::*;
1616 use crate::{
1617 db::{tests::TestDb, UserId},
1618 AppState,
1619 };
1620 use ::rpc::Peer;
1621 use client::{
1622 self, test::FakeHttpClient, Channel, ChannelDetails, ChannelList, Client, Credentials,
1623 EstablishConnectionError, UserStore, RECEIVE_TIMEOUT,
1624 };
1625 use collections::{BTreeMap, HashSet};
1626 use editor::{
1627 self, ConfirmCodeAction, ConfirmCompletion, ConfirmRename, Editor, Input, Redo, Rename,
1628 ToOffset, ToggleCodeActions, Undo,
1629 };
1630 use gpui::{
1631 executor::{self, Deterministic},
1632 geometry::vector::vec2f,
1633 ModelHandle, Task, TestAppContext, ViewHandle,
1634 };
1635 use language::{
1636 range_to_lsp, tree_sitter_rust, Diagnostic, DiagnosticEntry, FakeLspAdapter, Language,
1637 LanguageConfig, LanguageRegistry, OffsetRangeExt, Point, Rope,
1638 };
1639 use lsp::{self, FakeLanguageServer};
1640 use parking_lot::Mutex;
1641 use project::{
1642 fs::{FakeFs, Fs as _},
1643 search::SearchQuery,
1644 worktree::WorktreeHandle,
1645 DiagnosticSummary, Project, ProjectPath, WorktreeId,
1646 };
1647 use rand::prelude::*;
1648 use rpc::PeerId;
1649 use serde_json::json;
1650 use settings::Settings;
1651 use sqlx::types::time::OffsetDateTime;
1652 use std::{
1653 cell::RefCell,
1654 env,
1655 ops::Deref,
1656 path::{Path, PathBuf},
1657 rc::Rc,
1658 sync::{
1659 atomic::{AtomicBool, Ordering::SeqCst},
1660 Arc,
1661 },
1662 time::Duration,
1663 };
1664 use theme::ThemeRegistry;
1665 use workspace::{Item, SplitDirection, ToggleFollow, Workspace};
1666
1667 #[cfg(test)]
1668 #[ctor::ctor]
1669 fn init_logger() {
1670 if std::env::var("RUST_LOG").is_ok() {
1671 env_logger::init();
1672 }
1673 }
1674
1675 #[gpui::test(iterations = 10)]
1676 async fn test_share_project(
1677 deterministic: Arc<Deterministic>,
1678 cx_a: &mut TestAppContext,
1679 cx_b: &mut TestAppContext,
1680 cx_b2: &mut TestAppContext,
1681 ) {
1682 let (window_b, _) = cx_b.add_window(|_| EmptyView);
1683 let lang_registry = Arc::new(LanguageRegistry::test());
1684 let fs = FakeFs::new(cx_a.background());
1685 cx_a.foreground().forbid_parking();
1686
1687 // Connect to a server as 2 clients.
1688 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1689 let client_a = server.create_client(cx_a, "user_a").await;
1690 let mut client_b = server.create_client(cx_b, "user_b").await;
1691 server
1692 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
1693 .await;
1694
1695 // Share a project as client A
1696 fs.insert_tree(
1697 "/a",
1698 json!({
1699 ".gitignore": "ignored-dir",
1700 "a.txt": "a-contents",
1701 "b.txt": "b-contents",
1702 "ignored-dir": {
1703 "c.txt": "",
1704 "d.txt": "",
1705 }
1706 }),
1707 )
1708 .await;
1709 let project_a = cx_a.update(|cx| {
1710 Project::local(
1711 client_a.clone(),
1712 client_a.user_store.clone(),
1713 lang_registry.clone(),
1714 fs.clone(),
1715 cx,
1716 )
1717 });
1718 let project_id = project_a
1719 .read_with(cx_a, |project, _| project.next_remote_id())
1720 .await;
1721 let (worktree_a, _) = project_a
1722 .update(cx_a, |p, cx| {
1723 p.find_or_create_local_worktree("/a", true, cx)
1724 })
1725 .await
1726 .unwrap();
1727 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1728 worktree_a
1729 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1730 .await;
1731
1732 // Join that project as client B
1733 let client_b_peer_id = client_b.peer_id;
1734 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
1735 let replica_id_b = project_b.read_with(cx_b, |project, _| {
1736 assert_eq!(
1737 project
1738 .collaborators()
1739 .get(&client_a.peer_id)
1740 .unwrap()
1741 .user
1742 .github_login,
1743 "user_a"
1744 );
1745 project.replica_id()
1746 });
1747
1748 deterministic.run_until_parked();
1749 project_a.read_with(cx_a, |project, _| {
1750 let client_b_collaborator = project.collaborators().get(&client_b_peer_id).unwrap();
1751 assert_eq!(client_b_collaborator.replica_id, replica_id_b);
1752 assert_eq!(client_b_collaborator.user.github_login, "user_b");
1753 });
1754 project_b.read_with(cx_b, |project, cx| {
1755 let worktree = project.worktrees(cx).next().unwrap().read(cx);
1756 assert_eq!(
1757 worktree.paths().map(AsRef::as_ref).collect::<Vec<_>>(),
1758 [
1759 Path::new(".gitignore"),
1760 Path::new("a.txt"),
1761 Path::new("b.txt"),
1762 Path::new("ignored-dir"),
1763 Path::new("ignored-dir/c.txt"),
1764 Path::new("ignored-dir/d.txt"),
1765 ]
1766 );
1767 });
1768
1769 // Open the same file as client B and client A.
1770 let buffer_b = project_b
1771 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "b.txt"), cx))
1772 .await
1773 .unwrap();
1774 buffer_b.read_with(cx_b, |buf, _| assert_eq!(buf.text(), "b-contents"));
1775 project_a.read_with(cx_a, |project, cx| {
1776 assert!(project.has_open_buffer((worktree_id, "b.txt"), cx))
1777 });
1778 let buffer_a = project_a
1779 .update(cx_a, |p, cx| p.open_buffer((worktree_id, "b.txt"), cx))
1780 .await
1781 .unwrap();
1782
1783 let editor_b = cx_b.add_view(window_b, |cx| Editor::for_buffer(buffer_b, None, cx));
1784
1785 // TODO
1786 // // Create a selection set as client B and see that selection set as client A.
1787 // buffer_a
1788 // .condition(&cx_a, |buffer, _| buffer.selection_sets().count() == 1)
1789 // .await;
1790
1791 // Edit the buffer as client B and see that edit as client A.
1792 editor_b.update(cx_b, |editor, cx| {
1793 editor.handle_input(&Input("ok, ".into()), cx)
1794 });
1795 buffer_a
1796 .condition(&cx_a, |buffer, _| buffer.text() == "ok, b-contents")
1797 .await;
1798
1799 // TODO
1800 // // Remove the selection set as client B, see those selections disappear as client A.
1801 cx_b.update(move |_| drop(editor_b));
1802 // buffer_a
1803 // .condition(&cx_a, |buffer, _| buffer.selection_sets().count() == 0)
1804 // .await;
1805
1806 // Client B can join again on a different window because they are already a participant.
1807 let client_b2 = server.create_client(cx_b2, "user_b").await;
1808 let project_b2 = Project::remote(
1809 project_id,
1810 client_b2.client.clone(),
1811 client_b2.user_store.clone(),
1812 lang_registry.clone(),
1813 FakeFs::new(cx_b2.background()),
1814 &mut cx_b2.to_async(),
1815 )
1816 .await
1817 .unwrap();
1818 deterministic.run_until_parked();
1819 project_a.read_with(cx_a, |project, _| {
1820 assert_eq!(project.collaborators().len(), 2);
1821 });
1822 project_b.read_with(cx_b, |project, _| {
1823 assert_eq!(project.collaborators().len(), 2);
1824 });
1825 project_b2.read_with(cx_b2, |project, _| {
1826 assert_eq!(project.collaborators().len(), 2);
1827 });
1828
1829 // Dropping client B's first project removes only that from client A's collaborators.
1830 cx_b.update(move |_| {
1831 drop(client_b.project.take());
1832 drop(project_b);
1833 });
1834 deterministic.run_until_parked();
1835 project_a.read_with(cx_a, |project, _| {
1836 assert_eq!(project.collaborators().len(), 1);
1837 });
1838 project_b2.read_with(cx_b2, |project, _| {
1839 assert_eq!(project.collaborators().len(), 1);
1840 });
1841 }
1842
1843 #[gpui::test(iterations = 10)]
1844 async fn test_unshare_project(
1845 deterministic: Arc<Deterministic>,
1846 cx_a: &mut TestAppContext,
1847 cx_b: &mut TestAppContext,
1848 ) {
1849 let lang_registry = Arc::new(LanguageRegistry::test());
1850 let fs = FakeFs::new(cx_a.background());
1851 cx_a.foreground().forbid_parking();
1852
1853 // Connect to a server as 2 clients.
1854 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1855 let client_a = server.create_client(cx_a, "user_a").await;
1856 let mut client_b = server.create_client(cx_b, "user_b").await;
1857 server
1858 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
1859 .await;
1860
1861 // Share a project as client A
1862 fs.insert_tree(
1863 "/a",
1864 json!({
1865 "a.txt": "a-contents",
1866 "b.txt": "b-contents",
1867 }),
1868 )
1869 .await;
1870 let project_a = cx_a.update(|cx| {
1871 Project::local(
1872 client_a.clone(),
1873 client_a.user_store.clone(),
1874 lang_registry.clone(),
1875 fs.clone(),
1876 cx,
1877 )
1878 });
1879 let (worktree_a, _) = project_a
1880 .update(cx_a, |p, cx| {
1881 p.find_or_create_local_worktree("/a", true, cx)
1882 })
1883 .await
1884 .unwrap();
1885 worktree_a
1886 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1887 .await;
1888 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1889
1890 // Join that project as client B
1891 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
1892 assert!(worktree_a.read_with(cx_a, |tree, _| tree.as_local().unwrap().is_shared()));
1893 project_b
1894 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1895 .await
1896 .unwrap();
1897
1898 // When client B leaves the project, it gets automatically unshared.
1899 cx_b.update(|_| {
1900 drop(client_b.project.take());
1901 drop(project_b);
1902 });
1903 deterministic.run_until_parked();
1904 assert!(worktree_a.read_with(cx_a, |tree, _| !tree.as_local().unwrap().is_shared()));
1905
1906 // When client B joins again, the project gets re-shared.
1907 let project_b2 = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
1908 assert!(worktree_a.read_with(cx_a, |tree, _| tree.as_local().unwrap().is_shared()));
1909 project_b2
1910 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1911 .await
1912 .unwrap();
1913 }
1914
1915 #[gpui::test(iterations = 10)]
1916 async fn test_host_disconnect(
1917 deterministic: Arc<Deterministic>,
1918 cx_a: &mut TestAppContext,
1919 cx_b: &mut TestAppContext,
1920 cx_c: &mut TestAppContext,
1921 ) {
1922 let lang_registry = Arc::new(LanguageRegistry::test());
1923 let fs = FakeFs::new(cx_a.background());
1924 cx_a.foreground().forbid_parking();
1925
1926 // Connect to a server as 3 clients.
1927 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1928 let client_a = server.create_client(cx_a, "user_a").await;
1929 let mut client_b = server.create_client(cx_b, "user_b").await;
1930 let client_c = server.create_client(cx_c, "user_c").await;
1931 server
1932 .make_contacts(vec![
1933 (&client_a, cx_a),
1934 (&client_b, cx_b),
1935 (&client_c, cx_c),
1936 ])
1937 .await;
1938
1939 // Share a project as client A
1940 fs.insert_tree(
1941 "/a",
1942 json!({
1943 "a.txt": "a-contents",
1944 "b.txt": "b-contents",
1945 }),
1946 )
1947 .await;
1948 let project_a = cx_a.update(|cx| {
1949 Project::local(
1950 client_a.clone(),
1951 client_a.user_store.clone(),
1952 lang_registry.clone(),
1953 fs.clone(),
1954 cx,
1955 )
1956 });
1957 let project_id = project_a
1958 .read_with(cx_a, |project, _| project.next_remote_id())
1959 .await;
1960 let (worktree_a, _) = project_a
1961 .update(cx_a, |p, cx| {
1962 p.find_or_create_local_worktree("/a", true, cx)
1963 })
1964 .await
1965 .unwrap();
1966 worktree_a
1967 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1968 .await;
1969 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1970
1971 // Join that project as client B
1972 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
1973 assert!(worktree_a.read_with(cx_a, |tree, _| tree.as_local().unwrap().is_shared()));
1974 project_b
1975 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1976 .await
1977 .unwrap();
1978
1979 // Request to join that project as client C
1980 let project_c = cx_c.spawn(|mut cx| {
1981 let client = client_c.client.clone();
1982 let user_store = client_c.user_store.clone();
1983 let lang_registry = lang_registry.clone();
1984 async move {
1985 Project::remote(
1986 project_id,
1987 client,
1988 user_store,
1989 lang_registry.clone(),
1990 FakeFs::new(cx.background()),
1991 &mut cx,
1992 )
1993 .await
1994 }
1995 });
1996 deterministic.run_until_parked();
1997
1998 // Drop client A's connection. Collaborators should disappear and the project should not be shown as shared.
1999 server.disconnect_client(client_a.current_user_id(cx_a));
2000 cx_a.foreground().advance_clock(rpc::RECEIVE_TIMEOUT);
2001 project_a
2002 .condition(cx_a, |project, _| project.collaborators().is_empty())
2003 .await;
2004 project_a.read_with(cx_a, |project, _| assert!(!project.is_shared()));
2005 project_b
2006 .condition(cx_b, |project, _| project.is_read_only())
2007 .await;
2008 assert!(worktree_a.read_with(cx_a, |tree, _| !tree.as_local().unwrap().is_shared()));
2009 cx_b.update(|_| {
2010 drop(project_b);
2011 });
2012 assert!(matches!(
2013 project_c.await.unwrap_err(),
2014 project::JoinProjectError::HostWentOffline
2015 ));
2016
2017 // Ensure guests can still join.
2018 let project_b2 = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2019 assert!(worktree_a.read_with(cx_a, |tree, _| tree.as_local().unwrap().is_shared()));
2020 project_b2
2021 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
2022 .await
2023 .unwrap();
2024 }
2025
2026 #[gpui::test(iterations = 10)]
2027 async fn test_decline_join_request(
2028 deterministic: Arc<Deterministic>,
2029 cx_a: &mut TestAppContext,
2030 cx_b: &mut TestAppContext,
2031 ) {
2032 let lang_registry = Arc::new(LanguageRegistry::test());
2033 let fs = FakeFs::new(cx_a.background());
2034 cx_a.foreground().forbid_parking();
2035
2036 // Connect to a server as 2 clients.
2037 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2038 let client_a = server.create_client(cx_a, "user_a").await;
2039 let client_b = server.create_client(cx_b, "user_b").await;
2040 server
2041 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2042 .await;
2043
2044 // Share a project as client A
2045 fs.insert_tree("/a", json!({})).await;
2046 let project_a = cx_a.update(|cx| {
2047 Project::local(
2048 client_a.clone(),
2049 client_a.user_store.clone(),
2050 lang_registry.clone(),
2051 fs.clone(),
2052 cx,
2053 )
2054 });
2055 let project_id = project_a
2056 .read_with(cx_a, |project, _| project.next_remote_id())
2057 .await;
2058 let (worktree_a, _) = project_a
2059 .update(cx_a, |p, cx| {
2060 p.find_or_create_local_worktree("/a", true, cx)
2061 })
2062 .await
2063 .unwrap();
2064 worktree_a
2065 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2066 .await;
2067
2068 // Request to join that project as client B
2069 let project_b = cx_b.spawn(|mut cx| {
2070 let client = client_b.client.clone();
2071 let user_store = client_b.user_store.clone();
2072 let lang_registry = lang_registry.clone();
2073 async move {
2074 Project::remote(
2075 project_id,
2076 client,
2077 user_store,
2078 lang_registry.clone(),
2079 FakeFs::new(cx.background()),
2080 &mut cx,
2081 )
2082 .await
2083 }
2084 });
2085 deterministic.run_until_parked();
2086 project_a.update(cx_a, |project, cx| {
2087 project.respond_to_join_request(client_b.user_id().unwrap(), false, cx)
2088 });
2089 assert!(matches!(
2090 project_b.await.unwrap_err(),
2091 project::JoinProjectError::HostDeclined
2092 ));
2093
2094 // Request to join the project again as client B
2095 let project_b = cx_b.spawn(|mut cx| {
2096 let client = client_b.client.clone();
2097 let user_store = client_b.user_store.clone();
2098 let lang_registry = lang_registry.clone();
2099 async move {
2100 Project::remote(
2101 project_id,
2102 client,
2103 user_store,
2104 lang_registry.clone(),
2105 FakeFs::new(cx.background()),
2106 &mut cx,
2107 )
2108 .await
2109 }
2110 });
2111
2112 // Close the project on the host
2113 deterministic.run_until_parked();
2114 cx_a.update(|_| drop(project_a));
2115 deterministic.run_until_parked();
2116 assert!(matches!(
2117 project_b.await.unwrap_err(),
2118 project::JoinProjectError::HostClosedProject
2119 ));
2120 }
2121
2122 #[gpui::test(iterations = 10)]
2123 async fn test_cancel_join_request(
2124 deterministic: Arc<Deterministic>,
2125 cx_a: &mut TestAppContext,
2126 cx_b: &mut TestAppContext,
2127 ) {
2128 let lang_registry = Arc::new(LanguageRegistry::test());
2129 let fs = FakeFs::new(cx_a.background());
2130 cx_a.foreground().forbid_parking();
2131
2132 // Connect to a server as 2 clients.
2133 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2134 let client_a = server.create_client(cx_a, "user_a").await;
2135 let client_b = server.create_client(cx_b, "user_b").await;
2136 server
2137 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2138 .await;
2139
2140 // Share a project as client A
2141 fs.insert_tree("/a", json!({})).await;
2142 let project_a = cx_a.update(|cx| {
2143 Project::local(
2144 client_a.clone(),
2145 client_a.user_store.clone(),
2146 lang_registry.clone(),
2147 fs.clone(),
2148 cx,
2149 )
2150 });
2151 let project_id = project_a
2152 .read_with(cx_a, |project, _| project.next_remote_id())
2153 .await;
2154
2155 let project_a_events = Rc::new(RefCell::new(Vec::new()));
2156 let user_b = client_a
2157 .user_store
2158 .update(cx_a, |store, cx| {
2159 store.fetch_user(client_b.user_id().unwrap(), cx)
2160 })
2161 .await
2162 .unwrap();
2163 project_a.update(cx_a, {
2164 let project_a_events = project_a_events.clone();
2165 move |_, cx| {
2166 cx.subscribe(&cx.handle(), move |_, _, event, _| {
2167 project_a_events.borrow_mut().push(event.clone());
2168 })
2169 .detach();
2170 }
2171 });
2172
2173 let (worktree_a, _) = project_a
2174 .update(cx_a, |p, cx| {
2175 p.find_or_create_local_worktree("/a", true, cx)
2176 })
2177 .await
2178 .unwrap();
2179 worktree_a
2180 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2181 .await;
2182
2183 // Request to join that project as client B
2184 let project_b = cx_b.spawn(|mut cx| {
2185 let client = client_b.client.clone();
2186 let user_store = client_b.user_store.clone();
2187 let lang_registry = lang_registry.clone();
2188 async move {
2189 Project::remote(
2190 project_id,
2191 client,
2192 user_store,
2193 lang_registry.clone(),
2194 FakeFs::new(cx.background()),
2195 &mut cx,
2196 )
2197 .await
2198 }
2199 });
2200 deterministic.run_until_parked();
2201 assert_eq!(
2202 &*project_a_events.borrow(),
2203 &[project::Event::ContactRequestedJoin(user_b.clone())]
2204 );
2205 project_a_events.borrow_mut().clear();
2206
2207 // Cancel the join request by leaving the project
2208 client_b
2209 .client
2210 .send(proto::LeaveProject { project_id })
2211 .unwrap();
2212 drop(project_b);
2213
2214 deterministic.run_until_parked();
2215 assert_eq!(
2216 &*project_a_events.borrow(),
2217 &[project::Event::ContactCancelledJoinRequest(user_b.clone())]
2218 );
2219 }
2220
2221 #[gpui::test(iterations = 10)]
2222 async fn test_propagate_saves_and_fs_changes(
2223 cx_a: &mut TestAppContext,
2224 cx_b: &mut TestAppContext,
2225 cx_c: &mut TestAppContext,
2226 ) {
2227 let lang_registry = Arc::new(LanguageRegistry::test());
2228 let fs = FakeFs::new(cx_a.background());
2229 cx_a.foreground().forbid_parking();
2230
2231 // Connect to a server as 3 clients.
2232 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2233 let client_a = server.create_client(cx_a, "user_a").await;
2234 let mut client_b = server.create_client(cx_b, "user_b").await;
2235 let mut client_c = server.create_client(cx_c, "user_c").await;
2236 server
2237 .make_contacts(vec![
2238 (&client_a, cx_a),
2239 (&client_b, cx_b),
2240 (&client_c, cx_c),
2241 ])
2242 .await;
2243
2244 // Share a worktree as client A.
2245 fs.insert_tree(
2246 "/a",
2247 json!({
2248 "file1": "",
2249 "file2": ""
2250 }),
2251 )
2252 .await;
2253 let project_a = cx_a.update(|cx| {
2254 Project::local(
2255 client_a.clone(),
2256 client_a.user_store.clone(),
2257 lang_registry.clone(),
2258 fs.clone(),
2259 cx,
2260 )
2261 });
2262 let (worktree_a, _) = project_a
2263 .update(cx_a, |p, cx| {
2264 p.find_or_create_local_worktree("/a", true, cx)
2265 })
2266 .await
2267 .unwrap();
2268 worktree_a
2269 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2270 .await;
2271 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2272
2273 // Join that worktree as clients B and C.
2274 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2275 let project_c = client_c.build_remote_project(&project_a, cx_a, cx_c).await;
2276 let worktree_b = project_b.read_with(cx_b, |p, cx| p.worktrees(cx).next().unwrap());
2277 let worktree_c = project_c.read_with(cx_c, |p, cx| p.worktrees(cx).next().unwrap());
2278
2279 // Open and edit a buffer as both guests B and C.
2280 let buffer_b = project_b
2281 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "file1"), cx))
2282 .await
2283 .unwrap();
2284 let buffer_c = project_c
2285 .update(cx_c, |p, cx| p.open_buffer((worktree_id, "file1"), cx))
2286 .await
2287 .unwrap();
2288 buffer_b.update(cx_b, |buf, cx| buf.edit([(0..0, "i-am-b, ")], cx));
2289 buffer_c.update(cx_c, |buf, cx| buf.edit([(0..0, "i-am-c, ")], cx));
2290
2291 // Open and edit that buffer as the host.
2292 let buffer_a = project_a
2293 .update(cx_a, |p, cx| p.open_buffer((worktree_id, "file1"), cx))
2294 .await
2295 .unwrap();
2296
2297 buffer_a
2298 .condition(cx_a, |buf, _| buf.text() == "i-am-c, i-am-b, ")
2299 .await;
2300 buffer_a.update(cx_a, |buf, cx| {
2301 buf.edit([(buf.len()..buf.len(), "i-am-a")], cx)
2302 });
2303
2304 // Wait for edits to propagate
2305 buffer_a
2306 .condition(cx_a, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
2307 .await;
2308 buffer_b
2309 .condition(cx_b, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
2310 .await;
2311 buffer_c
2312 .condition(cx_c, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
2313 .await;
2314
2315 // Edit the buffer as the host and concurrently save as guest B.
2316 let save_b = buffer_b.update(cx_b, |buf, cx| buf.save(cx));
2317 buffer_a.update(cx_a, |buf, cx| buf.edit([(0..0, "hi-a, ")], cx));
2318 save_b.await.unwrap();
2319 assert_eq!(
2320 fs.load("/a/file1".as_ref()).await.unwrap(),
2321 "hi-a, i-am-c, i-am-b, i-am-a"
2322 );
2323 buffer_a.read_with(cx_a, |buf, _| assert!(!buf.is_dirty()));
2324 buffer_b.read_with(cx_b, |buf, _| assert!(!buf.is_dirty()));
2325 buffer_c.condition(cx_c, |buf, _| !buf.is_dirty()).await;
2326
2327 worktree_a.flush_fs_events(cx_a).await;
2328
2329 // Make changes on host's file system, see those changes on guest worktrees.
2330 fs.rename(
2331 "/a/file1".as_ref(),
2332 "/a/file1-renamed".as_ref(),
2333 Default::default(),
2334 )
2335 .await
2336 .unwrap();
2337
2338 fs.rename("/a/file2".as_ref(), "/a/file3".as_ref(), Default::default())
2339 .await
2340 .unwrap();
2341 fs.insert_file(Path::new("/a/file4"), "4".into()).await;
2342
2343 worktree_a
2344 .condition(&cx_a, |tree, _| {
2345 tree.paths()
2346 .map(|p| p.to_string_lossy())
2347 .collect::<Vec<_>>()
2348 == ["file1-renamed", "file3", "file4"]
2349 })
2350 .await;
2351 worktree_b
2352 .condition(&cx_b, |tree, _| {
2353 tree.paths()
2354 .map(|p| p.to_string_lossy())
2355 .collect::<Vec<_>>()
2356 == ["file1-renamed", "file3", "file4"]
2357 })
2358 .await;
2359 worktree_c
2360 .condition(&cx_c, |tree, _| {
2361 tree.paths()
2362 .map(|p| p.to_string_lossy())
2363 .collect::<Vec<_>>()
2364 == ["file1-renamed", "file3", "file4"]
2365 })
2366 .await;
2367
2368 // Ensure buffer files are updated as well.
2369 buffer_a
2370 .condition(&cx_a, |buf, _| {
2371 buf.file().unwrap().path().to_str() == Some("file1-renamed")
2372 })
2373 .await;
2374 buffer_b
2375 .condition(&cx_b, |buf, _| {
2376 buf.file().unwrap().path().to_str() == Some("file1-renamed")
2377 })
2378 .await;
2379 buffer_c
2380 .condition(&cx_c, |buf, _| {
2381 buf.file().unwrap().path().to_str() == Some("file1-renamed")
2382 })
2383 .await;
2384 }
2385
2386 #[gpui::test(iterations = 10)]
2387 async fn test_fs_operations(
2388 executor: Arc<Deterministic>,
2389 cx_a: &mut TestAppContext,
2390 cx_b: &mut TestAppContext,
2391 ) {
2392 executor.forbid_parking();
2393 let fs = FakeFs::new(cx_a.background());
2394
2395 // Connect to a server as 2 clients.
2396 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2397 let mut client_a = server.create_client(cx_a, "user_a").await;
2398 let mut client_b = server.create_client(cx_b, "user_b").await;
2399 server
2400 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2401 .await;
2402
2403 // Share a project as client A
2404 fs.insert_tree(
2405 "/dir",
2406 json!({
2407 "a.txt": "a-contents",
2408 "b.txt": "b-contents",
2409 }),
2410 )
2411 .await;
2412
2413 let (project_a, worktree_id) = client_a.build_local_project(fs, "/dir", cx_a).await;
2414 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2415
2416 let worktree_a =
2417 project_a.read_with(cx_a, |project, cx| project.worktrees(cx).next().unwrap());
2418 let worktree_b =
2419 project_b.read_with(cx_b, |project, cx| project.worktrees(cx).next().unwrap());
2420
2421 let entry = project_b
2422 .update(cx_b, |project, cx| {
2423 project
2424 .create_entry((worktree_id, "c.txt"), false, cx)
2425 .unwrap()
2426 })
2427 .await
2428 .unwrap();
2429 worktree_a.read_with(cx_a, |worktree, _| {
2430 assert_eq!(
2431 worktree
2432 .paths()
2433 .map(|p| p.to_string_lossy())
2434 .collect::<Vec<_>>(),
2435 ["a.txt", "b.txt", "c.txt"]
2436 );
2437 });
2438 worktree_b.read_with(cx_b, |worktree, _| {
2439 assert_eq!(
2440 worktree
2441 .paths()
2442 .map(|p| p.to_string_lossy())
2443 .collect::<Vec<_>>(),
2444 ["a.txt", "b.txt", "c.txt"]
2445 );
2446 });
2447
2448 project_b
2449 .update(cx_b, |project, cx| {
2450 project.rename_entry(entry.id, Path::new("d.txt"), cx)
2451 })
2452 .unwrap()
2453 .await
2454 .unwrap();
2455 worktree_a.read_with(cx_a, |worktree, _| {
2456 assert_eq!(
2457 worktree
2458 .paths()
2459 .map(|p| p.to_string_lossy())
2460 .collect::<Vec<_>>(),
2461 ["a.txt", "b.txt", "d.txt"]
2462 );
2463 });
2464 worktree_b.read_with(cx_b, |worktree, _| {
2465 assert_eq!(
2466 worktree
2467 .paths()
2468 .map(|p| p.to_string_lossy())
2469 .collect::<Vec<_>>(),
2470 ["a.txt", "b.txt", "d.txt"]
2471 );
2472 });
2473
2474 let dir_entry = project_b
2475 .update(cx_b, |project, cx| {
2476 project
2477 .create_entry((worktree_id, "DIR"), true, cx)
2478 .unwrap()
2479 })
2480 .await
2481 .unwrap();
2482 worktree_a.read_with(cx_a, |worktree, _| {
2483 assert_eq!(
2484 worktree
2485 .paths()
2486 .map(|p| p.to_string_lossy())
2487 .collect::<Vec<_>>(),
2488 ["DIR", "a.txt", "b.txt", "d.txt"]
2489 );
2490 });
2491 worktree_b.read_with(cx_b, |worktree, _| {
2492 assert_eq!(
2493 worktree
2494 .paths()
2495 .map(|p| p.to_string_lossy())
2496 .collect::<Vec<_>>(),
2497 ["DIR", "a.txt", "b.txt", "d.txt"]
2498 );
2499 });
2500
2501 project_b
2502 .update(cx_b, |project, cx| {
2503 project.delete_entry(dir_entry.id, cx).unwrap()
2504 })
2505 .await
2506 .unwrap();
2507 worktree_a.read_with(cx_a, |worktree, _| {
2508 assert_eq!(
2509 worktree
2510 .paths()
2511 .map(|p| p.to_string_lossy())
2512 .collect::<Vec<_>>(),
2513 ["a.txt", "b.txt", "d.txt"]
2514 );
2515 });
2516 worktree_b.read_with(cx_b, |worktree, _| {
2517 assert_eq!(
2518 worktree
2519 .paths()
2520 .map(|p| p.to_string_lossy())
2521 .collect::<Vec<_>>(),
2522 ["a.txt", "b.txt", "d.txt"]
2523 );
2524 });
2525
2526 project_b
2527 .update(cx_b, |project, cx| {
2528 project.delete_entry(entry.id, cx).unwrap()
2529 })
2530 .await
2531 .unwrap();
2532 worktree_a.read_with(cx_a, |worktree, _| {
2533 assert_eq!(
2534 worktree
2535 .paths()
2536 .map(|p| p.to_string_lossy())
2537 .collect::<Vec<_>>(),
2538 ["a.txt", "b.txt"]
2539 );
2540 });
2541 worktree_b.read_with(cx_b, |worktree, _| {
2542 assert_eq!(
2543 worktree
2544 .paths()
2545 .map(|p| p.to_string_lossy())
2546 .collect::<Vec<_>>(),
2547 ["a.txt", "b.txt"]
2548 );
2549 });
2550 }
2551
2552 #[gpui::test(iterations = 10)]
2553 async fn test_buffer_conflict_after_save(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
2554 cx_a.foreground().forbid_parking();
2555 let lang_registry = Arc::new(LanguageRegistry::test());
2556 let fs = FakeFs::new(cx_a.background());
2557
2558 // Connect to a server as 2 clients.
2559 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2560 let client_a = server.create_client(cx_a, "user_a").await;
2561 let mut client_b = server.create_client(cx_b, "user_b").await;
2562 server
2563 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2564 .await;
2565
2566 // Share a project as client A
2567 fs.insert_tree(
2568 "/dir",
2569 json!({
2570 "a.txt": "a-contents",
2571 }),
2572 )
2573 .await;
2574
2575 let project_a = cx_a.update(|cx| {
2576 Project::local(
2577 client_a.clone(),
2578 client_a.user_store.clone(),
2579 lang_registry.clone(),
2580 fs.clone(),
2581 cx,
2582 )
2583 });
2584 let (worktree_a, _) = project_a
2585 .update(cx_a, |p, cx| {
2586 p.find_or_create_local_worktree("/dir", true, cx)
2587 })
2588 .await
2589 .unwrap();
2590 worktree_a
2591 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2592 .await;
2593 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2594
2595 // Join that project as client B
2596 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2597
2598 // Open a buffer as client B
2599 let buffer_b = project_b
2600 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
2601 .await
2602 .unwrap();
2603
2604 buffer_b.update(cx_b, |buf, cx| buf.edit([(0..0, "world ")], cx));
2605 buffer_b.read_with(cx_b, |buf, _| {
2606 assert!(buf.is_dirty());
2607 assert!(!buf.has_conflict());
2608 });
2609
2610 buffer_b.update(cx_b, |buf, cx| buf.save(cx)).await.unwrap();
2611 buffer_b
2612 .condition(&cx_b, |buffer_b, _| !buffer_b.is_dirty())
2613 .await;
2614 buffer_b.read_with(cx_b, |buf, _| {
2615 assert!(!buf.has_conflict());
2616 });
2617
2618 buffer_b.update(cx_b, |buf, cx| buf.edit([(0..0, "hello ")], cx));
2619 buffer_b.read_with(cx_b, |buf, _| {
2620 assert!(buf.is_dirty());
2621 assert!(!buf.has_conflict());
2622 });
2623 }
2624
2625 #[gpui::test(iterations = 10)]
2626 async fn test_buffer_reloading(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
2627 cx_a.foreground().forbid_parking();
2628 let lang_registry = Arc::new(LanguageRegistry::test());
2629 let fs = FakeFs::new(cx_a.background());
2630
2631 // Connect to a server as 2 clients.
2632 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2633 let client_a = server.create_client(cx_a, "user_a").await;
2634 let mut client_b = server.create_client(cx_b, "user_b").await;
2635 server
2636 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2637 .await;
2638
2639 // Share a project as client A
2640 fs.insert_tree(
2641 "/dir",
2642 json!({
2643 "a.txt": "a-contents",
2644 }),
2645 )
2646 .await;
2647
2648 let project_a = cx_a.update(|cx| {
2649 Project::local(
2650 client_a.clone(),
2651 client_a.user_store.clone(),
2652 lang_registry.clone(),
2653 fs.clone(),
2654 cx,
2655 )
2656 });
2657 let (worktree_a, _) = project_a
2658 .update(cx_a, |p, cx| {
2659 p.find_or_create_local_worktree("/dir", true, cx)
2660 })
2661 .await
2662 .unwrap();
2663 worktree_a
2664 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2665 .await;
2666 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2667
2668 // Join that project as client B
2669 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2670 let _worktree_b = project_b.update(cx_b, |p, cx| p.worktrees(cx).next().unwrap());
2671
2672 // Open a buffer as client B
2673 let buffer_b = project_b
2674 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
2675 .await
2676 .unwrap();
2677 buffer_b.read_with(cx_b, |buf, _| {
2678 assert!(!buf.is_dirty());
2679 assert!(!buf.has_conflict());
2680 });
2681
2682 fs.save(Path::new("/dir/a.txt"), &"new contents".into())
2683 .await
2684 .unwrap();
2685 buffer_b
2686 .condition(&cx_b, |buf, _| {
2687 buf.text() == "new contents" && !buf.is_dirty()
2688 })
2689 .await;
2690 buffer_b.read_with(cx_b, |buf, _| {
2691 assert!(!buf.has_conflict());
2692 });
2693 }
2694
2695 #[gpui::test(iterations = 10)]
2696 async fn test_editing_while_guest_opens_buffer(
2697 cx_a: &mut TestAppContext,
2698 cx_b: &mut TestAppContext,
2699 ) {
2700 cx_a.foreground().forbid_parking();
2701 let lang_registry = Arc::new(LanguageRegistry::test());
2702 let fs = FakeFs::new(cx_a.background());
2703
2704 // Connect to a server as 2 clients.
2705 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2706 let client_a = server.create_client(cx_a, "user_a").await;
2707 let mut client_b = server.create_client(cx_b, "user_b").await;
2708 server
2709 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2710 .await;
2711
2712 // Share a project as client A
2713 fs.insert_tree(
2714 "/dir",
2715 json!({
2716 "a.txt": "a-contents",
2717 }),
2718 )
2719 .await;
2720 let project_a = cx_a.update(|cx| {
2721 Project::local(
2722 client_a.clone(),
2723 client_a.user_store.clone(),
2724 lang_registry.clone(),
2725 fs.clone(),
2726 cx,
2727 )
2728 });
2729 let (worktree_a, _) = project_a
2730 .update(cx_a, |p, cx| {
2731 p.find_or_create_local_worktree("/dir", true, cx)
2732 })
2733 .await
2734 .unwrap();
2735 worktree_a
2736 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2737 .await;
2738 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2739
2740 // Join that project as client B
2741 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2742
2743 // Open a buffer as client A
2744 let buffer_a = project_a
2745 .update(cx_a, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
2746 .await
2747 .unwrap();
2748
2749 // Start opening the same buffer as client B
2750 let buffer_b = cx_b
2751 .background()
2752 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx)));
2753
2754 // Edit the buffer as client A while client B is still opening it.
2755 cx_b.background().simulate_random_delay().await;
2756 buffer_a.update(cx_a, |buf, cx| buf.edit([(0..0, "X")], cx));
2757 cx_b.background().simulate_random_delay().await;
2758 buffer_a.update(cx_a, |buf, cx| buf.edit([(1..1, "Y")], cx));
2759
2760 let text = buffer_a.read_with(cx_a, |buf, _| buf.text());
2761 let buffer_b = buffer_b.await.unwrap();
2762 buffer_b.condition(&cx_b, |buf, _| buf.text() == text).await;
2763 }
2764
2765 #[gpui::test(iterations = 10)]
2766 async fn test_leaving_worktree_while_opening_buffer(
2767 cx_a: &mut TestAppContext,
2768 cx_b: &mut TestAppContext,
2769 ) {
2770 cx_a.foreground().forbid_parking();
2771 let lang_registry = Arc::new(LanguageRegistry::test());
2772 let fs = FakeFs::new(cx_a.background());
2773
2774 // Connect to a server as 2 clients.
2775 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2776 let client_a = server.create_client(cx_a, "user_a").await;
2777 let mut client_b = server.create_client(cx_b, "user_b").await;
2778 server
2779 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2780 .await;
2781
2782 // Share a project as client A
2783 fs.insert_tree(
2784 "/dir",
2785 json!({
2786 "a.txt": "a-contents",
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(cx_a, |p, cx| {
2801 p.find_or_create_local_worktree("/dir", true, 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 worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2809
2810 // Join that project as client B
2811 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2812
2813 // See that a guest has joined as client A.
2814 project_a
2815 .condition(&cx_a, |p, _| p.collaborators().len() == 1)
2816 .await;
2817
2818 // Begin opening a buffer as client B, but leave the project before the open completes.
2819 let buffer_b = cx_b
2820 .background()
2821 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx)));
2822 cx_b.update(|_| {
2823 drop(client_b.project.take());
2824 drop(project_b);
2825 });
2826 drop(buffer_b);
2827
2828 // See that the guest has left.
2829 project_a
2830 .condition(&cx_a, |p, _| p.collaborators().len() == 0)
2831 .await;
2832 }
2833
2834 #[gpui::test(iterations = 10)]
2835 async fn test_leaving_project(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
2836 cx_a.foreground().forbid_parking();
2837 let lang_registry = Arc::new(LanguageRegistry::test());
2838 let fs = FakeFs::new(cx_a.background());
2839
2840 // Connect to a server as 2 clients.
2841 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2842 let client_a = server.create_client(cx_a, "user_a").await;
2843 let mut client_b = server.create_client(cx_b, "user_b").await;
2844 server
2845 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2846 .await;
2847
2848 // Share a project as client A
2849 fs.insert_tree(
2850 "/a",
2851 json!({
2852 "a.txt": "a-contents",
2853 "b.txt": "b-contents",
2854 }),
2855 )
2856 .await;
2857 let project_a = cx_a.update(|cx| {
2858 Project::local(
2859 client_a.clone(),
2860 client_a.user_store.clone(),
2861 lang_registry.clone(),
2862 fs.clone(),
2863 cx,
2864 )
2865 });
2866 let (worktree_a, _) = project_a
2867 .update(cx_a, |p, cx| {
2868 p.find_or_create_local_worktree("/a", true, cx)
2869 })
2870 .await
2871 .unwrap();
2872 worktree_a
2873 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2874 .await;
2875
2876 // Join that project as client B
2877 let _project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2878
2879 // Client A sees that a guest has joined.
2880 project_a
2881 .condition(cx_a, |p, _| p.collaborators().len() == 1)
2882 .await;
2883
2884 // Drop client B's connection and ensure client A observes client B leaving the project.
2885 client_b.disconnect(&cx_b.to_async()).unwrap();
2886 project_a
2887 .condition(cx_a, |p, _| p.collaborators().len() == 0)
2888 .await;
2889
2890 // Rejoin the project as client B
2891 let _project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2892
2893 // Client A sees that a guest has re-joined.
2894 project_a
2895 .condition(cx_a, |p, _| p.collaborators().len() == 1)
2896 .await;
2897
2898 // Simulate connection loss for client B and ensure client A observes client B leaving the project.
2899 client_b.wait_for_current_user(cx_b).await;
2900 server.disconnect_client(client_b.current_user_id(cx_b));
2901 cx_a.foreground().advance_clock(rpc::RECEIVE_TIMEOUT);
2902 project_a
2903 .condition(cx_a, |p, _| p.collaborators().len() == 0)
2904 .await;
2905 }
2906
2907 #[gpui::test(iterations = 10)]
2908 async fn test_collaborating_with_diagnostics(
2909 deterministic: Arc<Deterministic>,
2910 cx_a: &mut TestAppContext,
2911 cx_b: &mut TestAppContext,
2912 cx_c: &mut TestAppContext,
2913 ) {
2914 deterministic.forbid_parking();
2915 let lang_registry = Arc::new(LanguageRegistry::test());
2916 let fs = FakeFs::new(cx_a.background());
2917
2918 // Set up a fake language server.
2919 let mut language = Language::new(
2920 LanguageConfig {
2921 name: "Rust".into(),
2922 path_suffixes: vec!["rs".to_string()],
2923 ..Default::default()
2924 },
2925 Some(tree_sitter_rust::language()),
2926 );
2927 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
2928 lang_registry.add(Arc::new(language));
2929
2930 // Connect to a server as 2 clients.
2931 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2932 let client_a = server.create_client(cx_a, "user_a").await;
2933 let mut client_b = server.create_client(cx_b, "user_b").await;
2934 let mut client_c = server.create_client(cx_c, "user_c").await;
2935 server
2936 .make_contacts(vec![
2937 (&client_a, cx_a),
2938 (&client_b, cx_b),
2939 (&client_c, cx_c),
2940 ])
2941 .await;
2942
2943 // Share a project as client A
2944 fs.insert_tree(
2945 "/a",
2946 json!({
2947 "a.rs": "let one = two",
2948 "other.rs": "",
2949 }),
2950 )
2951 .await;
2952 let project_a = cx_a.update(|cx| {
2953 Project::local(
2954 client_a.clone(),
2955 client_a.user_store.clone(),
2956 lang_registry.clone(),
2957 fs.clone(),
2958 cx,
2959 )
2960 });
2961 let (worktree_a, _) = project_a
2962 .update(cx_a, |p, cx| {
2963 p.find_or_create_local_worktree("/a", true, cx)
2964 })
2965 .await
2966 .unwrap();
2967 worktree_a
2968 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2969 .await;
2970 let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
2971 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2972
2973 // Cause the language server to start.
2974 let _buffer = cx_a
2975 .background()
2976 .spawn(project_a.update(cx_a, |project, cx| {
2977 project.open_buffer(
2978 ProjectPath {
2979 worktree_id,
2980 path: Path::new("other.rs").into(),
2981 },
2982 cx,
2983 )
2984 }))
2985 .await
2986 .unwrap();
2987
2988 // Join the worktree as client B.
2989 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2990
2991 // Simulate a language server reporting errors for a file.
2992 let mut fake_language_server = fake_language_servers.next().await.unwrap();
2993 fake_language_server
2994 .receive_notification::<lsp::notification::DidOpenTextDocument>()
2995 .await;
2996 fake_language_server.notify::<lsp::notification::PublishDiagnostics>(
2997 lsp::PublishDiagnosticsParams {
2998 uri: lsp::Url::from_file_path("/a/a.rs").unwrap(),
2999 version: None,
3000 diagnostics: vec![lsp::Diagnostic {
3001 severity: Some(lsp::DiagnosticSeverity::ERROR),
3002 range: lsp::Range::new(lsp::Position::new(0, 4), lsp::Position::new(0, 7)),
3003 message: "message 1".to_string(),
3004 ..Default::default()
3005 }],
3006 },
3007 );
3008
3009 // Wait for server to see the diagnostics update.
3010 deterministic.run_until_parked();
3011 {
3012 let store = server.store.read().await;
3013 let project = store.project(project_id).unwrap();
3014 let worktree = project.worktrees.get(&worktree_id.to_proto()).unwrap();
3015 assert!(!worktree.diagnostic_summaries.is_empty());
3016 }
3017
3018 // Ensure client B observes the new diagnostics.
3019 project_b.read_with(cx_b, |project, cx| {
3020 assert_eq!(
3021 project.diagnostic_summaries(cx).collect::<Vec<_>>(),
3022 &[(
3023 ProjectPath {
3024 worktree_id,
3025 path: Arc::from(Path::new("a.rs")),
3026 },
3027 DiagnosticSummary {
3028 error_count: 1,
3029 warning_count: 0,
3030 ..Default::default()
3031 },
3032 )]
3033 )
3034 });
3035
3036 // Join project as client C and observe the diagnostics.
3037 let project_c = client_c.build_remote_project(&project_a, cx_a, cx_c).await;
3038 project_c.read_with(cx_c, |project, cx| {
3039 assert_eq!(
3040 project.diagnostic_summaries(cx).collect::<Vec<_>>(),
3041 &[(
3042 ProjectPath {
3043 worktree_id,
3044 path: Arc::from(Path::new("a.rs")),
3045 },
3046 DiagnosticSummary {
3047 error_count: 1,
3048 warning_count: 0,
3049 ..Default::default()
3050 },
3051 )]
3052 )
3053 });
3054
3055 // Simulate a language server reporting more errors for a file.
3056 fake_language_server.notify::<lsp::notification::PublishDiagnostics>(
3057 lsp::PublishDiagnosticsParams {
3058 uri: lsp::Url::from_file_path("/a/a.rs").unwrap(),
3059 version: None,
3060 diagnostics: vec![
3061 lsp::Diagnostic {
3062 severity: Some(lsp::DiagnosticSeverity::ERROR),
3063 range: lsp::Range::new(lsp::Position::new(0, 4), lsp::Position::new(0, 7)),
3064 message: "message 1".to_string(),
3065 ..Default::default()
3066 },
3067 lsp::Diagnostic {
3068 severity: Some(lsp::DiagnosticSeverity::WARNING),
3069 range: lsp::Range::new(
3070 lsp::Position::new(0, 10),
3071 lsp::Position::new(0, 13),
3072 ),
3073 message: "message 2".to_string(),
3074 ..Default::default()
3075 },
3076 ],
3077 },
3078 );
3079
3080 // Clients B and C get the updated summaries
3081 deterministic.run_until_parked();
3082 project_b.read_with(cx_b, |project, cx| {
3083 assert_eq!(
3084 project.diagnostic_summaries(cx).collect::<Vec<_>>(),
3085 [(
3086 ProjectPath {
3087 worktree_id,
3088 path: Arc::from(Path::new("a.rs")),
3089 },
3090 DiagnosticSummary {
3091 error_count: 1,
3092 warning_count: 1,
3093 ..Default::default()
3094 },
3095 )]
3096 );
3097 });
3098 project_c.read_with(cx_c, |project, cx| {
3099 assert_eq!(
3100 project.diagnostic_summaries(cx).collect::<Vec<_>>(),
3101 [(
3102 ProjectPath {
3103 worktree_id,
3104 path: Arc::from(Path::new("a.rs")),
3105 },
3106 DiagnosticSummary {
3107 error_count: 1,
3108 warning_count: 1,
3109 ..Default::default()
3110 },
3111 )]
3112 );
3113 });
3114
3115 // Open the file with the errors on client B. They should be present.
3116 let buffer_b = cx_b
3117 .background()
3118 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
3119 .await
3120 .unwrap();
3121
3122 buffer_b.read_with(cx_b, |buffer, _| {
3123 assert_eq!(
3124 buffer
3125 .snapshot()
3126 .diagnostics_in_range::<_, Point>(0..buffer.len(), false)
3127 .map(|entry| entry)
3128 .collect::<Vec<_>>(),
3129 &[
3130 DiagnosticEntry {
3131 range: Point::new(0, 4)..Point::new(0, 7),
3132 diagnostic: Diagnostic {
3133 group_id: 0,
3134 message: "message 1".to_string(),
3135 severity: lsp::DiagnosticSeverity::ERROR,
3136 is_primary: true,
3137 ..Default::default()
3138 }
3139 },
3140 DiagnosticEntry {
3141 range: Point::new(0, 10)..Point::new(0, 13),
3142 diagnostic: Diagnostic {
3143 group_id: 1,
3144 severity: lsp::DiagnosticSeverity::WARNING,
3145 message: "message 2".to_string(),
3146 is_primary: true,
3147 ..Default::default()
3148 }
3149 }
3150 ]
3151 );
3152 });
3153
3154 // Simulate a language server reporting no errors for a file.
3155 fake_language_server.notify::<lsp::notification::PublishDiagnostics>(
3156 lsp::PublishDiagnosticsParams {
3157 uri: lsp::Url::from_file_path("/a/a.rs").unwrap(),
3158 version: None,
3159 diagnostics: vec![],
3160 },
3161 );
3162 deterministic.run_until_parked();
3163 project_a.read_with(cx_a, |project, cx| {
3164 assert_eq!(project.diagnostic_summaries(cx).collect::<Vec<_>>(), [])
3165 });
3166 project_b.read_with(cx_b, |project, cx| {
3167 assert_eq!(project.diagnostic_summaries(cx).collect::<Vec<_>>(), [])
3168 });
3169 project_c.read_with(cx_c, |project, cx| {
3170 assert_eq!(project.diagnostic_summaries(cx).collect::<Vec<_>>(), [])
3171 });
3172 }
3173
3174 #[gpui::test(iterations = 10)]
3175 async fn test_collaborating_with_completion(
3176 cx_a: &mut TestAppContext,
3177 cx_b: &mut TestAppContext,
3178 ) {
3179 cx_a.foreground().forbid_parking();
3180 let lang_registry = Arc::new(LanguageRegistry::test());
3181 let fs = FakeFs::new(cx_a.background());
3182
3183 // Set up a fake language server.
3184 let mut language = Language::new(
3185 LanguageConfig {
3186 name: "Rust".into(),
3187 path_suffixes: vec!["rs".to_string()],
3188 ..Default::default()
3189 },
3190 Some(tree_sitter_rust::language()),
3191 );
3192 let mut fake_language_servers = language.set_fake_lsp_adapter(FakeLspAdapter {
3193 capabilities: lsp::ServerCapabilities {
3194 completion_provider: Some(lsp::CompletionOptions {
3195 trigger_characters: Some(vec![".".to_string()]),
3196 ..Default::default()
3197 }),
3198 ..Default::default()
3199 },
3200 ..Default::default()
3201 });
3202 lang_registry.add(Arc::new(language));
3203
3204 // Connect to a server as 2 clients.
3205 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3206 let client_a = server.create_client(cx_a, "user_a").await;
3207 let mut client_b = server.create_client(cx_b, "user_b").await;
3208 server
3209 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3210 .await;
3211
3212 // Share a project as client A
3213 fs.insert_tree(
3214 "/a",
3215 json!({
3216 "main.rs": "fn main() { a }",
3217 "other.rs": "",
3218 }),
3219 )
3220 .await;
3221 let project_a = cx_a.update(|cx| {
3222 Project::local(
3223 client_a.clone(),
3224 client_a.user_store.clone(),
3225 lang_registry.clone(),
3226 fs.clone(),
3227 cx,
3228 )
3229 });
3230 let (worktree_a, _) = project_a
3231 .update(cx_a, |p, cx| {
3232 p.find_or_create_local_worktree("/a", true, cx)
3233 })
3234 .await
3235 .unwrap();
3236 worktree_a
3237 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3238 .await;
3239 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3240
3241 // Join the worktree as client B.
3242 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3243
3244 // Open a file in an editor as the guest.
3245 let buffer_b = project_b
3246 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "main.rs"), cx))
3247 .await
3248 .unwrap();
3249 let (window_b, _) = cx_b.add_window(|_| EmptyView);
3250 let editor_b = cx_b.add_view(window_b, |cx| {
3251 Editor::for_buffer(buffer_b.clone(), Some(project_b.clone()), cx)
3252 });
3253
3254 let fake_language_server = fake_language_servers.next().await.unwrap();
3255 buffer_b
3256 .condition(&cx_b, |buffer, _| !buffer.completion_triggers().is_empty())
3257 .await;
3258
3259 // Type a completion trigger character as the guest.
3260 editor_b.update(cx_b, |editor, cx| {
3261 editor.change_selections(None, cx, |s| s.select_ranges([13..13]));
3262 editor.handle_input(&Input(".".into()), cx);
3263 cx.focus(&editor_b);
3264 });
3265
3266 // Receive a completion request as the host's language server.
3267 // Return some completions from the host's language server.
3268 cx_a.foreground().start_waiting();
3269 fake_language_server
3270 .handle_request::<lsp::request::Completion, _, _>(|params, _| async move {
3271 assert_eq!(
3272 params.text_document_position.text_document.uri,
3273 lsp::Url::from_file_path("/a/main.rs").unwrap(),
3274 );
3275 assert_eq!(
3276 params.text_document_position.position,
3277 lsp::Position::new(0, 14),
3278 );
3279
3280 Ok(Some(lsp::CompletionResponse::Array(vec![
3281 lsp::CompletionItem {
3282 label: "first_method(…)".into(),
3283 detail: Some("fn(&mut self, B) -> C".into()),
3284 text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
3285 new_text: "first_method($1)".to_string(),
3286 range: lsp::Range::new(
3287 lsp::Position::new(0, 14),
3288 lsp::Position::new(0, 14),
3289 ),
3290 })),
3291 insert_text_format: Some(lsp::InsertTextFormat::SNIPPET),
3292 ..Default::default()
3293 },
3294 lsp::CompletionItem {
3295 label: "second_method(…)".into(),
3296 detail: Some("fn(&mut self, C) -> D<E>".into()),
3297 text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
3298 new_text: "second_method()".to_string(),
3299 range: lsp::Range::new(
3300 lsp::Position::new(0, 14),
3301 lsp::Position::new(0, 14),
3302 ),
3303 })),
3304 insert_text_format: Some(lsp::InsertTextFormat::SNIPPET),
3305 ..Default::default()
3306 },
3307 ])))
3308 })
3309 .next()
3310 .await
3311 .unwrap();
3312 cx_a.foreground().finish_waiting();
3313
3314 // Open the buffer on the host.
3315 let buffer_a = project_a
3316 .update(cx_a, |p, cx| p.open_buffer((worktree_id, "main.rs"), cx))
3317 .await
3318 .unwrap();
3319 buffer_a
3320 .condition(&cx_a, |buffer, _| buffer.text() == "fn main() { a. }")
3321 .await;
3322
3323 // Confirm a completion on the guest.
3324 editor_b
3325 .condition(&cx_b, |editor, _| editor.context_menu_visible())
3326 .await;
3327 editor_b.update(cx_b, |editor, cx| {
3328 editor.confirm_completion(&ConfirmCompletion { item_ix: Some(0) }, cx);
3329 assert_eq!(editor.text(cx), "fn main() { a.first_method() }");
3330 });
3331
3332 // Return a resolved completion from the host's language server.
3333 // The resolved completion has an additional text edit.
3334 fake_language_server.handle_request::<lsp::request::ResolveCompletionItem, _, _>(
3335 |params, _| async move {
3336 assert_eq!(params.label, "first_method(…)");
3337 Ok(lsp::CompletionItem {
3338 label: "first_method(…)".into(),
3339 detail: Some("fn(&mut self, B) -> C".into()),
3340 text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
3341 new_text: "first_method($1)".to_string(),
3342 range: lsp::Range::new(
3343 lsp::Position::new(0, 14),
3344 lsp::Position::new(0, 14),
3345 ),
3346 })),
3347 additional_text_edits: Some(vec![lsp::TextEdit {
3348 new_text: "use d::SomeTrait;\n".to_string(),
3349 range: lsp::Range::new(lsp::Position::new(0, 0), lsp::Position::new(0, 0)),
3350 }]),
3351 insert_text_format: Some(lsp::InsertTextFormat::SNIPPET),
3352 ..Default::default()
3353 })
3354 },
3355 );
3356
3357 // The additional edit is applied.
3358 buffer_a
3359 .condition(&cx_a, |buffer, _| {
3360 buffer.text() == "use d::SomeTrait;\nfn main() { a.first_method() }"
3361 })
3362 .await;
3363 buffer_b
3364 .condition(&cx_b, |buffer, _| {
3365 buffer.text() == "use d::SomeTrait;\nfn main() { a.first_method() }"
3366 })
3367 .await;
3368 }
3369
3370 #[gpui::test(iterations = 10)]
3371 async fn test_reloading_buffer_manually(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3372 cx_a.foreground().forbid_parking();
3373 let lang_registry = Arc::new(LanguageRegistry::test());
3374 let fs = FakeFs::new(cx_a.background());
3375
3376 // Connect to a server as 2 clients.
3377 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3378 let client_a = server.create_client(cx_a, "user_a").await;
3379 let mut client_b = server.create_client(cx_b, "user_b").await;
3380 server
3381 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3382 .await;
3383
3384 // Share a project as client A
3385 fs.insert_tree(
3386 "/a",
3387 json!({
3388 "a.rs": "let one = 1;",
3389 }),
3390 )
3391 .await;
3392 let project_a = cx_a.update(|cx| {
3393 Project::local(
3394 client_a.clone(),
3395 client_a.user_store.clone(),
3396 lang_registry.clone(),
3397 fs.clone(),
3398 cx,
3399 )
3400 });
3401 let (worktree_a, _) = project_a
3402 .update(cx_a, |p, cx| {
3403 p.find_or_create_local_worktree("/a", true, cx)
3404 })
3405 .await
3406 .unwrap();
3407 worktree_a
3408 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3409 .await;
3410 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3411 let buffer_a = project_a
3412 .update(cx_a, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx))
3413 .await
3414 .unwrap();
3415
3416 // Join the worktree as client B.
3417 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3418
3419 let buffer_b = cx_b
3420 .background()
3421 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
3422 .await
3423 .unwrap();
3424 buffer_b.update(cx_b, |buffer, cx| {
3425 buffer.edit([(4..7, "six")], cx);
3426 buffer.edit([(10..11, "6")], cx);
3427 assert_eq!(buffer.text(), "let six = 6;");
3428 assert!(buffer.is_dirty());
3429 assert!(!buffer.has_conflict());
3430 });
3431 buffer_a
3432 .condition(cx_a, |buffer, _| buffer.text() == "let six = 6;")
3433 .await;
3434
3435 fs.save(Path::new("/a/a.rs"), &Rope::from("let seven = 7;"))
3436 .await
3437 .unwrap();
3438 buffer_a
3439 .condition(cx_a, |buffer, _| buffer.has_conflict())
3440 .await;
3441 buffer_b
3442 .condition(cx_b, |buffer, _| buffer.has_conflict())
3443 .await;
3444
3445 project_b
3446 .update(cx_b, |project, cx| {
3447 project.reload_buffers(HashSet::from_iter([buffer_b.clone()]), true, cx)
3448 })
3449 .await
3450 .unwrap();
3451 buffer_a.read_with(cx_a, |buffer, _| {
3452 assert_eq!(buffer.text(), "let seven = 7;");
3453 assert!(!buffer.is_dirty());
3454 assert!(!buffer.has_conflict());
3455 });
3456 buffer_b.read_with(cx_b, |buffer, _| {
3457 assert_eq!(buffer.text(), "let seven = 7;");
3458 assert!(!buffer.is_dirty());
3459 assert!(!buffer.has_conflict());
3460 });
3461
3462 buffer_a.update(cx_a, |buffer, cx| {
3463 // Undoing on the host is a no-op when the reload was initiated by the guest.
3464 buffer.undo(cx);
3465 assert_eq!(buffer.text(), "let seven = 7;");
3466 assert!(!buffer.is_dirty());
3467 assert!(!buffer.has_conflict());
3468 });
3469 buffer_b.update(cx_b, |buffer, cx| {
3470 // Undoing on the guest rolls back the buffer to before it was reloaded but the conflict gets cleared.
3471 buffer.undo(cx);
3472 assert_eq!(buffer.text(), "let six = 6;");
3473 assert!(buffer.is_dirty());
3474 assert!(!buffer.has_conflict());
3475 });
3476 }
3477
3478 #[gpui::test(iterations = 10)]
3479 async fn test_formatting_buffer(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3480 cx_a.foreground().forbid_parking();
3481 let lang_registry = Arc::new(LanguageRegistry::test());
3482 let fs = FakeFs::new(cx_a.background());
3483
3484 // Set up a fake language server.
3485 let mut language = Language::new(
3486 LanguageConfig {
3487 name: "Rust".into(),
3488 path_suffixes: vec!["rs".to_string()],
3489 ..Default::default()
3490 },
3491 Some(tree_sitter_rust::language()),
3492 );
3493 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3494 lang_registry.add(Arc::new(language));
3495
3496 // Connect to a server as 2 clients.
3497 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3498 let client_a = server.create_client(cx_a, "user_a").await;
3499 let mut client_b = server.create_client(cx_b, "user_b").await;
3500 server
3501 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3502 .await;
3503
3504 // Share a project as client A
3505 fs.insert_tree(
3506 "/a",
3507 json!({
3508 "a.rs": "let one = two",
3509 }),
3510 )
3511 .await;
3512 let project_a = cx_a.update(|cx| {
3513 Project::local(
3514 client_a.clone(),
3515 client_a.user_store.clone(),
3516 lang_registry.clone(),
3517 fs.clone(),
3518 cx,
3519 )
3520 });
3521 let (worktree_a, _) = project_a
3522 .update(cx_a, |p, cx| {
3523 p.find_or_create_local_worktree("/a", true, cx)
3524 })
3525 .await
3526 .unwrap();
3527 worktree_a
3528 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3529 .await;
3530 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3531
3532 // Join the project as client B.
3533 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3534
3535 let buffer_b = cx_b
3536 .background()
3537 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
3538 .await
3539 .unwrap();
3540
3541 let fake_language_server = fake_language_servers.next().await.unwrap();
3542 fake_language_server.handle_request::<lsp::request::Formatting, _, _>(|_, _| async move {
3543 Ok(Some(vec![
3544 lsp::TextEdit {
3545 range: lsp::Range::new(lsp::Position::new(0, 4), lsp::Position::new(0, 4)),
3546 new_text: "h".to_string(),
3547 },
3548 lsp::TextEdit {
3549 range: lsp::Range::new(lsp::Position::new(0, 7), lsp::Position::new(0, 7)),
3550 new_text: "y".to_string(),
3551 },
3552 ]))
3553 });
3554
3555 project_b
3556 .update(cx_b, |project, cx| {
3557 project.format(HashSet::from_iter([buffer_b.clone()]), true, cx)
3558 })
3559 .await
3560 .unwrap();
3561 assert_eq!(
3562 buffer_b.read_with(cx_b, |buffer, _| buffer.text()),
3563 "let honey = two"
3564 );
3565 }
3566
3567 #[gpui::test(iterations = 10)]
3568 async fn test_definition(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3569 cx_a.foreground().forbid_parking();
3570 let lang_registry = Arc::new(LanguageRegistry::test());
3571 let fs = FakeFs::new(cx_a.background());
3572 fs.insert_tree(
3573 "/root-1",
3574 json!({
3575 "a.rs": "const ONE: usize = b::TWO + b::THREE;",
3576 }),
3577 )
3578 .await;
3579 fs.insert_tree(
3580 "/root-2",
3581 json!({
3582 "b.rs": "const TWO: usize = 2;\nconst THREE: usize = 3;",
3583 }),
3584 )
3585 .await;
3586
3587 // Set up a fake language server.
3588 let mut language = Language::new(
3589 LanguageConfig {
3590 name: "Rust".into(),
3591 path_suffixes: vec!["rs".to_string()],
3592 ..Default::default()
3593 },
3594 Some(tree_sitter_rust::language()),
3595 );
3596 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3597 lang_registry.add(Arc::new(language));
3598
3599 // Connect to a server as 2 clients.
3600 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3601 let client_a = server.create_client(cx_a, "user_a").await;
3602 let mut client_b = server.create_client(cx_b, "user_b").await;
3603 server
3604 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3605 .await;
3606
3607 // Share a project as client A
3608 let project_a = cx_a.update(|cx| {
3609 Project::local(
3610 client_a.clone(),
3611 client_a.user_store.clone(),
3612 lang_registry.clone(),
3613 fs.clone(),
3614 cx,
3615 )
3616 });
3617 let (worktree_a, _) = project_a
3618 .update(cx_a, |p, cx| {
3619 p.find_or_create_local_worktree("/root-1", true, cx)
3620 })
3621 .await
3622 .unwrap();
3623 worktree_a
3624 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3625 .await;
3626 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3627
3628 // Join the project as client B.
3629 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3630
3631 // Open the file on client B.
3632 let buffer_b = cx_b
3633 .background()
3634 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
3635 .await
3636 .unwrap();
3637
3638 // Request the definition of a symbol as the guest.
3639 let fake_language_server = fake_language_servers.next().await.unwrap();
3640 fake_language_server.handle_request::<lsp::request::GotoDefinition, _, _>(
3641 |_, _| async move {
3642 Ok(Some(lsp::GotoDefinitionResponse::Scalar(
3643 lsp::Location::new(
3644 lsp::Url::from_file_path("/root-2/b.rs").unwrap(),
3645 lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
3646 ),
3647 )))
3648 },
3649 );
3650
3651 let definitions_1 = project_b
3652 .update(cx_b, |p, cx| p.definition(&buffer_b, 23, cx))
3653 .await
3654 .unwrap();
3655 cx_b.read(|cx| {
3656 assert_eq!(definitions_1.len(), 1);
3657 assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
3658 let target_buffer = definitions_1[0].buffer.read(cx);
3659 assert_eq!(
3660 target_buffer.text(),
3661 "const TWO: usize = 2;\nconst THREE: usize = 3;"
3662 );
3663 assert_eq!(
3664 definitions_1[0].range.to_point(target_buffer),
3665 Point::new(0, 6)..Point::new(0, 9)
3666 );
3667 });
3668
3669 // Try getting more definitions for the same buffer, ensuring the buffer gets reused from
3670 // the previous call to `definition`.
3671 fake_language_server.handle_request::<lsp::request::GotoDefinition, _, _>(
3672 |_, _| async move {
3673 Ok(Some(lsp::GotoDefinitionResponse::Scalar(
3674 lsp::Location::new(
3675 lsp::Url::from_file_path("/root-2/b.rs").unwrap(),
3676 lsp::Range::new(lsp::Position::new(1, 6), lsp::Position::new(1, 11)),
3677 ),
3678 )))
3679 },
3680 );
3681
3682 let definitions_2 = project_b
3683 .update(cx_b, |p, cx| p.definition(&buffer_b, 33, cx))
3684 .await
3685 .unwrap();
3686 cx_b.read(|cx| {
3687 assert_eq!(definitions_2.len(), 1);
3688 assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
3689 let target_buffer = definitions_2[0].buffer.read(cx);
3690 assert_eq!(
3691 target_buffer.text(),
3692 "const TWO: usize = 2;\nconst THREE: usize = 3;"
3693 );
3694 assert_eq!(
3695 definitions_2[0].range.to_point(target_buffer),
3696 Point::new(1, 6)..Point::new(1, 11)
3697 );
3698 });
3699 assert_eq!(definitions_1[0].buffer, definitions_2[0].buffer);
3700 }
3701
3702 #[gpui::test(iterations = 10)]
3703 async fn test_references(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3704 cx_a.foreground().forbid_parking();
3705 let lang_registry = Arc::new(LanguageRegistry::test());
3706 let fs = FakeFs::new(cx_a.background());
3707 fs.insert_tree(
3708 "/root-1",
3709 json!({
3710 "one.rs": "const ONE: usize = 1;",
3711 "two.rs": "const TWO: usize = one::ONE + one::ONE;",
3712 }),
3713 )
3714 .await;
3715 fs.insert_tree(
3716 "/root-2",
3717 json!({
3718 "three.rs": "const THREE: usize = two::TWO + one::ONE;",
3719 }),
3720 )
3721 .await;
3722
3723 // Set up a fake language server.
3724 let mut language = Language::new(
3725 LanguageConfig {
3726 name: "Rust".into(),
3727 path_suffixes: vec!["rs".to_string()],
3728 ..Default::default()
3729 },
3730 Some(tree_sitter_rust::language()),
3731 );
3732 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3733 lang_registry.add(Arc::new(language));
3734
3735 // Connect to a server as 2 clients.
3736 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3737 let client_a = server.create_client(cx_a, "user_a").await;
3738 let mut client_b = server.create_client(cx_b, "user_b").await;
3739 server
3740 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3741 .await;
3742
3743 // Share a project as client A
3744 let project_a = cx_a.update(|cx| {
3745 Project::local(
3746 client_a.clone(),
3747 client_a.user_store.clone(),
3748 lang_registry.clone(),
3749 fs.clone(),
3750 cx,
3751 )
3752 });
3753 let (worktree_a, _) = project_a
3754 .update(cx_a, |p, cx| {
3755 p.find_or_create_local_worktree("/root-1", true, cx)
3756 })
3757 .await
3758 .unwrap();
3759 worktree_a
3760 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3761 .await;
3762 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3763
3764 // Join the worktree as client B.
3765 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3766
3767 // Open the file on client B.
3768 let buffer_b = cx_b
3769 .background()
3770 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "one.rs"), cx)))
3771 .await
3772 .unwrap();
3773
3774 // Request references to a symbol as the guest.
3775 let fake_language_server = fake_language_servers.next().await.unwrap();
3776 fake_language_server.handle_request::<lsp::request::References, _, _>(
3777 |params, _| async move {
3778 assert_eq!(
3779 params.text_document_position.text_document.uri.as_str(),
3780 "file:///root-1/one.rs"
3781 );
3782 Ok(Some(vec![
3783 lsp::Location {
3784 uri: lsp::Url::from_file_path("/root-1/two.rs").unwrap(),
3785 range: lsp::Range::new(
3786 lsp::Position::new(0, 24),
3787 lsp::Position::new(0, 27),
3788 ),
3789 },
3790 lsp::Location {
3791 uri: lsp::Url::from_file_path("/root-1/two.rs").unwrap(),
3792 range: lsp::Range::new(
3793 lsp::Position::new(0, 35),
3794 lsp::Position::new(0, 38),
3795 ),
3796 },
3797 lsp::Location {
3798 uri: lsp::Url::from_file_path("/root-2/three.rs").unwrap(),
3799 range: lsp::Range::new(
3800 lsp::Position::new(0, 37),
3801 lsp::Position::new(0, 40),
3802 ),
3803 },
3804 ]))
3805 },
3806 );
3807
3808 let references = project_b
3809 .update(cx_b, |p, cx| p.references(&buffer_b, 7, cx))
3810 .await
3811 .unwrap();
3812 cx_b.read(|cx| {
3813 assert_eq!(references.len(), 3);
3814 assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
3815
3816 let two_buffer = references[0].buffer.read(cx);
3817 let three_buffer = references[2].buffer.read(cx);
3818 assert_eq!(
3819 two_buffer.file().unwrap().path().as_ref(),
3820 Path::new("two.rs")
3821 );
3822 assert_eq!(references[1].buffer, references[0].buffer);
3823 assert_eq!(
3824 three_buffer.file().unwrap().full_path(cx),
3825 Path::new("three.rs")
3826 );
3827
3828 assert_eq!(references[0].range.to_offset(&two_buffer), 24..27);
3829 assert_eq!(references[1].range.to_offset(&two_buffer), 35..38);
3830 assert_eq!(references[2].range.to_offset(&three_buffer), 37..40);
3831 });
3832 }
3833
3834 #[gpui::test(iterations = 10)]
3835 async fn test_project_search(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3836 cx_a.foreground().forbid_parking();
3837 let lang_registry = Arc::new(LanguageRegistry::test());
3838 let fs = FakeFs::new(cx_a.background());
3839 fs.insert_tree(
3840 "/root-1",
3841 json!({
3842 "a": "hello world",
3843 "b": "goodnight moon",
3844 "c": "a world of goo",
3845 "d": "world champion of clown world",
3846 }),
3847 )
3848 .await;
3849 fs.insert_tree(
3850 "/root-2",
3851 json!({
3852 "e": "disney world is fun",
3853 }),
3854 )
3855 .await;
3856
3857 // Connect to a server as 2 clients.
3858 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3859 let client_a = server.create_client(cx_a, "user_a").await;
3860 let mut client_b = server.create_client(cx_b, "user_b").await;
3861 server
3862 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3863 .await;
3864
3865 // Share a project as client A
3866 let project_a = cx_a.update(|cx| {
3867 Project::local(
3868 client_a.clone(),
3869 client_a.user_store.clone(),
3870 lang_registry.clone(),
3871 fs.clone(),
3872 cx,
3873 )
3874 });
3875
3876 let (worktree_1, _) = project_a
3877 .update(cx_a, |p, cx| {
3878 p.find_or_create_local_worktree("/root-1", true, cx)
3879 })
3880 .await
3881 .unwrap();
3882 worktree_1
3883 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3884 .await;
3885 let (worktree_2, _) = project_a
3886 .update(cx_a, |p, cx| {
3887 p.find_or_create_local_worktree("/root-2", true, cx)
3888 })
3889 .await
3890 .unwrap();
3891 worktree_2
3892 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3893 .await;
3894
3895 // Join the worktree as client B.
3896 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3897 let results = project_b
3898 .update(cx_b, |project, cx| {
3899 project.search(SearchQuery::text("world", false, false), cx)
3900 })
3901 .await
3902 .unwrap();
3903
3904 let mut ranges_by_path = results
3905 .into_iter()
3906 .map(|(buffer, ranges)| {
3907 buffer.read_with(cx_b, |buffer, cx| {
3908 let path = buffer.file().unwrap().full_path(cx);
3909 let offset_ranges = ranges
3910 .into_iter()
3911 .map(|range| range.to_offset(buffer))
3912 .collect::<Vec<_>>();
3913 (path, offset_ranges)
3914 })
3915 })
3916 .collect::<Vec<_>>();
3917 ranges_by_path.sort_by_key(|(path, _)| path.clone());
3918
3919 assert_eq!(
3920 ranges_by_path,
3921 &[
3922 (PathBuf::from("root-1/a"), vec![6..11]),
3923 (PathBuf::from("root-1/c"), vec![2..7]),
3924 (PathBuf::from("root-1/d"), vec![0..5, 24..29]),
3925 (PathBuf::from("root-2/e"), vec![7..12]),
3926 ]
3927 );
3928 }
3929
3930 #[gpui::test(iterations = 10)]
3931 async fn test_document_highlights(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3932 cx_a.foreground().forbid_parking();
3933 let lang_registry = Arc::new(LanguageRegistry::test());
3934 let fs = FakeFs::new(cx_a.background());
3935 fs.insert_tree(
3936 "/root-1",
3937 json!({
3938 "main.rs": "fn double(number: i32) -> i32 { number + number }",
3939 }),
3940 )
3941 .await;
3942
3943 // Set up a fake language server.
3944 let mut language = Language::new(
3945 LanguageConfig {
3946 name: "Rust".into(),
3947 path_suffixes: vec!["rs".to_string()],
3948 ..Default::default()
3949 },
3950 Some(tree_sitter_rust::language()),
3951 );
3952 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3953 lang_registry.add(Arc::new(language));
3954
3955 // Connect to a server as 2 clients.
3956 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3957 let client_a = server.create_client(cx_a, "user_a").await;
3958 let mut client_b = server.create_client(cx_b, "user_b").await;
3959 server
3960 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3961 .await;
3962
3963 // Share a project as client A
3964 let project_a = cx_a.update(|cx| {
3965 Project::local(
3966 client_a.clone(),
3967 client_a.user_store.clone(),
3968 lang_registry.clone(),
3969 fs.clone(),
3970 cx,
3971 )
3972 });
3973 let (worktree_a, _) = project_a
3974 .update(cx_a, |p, cx| {
3975 p.find_or_create_local_worktree("/root-1", true, cx)
3976 })
3977 .await
3978 .unwrap();
3979 worktree_a
3980 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3981 .await;
3982 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3983
3984 // Join the worktree as client B.
3985 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3986
3987 // Open the file on client B.
3988 let buffer_b = cx_b
3989 .background()
3990 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "main.rs"), cx)))
3991 .await
3992 .unwrap();
3993
3994 // Request document highlights as the guest.
3995 let fake_language_server = fake_language_servers.next().await.unwrap();
3996 fake_language_server.handle_request::<lsp::request::DocumentHighlightRequest, _, _>(
3997 |params, _| async move {
3998 assert_eq!(
3999 params
4000 .text_document_position_params
4001 .text_document
4002 .uri
4003 .as_str(),
4004 "file:///root-1/main.rs"
4005 );
4006 assert_eq!(
4007 params.text_document_position_params.position,
4008 lsp::Position::new(0, 34)
4009 );
4010 Ok(Some(vec![
4011 lsp::DocumentHighlight {
4012 kind: Some(lsp::DocumentHighlightKind::WRITE),
4013 range: lsp::Range::new(
4014 lsp::Position::new(0, 10),
4015 lsp::Position::new(0, 16),
4016 ),
4017 },
4018 lsp::DocumentHighlight {
4019 kind: Some(lsp::DocumentHighlightKind::READ),
4020 range: lsp::Range::new(
4021 lsp::Position::new(0, 32),
4022 lsp::Position::new(0, 38),
4023 ),
4024 },
4025 lsp::DocumentHighlight {
4026 kind: Some(lsp::DocumentHighlightKind::READ),
4027 range: lsp::Range::new(
4028 lsp::Position::new(0, 41),
4029 lsp::Position::new(0, 47),
4030 ),
4031 },
4032 ]))
4033 },
4034 );
4035
4036 let highlights = project_b
4037 .update(cx_b, |p, cx| p.document_highlights(&buffer_b, 34, cx))
4038 .await
4039 .unwrap();
4040 buffer_b.read_with(cx_b, |buffer, _| {
4041 let snapshot = buffer.snapshot();
4042
4043 let highlights = highlights
4044 .into_iter()
4045 .map(|highlight| (highlight.kind, highlight.range.to_offset(&snapshot)))
4046 .collect::<Vec<_>>();
4047 assert_eq!(
4048 highlights,
4049 &[
4050 (lsp::DocumentHighlightKind::WRITE, 10..16),
4051 (lsp::DocumentHighlightKind::READ, 32..38),
4052 (lsp::DocumentHighlightKind::READ, 41..47)
4053 ]
4054 )
4055 });
4056 }
4057
4058 #[gpui::test(iterations = 10)]
4059 async fn test_project_symbols(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4060 cx_a.foreground().forbid_parking();
4061 let lang_registry = Arc::new(LanguageRegistry::test());
4062 let fs = FakeFs::new(cx_a.background());
4063 fs.insert_tree(
4064 "/code",
4065 json!({
4066 "crate-1": {
4067 "one.rs": "const ONE: usize = 1;",
4068 },
4069 "crate-2": {
4070 "two.rs": "const TWO: usize = 2; const THREE: usize = 3;",
4071 },
4072 "private": {
4073 "passwords.txt": "the-password",
4074 }
4075 }),
4076 )
4077 .await;
4078
4079 // Set up a fake language server.
4080 let mut language = Language::new(
4081 LanguageConfig {
4082 name: "Rust".into(),
4083 path_suffixes: vec!["rs".to_string()],
4084 ..Default::default()
4085 },
4086 Some(tree_sitter_rust::language()),
4087 );
4088 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
4089 lang_registry.add(Arc::new(language));
4090
4091 // Connect to a server as 2 clients.
4092 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4093 let client_a = server.create_client(cx_a, "user_a").await;
4094 let mut client_b = server.create_client(cx_b, "user_b").await;
4095 server
4096 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
4097 .await;
4098
4099 // Share a project as client A
4100 let project_a = cx_a.update(|cx| {
4101 Project::local(
4102 client_a.clone(),
4103 client_a.user_store.clone(),
4104 lang_registry.clone(),
4105 fs.clone(),
4106 cx,
4107 )
4108 });
4109 let (worktree_a, _) = project_a
4110 .update(cx_a, |p, cx| {
4111 p.find_or_create_local_worktree("/code/crate-1", true, cx)
4112 })
4113 .await
4114 .unwrap();
4115 worktree_a
4116 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
4117 .await;
4118 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
4119
4120 // Join the worktree as client B.
4121 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
4122
4123 // Cause the language server to start.
4124 let _buffer = cx_b
4125 .background()
4126 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "one.rs"), cx)))
4127 .await
4128 .unwrap();
4129
4130 let fake_language_server = fake_language_servers.next().await.unwrap();
4131 fake_language_server.handle_request::<lsp::request::WorkspaceSymbol, _, _>(
4132 |_, _| async move {
4133 #[allow(deprecated)]
4134 Ok(Some(vec![lsp::SymbolInformation {
4135 name: "TWO".into(),
4136 location: lsp::Location {
4137 uri: lsp::Url::from_file_path("/code/crate-2/two.rs").unwrap(),
4138 range: lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
4139 },
4140 kind: lsp::SymbolKind::CONSTANT,
4141 tags: None,
4142 container_name: None,
4143 deprecated: None,
4144 }]))
4145 },
4146 );
4147
4148 // Request the definition of a symbol as the guest.
4149 let symbols = project_b
4150 .update(cx_b, |p, cx| p.symbols("two", cx))
4151 .await
4152 .unwrap();
4153 assert_eq!(symbols.len(), 1);
4154 assert_eq!(symbols[0].name, "TWO");
4155
4156 // Open one of the returned symbols.
4157 let buffer_b_2 = project_b
4158 .update(cx_b, |project, cx| {
4159 project.open_buffer_for_symbol(&symbols[0], cx)
4160 })
4161 .await
4162 .unwrap();
4163 buffer_b_2.read_with(cx_b, |buffer, _| {
4164 assert_eq!(
4165 buffer.file().unwrap().path().as_ref(),
4166 Path::new("../crate-2/two.rs")
4167 );
4168 });
4169
4170 // Attempt to craft a symbol and violate host's privacy by opening an arbitrary file.
4171 let mut fake_symbol = symbols[0].clone();
4172 fake_symbol.path = Path::new("/code/secrets").into();
4173 let error = project_b
4174 .update(cx_b, |project, cx| {
4175 project.open_buffer_for_symbol(&fake_symbol, cx)
4176 })
4177 .await
4178 .unwrap_err();
4179 assert!(error.to_string().contains("invalid symbol signature"));
4180 }
4181
4182 #[gpui::test(iterations = 10)]
4183 async fn test_open_buffer_while_getting_definition_pointing_to_it(
4184 cx_a: &mut TestAppContext,
4185 cx_b: &mut TestAppContext,
4186 mut rng: StdRng,
4187 ) {
4188 cx_a.foreground().forbid_parking();
4189 let lang_registry = Arc::new(LanguageRegistry::test());
4190 let fs = FakeFs::new(cx_a.background());
4191 fs.insert_tree(
4192 "/root",
4193 json!({
4194 "a.rs": "const ONE: usize = b::TWO;",
4195 "b.rs": "const TWO: usize = 2",
4196 }),
4197 )
4198 .await;
4199
4200 // Set up a fake language server.
4201 let mut language = Language::new(
4202 LanguageConfig {
4203 name: "Rust".into(),
4204 path_suffixes: vec!["rs".to_string()],
4205 ..Default::default()
4206 },
4207 Some(tree_sitter_rust::language()),
4208 );
4209 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
4210 lang_registry.add(Arc::new(language));
4211
4212 // Connect to a server as 2 clients.
4213 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4214 let client_a = server.create_client(cx_a, "user_a").await;
4215 let mut client_b = server.create_client(cx_b, "user_b").await;
4216 server
4217 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
4218 .await;
4219
4220 // Share a project as client A
4221 let project_a = cx_a.update(|cx| {
4222 Project::local(
4223 client_a.clone(),
4224 client_a.user_store.clone(),
4225 lang_registry.clone(),
4226 fs.clone(),
4227 cx,
4228 )
4229 });
4230
4231 let (worktree_a, _) = project_a
4232 .update(cx_a, |p, cx| {
4233 p.find_or_create_local_worktree("/root", true, cx)
4234 })
4235 .await
4236 .unwrap();
4237 worktree_a
4238 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
4239 .await;
4240 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
4241
4242 // Join the project as client B.
4243 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
4244
4245 let buffer_b1 = cx_b
4246 .background()
4247 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
4248 .await
4249 .unwrap();
4250
4251 let fake_language_server = fake_language_servers.next().await.unwrap();
4252 fake_language_server.handle_request::<lsp::request::GotoDefinition, _, _>(
4253 |_, _| async move {
4254 Ok(Some(lsp::GotoDefinitionResponse::Scalar(
4255 lsp::Location::new(
4256 lsp::Url::from_file_path("/root/b.rs").unwrap(),
4257 lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
4258 ),
4259 )))
4260 },
4261 );
4262
4263 let definitions;
4264 let buffer_b2;
4265 if rng.gen() {
4266 definitions = project_b.update(cx_b, |p, cx| p.definition(&buffer_b1, 23, cx));
4267 buffer_b2 = project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "b.rs"), cx));
4268 } else {
4269 buffer_b2 = project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "b.rs"), cx));
4270 definitions = project_b.update(cx_b, |p, cx| p.definition(&buffer_b1, 23, cx));
4271 }
4272
4273 let buffer_b2 = buffer_b2.await.unwrap();
4274 let definitions = definitions.await.unwrap();
4275 assert_eq!(definitions.len(), 1);
4276 assert_eq!(definitions[0].buffer, buffer_b2);
4277 }
4278
4279 #[gpui::test(iterations = 10)]
4280 async fn test_collaborating_with_code_actions(
4281 cx_a: &mut TestAppContext,
4282 cx_b: &mut TestAppContext,
4283 ) {
4284 cx_a.foreground().forbid_parking();
4285 let lang_registry = Arc::new(LanguageRegistry::test());
4286 let fs = FakeFs::new(cx_a.background());
4287 cx_b.update(|cx| editor::init(cx));
4288
4289 // Set up a fake language server.
4290 let mut language = Language::new(
4291 LanguageConfig {
4292 name: "Rust".into(),
4293 path_suffixes: vec!["rs".to_string()],
4294 ..Default::default()
4295 },
4296 Some(tree_sitter_rust::language()),
4297 );
4298 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
4299 lang_registry.add(Arc::new(language));
4300
4301 // Connect to a server as 2 clients.
4302 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4303 let client_a = server.create_client(cx_a, "user_a").await;
4304 let mut client_b = server.create_client(cx_b, "user_b").await;
4305 server
4306 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
4307 .await;
4308
4309 // Share a project as client A
4310 fs.insert_tree(
4311 "/a",
4312 json!({
4313 "main.rs": "mod other;\nfn main() { let foo = other::foo(); }",
4314 "other.rs": "pub fn foo() -> usize { 4 }",
4315 }),
4316 )
4317 .await;
4318 let project_a = cx_a.update(|cx| {
4319 Project::local(
4320 client_a.clone(),
4321 client_a.user_store.clone(),
4322 lang_registry.clone(),
4323 fs.clone(),
4324 cx,
4325 )
4326 });
4327 let (worktree_a, _) = project_a
4328 .update(cx_a, |p, cx| {
4329 p.find_or_create_local_worktree("/a", true, cx)
4330 })
4331 .await
4332 .unwrap();
4333 worktree_a
4334 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
4335 .await;
4336 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
4337
4338 // Join the project as client B.
4339 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
4340 let (_window_b, workspace_b) = cx_b.add_window(|cx| Workspace::new(project_b.clone(), cx));
4341 let editor_b = workspace_b
4342 .update(cx_b, |workspace, cx| {
4343 workspace.open_path((worktree_id, "main.rs"), true, cx)
4344 })
4345 .await
4346 .unwrap()
4347 .downcast::<Editor>()
4348 .unwrap();
4349
4350 let mut fake_language_server = fake_language_servers.next().await.unwrap();
4351 fake_language_server
4352 .handle_request::<lsp::request::CodeActionRequest, _, _>(|params, _| async move {
4353 assert_eq!(
4354 params.text_document.uri,
4355 lsp::Url::from_file_path("/a/main.rs").unwrap(),
4356 );
4357 assert_eq!(params.range.start, lsp::Position::new(0, 0));
4358 assert_eq!(params.range.end, lsp::Position::new(0, 0));
4359 Ok(None)
4360 })
4361 .next()
4362 .await;
4363
4364 // Move cursor to a location that contains code actions.
4365 editor_b.update(cx_b, |editor, cx| {
4366 editor.change_selections(None, cx, |s| {
4367 s.select_ranges([Point::new(1, 31)..Point::new(1, 31)])
4368 });
4369 cx.focus(&editor_b);
4370 });
4371
4372 fake_language_server
4373 .handle_request::<lsp::request::CodeActionRequest, _, _>(|params, _| async move {
4374 assert_eq!(
4375 params.text_document.uri,
4376 lsp::Url::from_file_path("/a/main.rs").unwrap(),
4377 );
4378 assert_eq!(params.range.start, lsp::Position::new(1, 31));
4379 assert_eq!(params.range.end, lsp::Position::new(1, 31));
4380
4381 Ok(Some(vec![lsp::CodeActionOrCommand::CodeAction(
4382 lsp::CodeAction {
4383 title: "Inline into all callers".to_string(),
4384 edit: Some(lsp::WorkspaceEdit {
4385 changes: Some(
4386 [
4387 (
4388 lsp::Url::from_file_path("/a/main.rs").unwrap(),
4389 vec![lsp::TextEdit::new(
4390 lsp::Range::new(
4391 lsp::Position::new(1, 22),
4392 lsp::Position::new(1, 34),
4393 ),
4394 "4".to_string(),
4395 )],
4396 ),
4397 (
4398 lsp::Url::from_file_path("/a/other.rs").unwrap(),
4399 vec![lsp::TextEdit::new(
4400 lsp::Range::new(
4401 lsp::Position::new(0, 0),
4402 lsp::Position::new(0, 27),
4403 ),
4404 "".to_string(),
4405 )],
4406 ),
4407 ]
4408 .into_iter()
4409 .collect(),
4410 ),
4411 ..Default::default()
4412 }),
4413 data: Some(json!({
4414 "codeActionParams": {
4415 "range": {
4416 "start": {"line": 1, "column": 31},
4417 "end": {"line": 1, "column": 31},
4418 }
4419 }
4420 })),
4421 ..Default::default()
4422 },
4423 )]))
4424 })
4425 .next()
4426 .await;
4427
4428 // Toggle code actions and wait for them to display.
4429 editor_b.update(cx_b, |editor, cx| {
4430 editor.toggle_code_actions(
4431 &ToggleCodeActions {
4432 deployed_from_indicator: false,
4433 },
4434 cx,
4435 );
4436 });
4437 editor_b
4438 .condition(&cx_b, |editor, _| editor.context_menu_visible())
4439 .await;
4440
4441 fake_language_server.remove_request_handler::<lsp::request::CodeActionRequest>();
4442
4443 // Confirming the code action will trigger a resolve request.
4444 let confirm_action = workspace_b
4445 .update(cx_b, |workspace, cx| {
4446 Editor::confirm_code_action(workspace, &ConfirmCodeAction { item_ix: Some(0) }, cx)
4447 })
4448 .unwrap();
4449 fake_language_server.handle_request::<lsp::request::CodeActionResolveRequest, _, _>(
4450 |_, _| async move {
4451 Ok(lsp::CodeAction {
4452 title: "Inline into all callers".to_string(),
4453 edit: Some(lsp::WorkspaceEdit {
4454 changes: Some(
4455 [
4456 (
4457 lsp::Url::from_file_path("/a/main.rs").unwrap(),
4458 vec![lsp::TextEdit::new(
4459 lsp::Range::new(
4460 lsp::Position::new(1, 22),
4461 lsp::Position::new(1, 34),
4462 ),
4463 "4".to_string(),
4464 )],
4465 ),
4466 (
4467 lsp::Url::from_file_path("/a/other.rs").unwrap(),
4468 vec![lsp::TextEdit::new(
4469 lsp::Range::new(
4470 lsp::Position::new(0, 0),
4471 lsp::Position::new(0, 27),
4472 ),
4473 "".to_string(),
4474 )],
4475 ),
4476 ]
4477 .into_iter()
4478 .collect(),
4479 ),
4480 ..Default::default()
4481 }),
4482 ..Default::default()
4483 })
4484 },
4485 );
4486
4487 // After the action is confirmed, an editor containing both modified files is opened.
4488 confirm_action.await.unwrap();
4489 let code_action_editor = workspace_b.read_with(cx_b, |workspace, cx| {
4490 workspace
4491 .active_item(cx)
4492 .unwrap()
4493 .downcast::<Editor>()
4494 .unwrap()
4495 });
4496 code_action_editor.update(cx_b, |editor, cx| {
4497 assert_eq!(editor.text(cx), "mod other;\nfn main() { let foo = 4; }\n");
4498 editor.undo(&Undo, cx);
4499 assert_eq!(
4500 editor.text(cx),
4501 "mod other;\nfn main() { let foo = other::foo(); }\npub fn foo() -> usize { 4 }"
4502 );
4503 editor.redo(&Redo, cx);
4504 assert_eq!(editor.text(cx), "mod other;\nfn main() { let foo = 4; }\n");
4505 });
4506 }
4507
4508 #[gpui::test(iterations = 10)]
4509 async fn test_collaborating_with_renames(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4510 cx_a.foreground().forbid_parking();
4511 let lang_registry = Arc::new(LanguageRegistry::test());
4512 let fs = FakeFs::new(cx_a.background());
4513 cx_b.update(|cx| editor::init(cx));
4514
4515 // Set up a fake language server.
4516 let mut language = Language::new(
4517 LanguageConfig {
4518 name: "Rust".into(),
4519 path_suffixes: vec!["rs".to_string()],
4520 ..Default::default()
4521 },
4522 Some(tree_sitter_rust::language()),
4523 );
4524 let mut fake_language_servers = language.set_fake_lsp_adapter(FakeLspAdapter {
4525 capabilities: lsp::ServerCapabilities {
4526 rename_provider: Some(lsp::OneOf::Right(lsp::RenameOptions {
4527 prepare_provider: Some(true),
4528 work_done_progress_options: Default::default(),
4529 })),
4530 ..Default::default()
4531 },
4532 ..Default::default()
4533 });
4534 lang_registry.add(Arc::new(language));
4535
4536 // Connect to a server as 2 clients.
4537 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4538 let client_a = server.create_client(cx_a, "user_a").await;
4539 let mut client_b = server.create_client(cx_b, "user_b").await;
4540 server
4541 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
4542 .await;
4543
4544 // Share a project as client A
4545 fs.insert_tree(
4546 "/dir",
4547 json!({
4548 "one.rs": "const ONE: usize = 1;",
4549 "two.rs": "const TWO: usize = one::ONE + one::ONE;"
4550 }),
4551 )
4552 .await;
4553 let project_a = cx_a.update(|cx| {
4554 Project::local(
4555 client_a.clone(),
4556 client_a.user_store.clone(),
4557 lang_registry.clone(),
4558 fs.clone(),
4559 cx,
4560 )
4561 });
4562 let (worktree_a, _) = project_a
4563 .update(cx_a, |p, cx| {
4564 p.find_or_create_local_worktree("/dir", true, cx)
4565 })
4566 .await
4567 .unwrap();
4568 worktree_a
4569 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
4570 .await;
4571 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
4572
4573 // Join the worktree as client B.
4574 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
4575 let (_window_b, workspace_b) = cx_b.add_window(|cx| Workspace::new(project_b.clone(), cx));
4576 let editor_b = workspace_b
4577 .update(cx_b, |workspace, cx| {
4578 workspace.open_path((worktree_id, "one.rs"), true, cx)
4579 })
4580 .await
4581 .unwrap()
4582 .downcast::<Editor>()
4583 .unwrap();
4584 let fake_language_server = fake_language_servers.next().await.unwrap();
4585
4586 // Move cursor to a location that can be renamed.
4587 let prepare_rename = editor_b.update(cx_b, |editor, cx| {
4588 editor.change_selections(None, cx, |s| s.select_ranges([7..7]));
4589 editor.rename(&Rename, cx).unwrap()
4590 });
4591
4592 fake_language_server
4593 .handle_request::<lsp::request::PrepareRenameRequest, _, _>(|params, _| async move {
4594 assert_eq!(params.text_document.uri.as_str(), "file:///dir/one.rs");
4595 assert_eq!(params.position, lsp::Position::new(0, 7));
4596 Ok(Some(lsp::PrepareRenameResponse::Range(lsp::Range::new(
4597 lsp::Position::new(0, 6),
4598 lsp::Position::new(0, 9),
4599 ))))
4600 })
4601 .next()
4602 .await
4603 .unwrap();
4604 prepare_rename.await.unwrap();
4605 editor_b.update(cx_b, |editor, cx| {
4606 let rename = editor.pending_rename().unwrap();
4607 let buffer = editor.buffer().read(cx).snapshot(cx);
4608 assert_eq!(
4609 rename.range.start.to_offset(&buffer)..rename.range.end.to_offset(&buffer),
4610 6..9
4611 );
4612 rename.editor.update(cx, |rename_editor, cx| {
4613 rename_editor.buffer().update(cx, |rename_buffer, cx| {
4614 rename_buffer.edit([(0..3, "THREE")], cx);
4615 });
4616 });
4617 });
4618
4619 let confirm_rename = workspace_b.update(cx_b, |workspace, cx| {
4620 Editor::confirm_rename(workspace, &ConfirmRename, cx).unwrap()
4621 });
4622 fake_language_server
4623 .handle_request::<lsp::request::Rename, _, _>(|params, _| async move {
4624 assert_eq!(
4625 params.text_document_position.text_document.uri.as_str(),
4626 "file:///dir/one.rs"
4627 );
4628 assert_eq!(
4629 params.text_document_position.position,
4630 lsp::Position::new(0, 6)
4631 );
4632 assert_eq!(params.new_name, "THREE");
4633 Ok(Some(lsp::WorkspaceEdit {
4634 changes: Some(
4635 [
4636 (
4637 lsp::Url::from_file_path("/dir/one.rs").unwrap(),
4638 vec![lsp::TextEdit::new(
4639 lsp::Range::new(
4640 lsp::Position::new(0, 6),
4641 lsp::Position::new(0, 9),
4642 ),
4643 "THREE".to_string(),
4644 )],
4645 ),
4646 (
4647 lsp::Url::from_file_path("/dir/two.rs").unwrap(),
4648 vec![
4649 lsp::TextEdit::new(
4650 lsp::Range::new(
4651 lsp::Position::new(0, 24),
4652 lsp::Position::new(0, 27),
4653 ),
4654 "THREE".to_string(),
4655 ),
4656 lsp::TextEdit::new(
4657 lsp::Range::new(
4658 lsp::Position::new(0, 35),
4659 lsp::Position::new(0, 38),
4660 ),
4661 "THREE".to_string(),
4662 ),
4663 ],
4664 ),
4665 ]
4666 .into_iter()
4667 .collect(),
4668 ),
4669 ..Default::default()
4670 }))
4671 })
4672 .next()
4673 .await
4674 .unwrap();
4675 confirm_rename.await.unwrap();
4676
4677 let rename_editor = workspace_b.read_with(cx_b, |workspace, cx| {
4678 workspace
4679 .active_item(cx)
4680 .unwrap()
4681 .downcast::<Editor>()
4682 .unwrap()
4683 });
4684 rename_editor.update(cx_b, |editor, cx| {
4685 assert_eq!(
4686 editor.text(cx),
4687 "const THREE: usize = 1;\nconst TWO: usize = one::THREE + one::THREE;"
4688 );
4689 editor.undo(&Undo, cx);
4690 assert_eq!(
4691 editor.text(cx),
4692 "const ONE: usize = 1;\nconst TWO: usize = one::ONE + one::ONE;"
4693 );
4694 editor.redo(&Redo, cx);
4695 assert_eq!(
4696 editor.text(cx),
4697 "const THREE: usize = 1;\nconst TWO: usize = one::THREE + one::THREE;"
4698 );
4699 });
4700
4701 // Ensure temporary rename edits cannot be undone/redone.
4702 editor_b.update(cx_b, |editor, cx| {
4703 editor.undo(&Undo, cx);
4704 assert_eq!(editor.text(cx), "const ONE: usize = 1;");
4705 editor.undo(&Undo, cx);
4706 assert_eq!(editor.text(cx), "const ONE: usize = 1;");
4707 editor.redo(&Redo, cx);
4708 assert_eq!(editor.text(cx), "const THREE: usize = 1;");
4709 })
4710 }
4711
4712 #[gpui::test(iterations = 10)]
4713 async fn test_basic_chat(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4714 cx_a.foreground().forbid_parking();
4715
4716 // Connect to a server as 2 clients.
4717 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4718 let client_a = server.create_client(cx_a, "user_a").await;
4719 let client_b = server.create_client(cx_b, "user_b").await;
4720
4721 // Create an org that includes these 2 users.
4722 let db = &server.app_state.db;
4723 let org_id = db.create_org("Test Org", "test-org").await.unwrap();
4724 db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
4725 .await
4726 .unwrap();
4727 db.add_org_member(org_id, client_b.current_user_id(&cx_b), false)
4728 .await
4729 .unwrap();
4730
4731 // Create a channel that includes all the users.
4732 let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
4733 db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
4734 .await
4735 .unwrap();
4736 db.add_channel_member(channel_id, client_b.current_user_id(&cx_b), false)
4737 .await
4738 .unwrap();
4739 db.create_channel_message(
4740 channel_id,
4741 client_b.current_user_id(&cx_b),
4742 "hello A, it's B.",
4743 OffsetDateTime::now_utc(),
4744 1,
4745 )
4746 .await
4747 .unwrap();
4748
4749 let channels_a = cx_a
4750 .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
4751 channels_a
4752 .condition(cx_a, |list, _| list.available_channels().is_some())
4753 .await;
4754 channels_a.read_with(cx_a, |list, _| {
4755 assert_eq!(
4756 list.available_channels().unwrap(),
4757 &[ChannelDetails {
4758 id: channel_id.to_proto(),
4759 name: "test-channel".to_string()
4760 }]
4761 )
4762 });
4763 let channel_a = channels_a.update(cx_a, |this, cx| {
4764 this.get_channel(channel_id.to_proto(), cx).unwrap()
4765 });
4766 channel_a.read_with(cx_a, |channel, _| assert!(channel.messages().is_empty()));
4767 channel_a
4768 .condition(&cx_a, |channel, _| {
4769 channel_messages(channel)
4770 == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4771 })
4772 .await;
4773
4774 let channels_b = cx_b
4775 .add_model(|cx| ChannelList::new(client_b.user_store.clone(), client_b.clone(), cx));
4776 channels_b
4777 .condition(cx_b, |list, _| list.available_channels().is_some())
4778 .await;
4779 channels_b.read_with(cx_b, |list, _| {
4780 assert_eq!(
4781 list.available_channels().unwrap(),
4782 &[ChannelDetails {
4783 id: channel_id.to_proto(),
4784 name: "test-channel".to_string()
4785 }]
4786 )
4787 });
4788
4789 let channel_b = channels_b.update(cx_b, |this, cx| {
4790 this.get_channel(channel_id.to_proto(), cx).unwrap()
4791 });
4792 channel_b.read_with(cx_b, |channel, _| assert!(channel.messages().is_empty()));
4793 channel_b
4794 .condition(&cx_b, |channel, _| {
4795 channel_messages(channel)
4796 == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4797 })
4798 .await;
4799
4800 channel_a
4801 .update(cx_a, |channel, cx| {
4802 channel
4803 .send_message("oh, hi B.".to_string(), cx)
4804 .unwrap()
4805 .detach();
4806 let task = channel.send_message("sup".to_string(), cx).unwrap();
4807 assert_eq!(
4808 channel_messages(channel),
4809 &[
4810 ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4811 ("user_a".to_string(), "oh, hi B.".to_string(), true),
4812 ("user_a".to_string(), "sup".to_string(), true)
4813 ]
4814 );
4815 task
4816 })
4817 .await
4818 .unwrap();
4819
4820 channel_b
4821 .condition(&cx_b, |channel, _| {
4822 channel_messages(channel)
4823 == [
4824 ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4825 ("user_a".to_string(), "oh, hi B.".to_string(), false),
4826 ("user_a".to_string(), "sup".to_string(), false),
4827 ]
4828 })
4829 .await;
4830
4831 assert_eq!(
4832 server
4833 .state()
4834 .await
4835 .channel(channel_id)
4836 .unwrap()
4837 .connection_ids
4838 .len(),
4839 2
4840 );
4841 cx_b.update(|_| drop(channel_b));
4842 server
4843 .condition(|state| state.channel(channel_id).unwrap().connection_ids.len() == 1)
4844 .await;
4845
4846 cx_a.update(|_| drop(channel_a));
4847 server
4848 .condition(|state| state.channel(channel_id).is_none())
4849 .await;
4850 }
4851
4852 #[gpui::test(iterations = 10)]
4853 async fn test_chat_message_validation(cx_a: &mut TestAppContext) {
4854 cx_a.foreground().forbid_parking();
4855
4856 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4857 let client_a = server.create_client(cx_a, "user_a").await;
4858
4859 let db = &server.app_state.db;
4860 let org_id = db.create_org("Test Org", "test-org").await.unwrap();
4861 let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
4862 db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
4863 .await
4864 .unwrap();
4865 db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
4866 .await
4867 .unwrap();
4868
4869 let channels_a = cx_a
4870 .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
4871 channels_a
4872 .condition(cx_a, |list, _| list.available_channels().is_some())
4873 .await;
4874 let channel_a = channels_a.update(cx_a, |this, cx| {
4875 this.get_channel(channel_id.to_proto(), cx).unwrap()
4876 });
4877
4878 // Messages aren't allowed to be too long.
4879 channel_a
4880 .update(cx_a, |channel, cx| {
4881 let long_body = "this is long.\n".repeat(1024);
4882 channel.send_message(long_body, cx).unwrap()
4883 })
4884 .await
4885 .unwrap_err();
4886
4887 // Messages aren't allowed to be blank.
4888 channel_a.update(cx_a, |channel, cx| {
4889 channel.send_message(String::new(), cx).unwrap_err()
4890 });
4891
4892 // Leading and trailing whitespace are trimmed.
4893 channel_a
4894 .update(cx_a, |channel, cx| {
4895 channel
4896 .send_message("\n surrounded by whitespace \n".to_string(), cx)
4897 .unwrap()
4898 })
4899 .await
4900 .unwrap();
4901 assert_eq!(
4902 db.get_channel_messages(channel_id, 10, None)
4903 .await
4904 .unwrap()
4905 .iter()
4906 .map(|m| &m.body)
4907 .collect::<Vec<_>>(),
4908 &["surrounded by whitespace"]
4909 );
4910 }
4911
4912 #[gpui::test(iterations = 10)]
4913 async fn test_chat_reconnection(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4914 cx_a.foreground().forbid_parking();
4915
4916 // Connect to a server as 2 clients.
4917 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4918 let client_a = server.create_client(cx_a, "user_a").await;
4919 let client_b = server.create_client(cx_b, "user_b").await;
4920 let mut status_b = client_b.status();
4921
4922 // Create an org that includes these 2 users.
4923 let db = &server.app_state.db;
4924 let org_id = db.create_org("Test Org", "test-org").await.unwrap();
4925 db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
4926 .await
4927 .unwrap();
4928 db.add_org_member(org_id, client_b.current_user_id(&cx_b), false)
4929 .await
4930 .unwrap();
4931
4932 // Create a channel that includes all the users.
4933 let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
4934 db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
4935 .await
4936 .unwrap();
4937 db.add_channel_member(channel_id, client_b.current_user_id(&cx_b), false)
4938 .await
4939 .unwrap();
4940 db.create_channel_message(
4941 channel_id,
4942 client_b.current_user_id(&cx_b),
4943 "hello A, it's B.",
4944 OffsetDateTime::now_utc(),
4945 2,
4946 )
4947 .await
4948 .unwrap();
4949
4950 let channels_a = cx_a
4951 .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
4952 channels_a
4953 .condition(cx_a, |list, _| list.available_channels().is_some())
4954 .await;
4955
4956 channels_a.read_with(cx_a, |list, _| {
4957 assert_eq!(
4958 list.available_channels().unwrap(),
4959 &[ChannelDetails {
4960 id: channel_id.to_proto(),
4961 name: "test-channel".to_string()
4962 }]
4963 )
4964 });
4965 let channel_a = channels_a.update(cx_a, |this, cx| {
4966 this.get_channel(channel_id.to_proto(), cx).unwrap()
4967 });
4968 channel_a.read_with(cx_a, |channel, _| assert!(channel.messages().is_empty()));
4969 channel_a
4970 .condition(&cx_a, |channel, _| {
4971 channel_messages(channel)
4972 == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4973 })
4974 .await;
4975
4976 let channels_b = cx_b
4977 .add_model(|cx| ChannelList::new(client_b.user_store.clone(), client_b.clone(), cx));
4978 channels_b
4979 .condition(cx_b, |list, _| list.available_channels().is_some())
4980 .await;
4981 channels_b.read_with(cx_b, |list, _| {
4982 assert_eq!(
4983 list.available_channels().unwrap(),
4984 &[ChannelDetails {
4985 id: channel_id.to_proto(),
4986 name: "test-channel".to_string()
4987 }]
4988 )
4989 });
4990
4991 let channel_b = channels_b.update(cx_b, |this, cx| {
4992 this.get_channel(channel_id.to_proto(), cx).unwrap()
4993 });
4994 channel_b.read_with(cx_b, |channel, _| assert!(channel.messages().is_empty()));
4995 channel_b
4996 .condition(&cx_b, |channel, _| {
4997 channel_messages(channel)
4998 == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4999 })
5000 .await;
5001
5002 // Disconnect client B, ensuring we can still access its cached channel data.
5003 server.forbid_connections();
5004 server.disconnect_client(client_b.current_user_id(&cx_b));
5005 cx_b.foreground().advance_clock(rpc::RECEIVE_TIMEOUT);
5006 while !matches!(
5007 status_b.next().await,
5008 Some(client::Status::ReconnectionError { .. })
5009 ) {}
5010
5011 channels_b.read_with(cx_b, |channels, _| {
5012 assert_eq!(
5013 channels.available_channels().unwrap(),
5014 [ChannelDetails {
5015 id: channel_id.to_proto(),
5016 name: "test-channel".to_string()
5017 }]
5018 )
5019 });
5020 channel_b.read_with(cx_b, |channel, _| {
5021 assert_eq!(
5022 channel_messages(channel),
5023 [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
5024 )
5025 });
5026
5027 // Send a message from client B while it is disconnected.
5028 channel_b
5029 .update(cx_b, |channel, cx| {
5030 let task = channel
5031 .send_message("can you see this?".to_string(), cx)
5032 .unwrap();
5033 assert_eq!(
5034 channel_messages(channel),
5035 &[
5036 ("user_b".to_string(), "hello A, it's B.".to_string(), false),
5037 ("user_b".to_string(), "can you see this?".to_string(), true)
5038 ]
5039 );
5040 task
5041 })
5042 .await
5043 .unwrap_err();
5044
5045 // Send a message from client A while B is disconnected.
5046 channel_a
5047 .update(cx_a, |channel, cx| {
5048 channel
5049 .send_message("oh, hi B.".to_string(), cx)
5050 .unwrap()
5051 .detach();
5052 let task = channel.send_message("sup".to_string(), cx).unwrap();
5053 assert_eq!(
5054 channel_messages(channel),
5055 &[
5056 ("user_b".to_string(), "hello A, it's B.".to_string(), false),
5057 ("user_a".to_string(), "oh, hi B.".to_string(), true),
5058 ("user_a".to_string(), "sup".to_string(), true)
5059 ]
5060 );
5061 task
5062 })
5063 .await
5064 .unwrap();
5065
5066 // Give client B a chance to reconnect.
5067 server.allow_connections();
5068 cx_b.foreground().advance_clock(Duration::from_secs(10));
5069
5070 // Verify that B sees the new messages upon reconnection, as well as the message client B
5071 // sent while offline.
5072 channel_b
5073 .condition(&cx_b, |channel, _| {
5074 channel_messages(channel)
5075 == [
5076 ("user_b".to_string(), "hello A, it's B.".to_string(), false),
5077 ("user_a".to_string(), "oh, hi B.".to_string(), false),
5078 ("user_a".to_string(), "sup".to_string(), false),
5079 ("user_b".to_string(), "can you see this?".to_string(), false),
5080 ]
5081 })
5082 .await;
5083
5084 // Ensure client A and B can communicate normally after reconnection.
5085 channel_a
5086 .update(cx_a, |channel, cx| {
5087 channel.send_message("you online?".to_string(), cx).unwrap()
5088 })
5089 .await
5090 .unwrap();
5091 channel_b
5092 .condition(&cx_b, |channel, _| {
5093 channel_messages(channel)
5094 == [
5095 ("user_b".to_string(), "hello A, it's B.".to_string(), false),
5096 ("user_a".to_string(), "oh, hi B.".to_string(), false),
5097 ("user_a".to_string(), "sup".to_string(), false),
5098 ("user_b".to_string(), "can you see this?".to_string(), false),
5099 ("user_a".to_string(), "you online?".to_string(), false),
5100 ]
5101 })
5102 .await;
5103
5104 channel_b
5105 .update(cx_b, |channel, cx| {
5106 channel.send_message("yep".to_string(), cx).unwrap()
5107 })
5108 .await
5109 .unwrap();
5110 channel_a
5111 .condition(&cx_a, |channel, _| {
5112 channel_messages(channel)
5113 == [
5114 ("user_b".to_string(), "hello A, it's B.".to_string(), false),
5115 ("user_a".to_string(), "oh, hi B.".to_string(), false),
5116 ("user_a".to_string(), "sup".to_string(), false),
5117 ("user_b".to_string(), "can you see this?".to_string(), false),
5118 ("user_a".to_string(), "you online?".to_string(), false),
5119 ("user_b".to_string(), "yep".to_string(), false),
5120 ]
5121 })
5122 .await;
5123 }
5124
5125 #[gpui::test(iterations = 10)]
5126 async fn test_contacts(
5127 deterministic: Arc<Deterministic>,
5128 cx_a: &mut TestAppContext,
5129 cx_b: &mut TestAppContext,
5130 cx_c: &mut TestAppContext,
5131 ) {
5132 cx_a.foreground().forbid_parking();
5133
5134 // Connect to a server as 3 clients.
5135 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
5136 let mut client_a = server.create_client(cx_a, "user_a").await;
5137 let mut client_b = server.create_client(cx_b, "user_b").await;
5138 let client_c = server.create_client(cx_c, "user_c").await;
5139 server
5140 .make_contacts(vec![
5141 (&client_a, cx_a),
5142 (&client_b, cx_b),
5143 (&client_c, cx_c),
5144 ])
5145 .await;
5146
5147 deterministic.run_until_parked();
5148 for (client, cx) in [(&client_a, &cx_a), (&client_b, &cx_b), (&client_c, &cx_c)] {
5149 client.user_store.read_with(*cx, |store, _| {
5150 assert_eq!(
5151 contacts(store),
5152 [
5153 ("user_a", true, vec![]),
5154 ("user_b", true, vec![]),
5155 ("user_c", true, vec![])
5156 ],
5157 "{} has the wrong contacts",
5158 client.username
5159 )
5160 });
5161 }
5162
5163 // Share a project as client A.
5164 let fs = FakeFs::new(cx_a.background());
5165 fs.create_dir(Path::new("/a")).await.unwrap();
5166 let (project_a, _) = client_a.build_local_project(fs, "/a", cx_a).await;
5167
5168 deterministic.run_until_parked();
5169 for (client, cx) in [(&client_a, &cx_a), (&client_b, &cx_b), (&client_c, &cx_c)] {
5170 client.user_store.read_with(*cx, |store, _| {
5171 assert_eq!(
5172 contacts(store),
5173 [
5174 ("user_a", true, vec![("a", vec![])]),
5175 ("user_b", true, vec![]),
5176 ("user_c", true, vec![])
5177 ],
5178 "{} has the wrong contacts",
5179 client.username
5180 )
5181 });
5182 }
5183
5184 let _project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
5185
5186 deterministic.run_until_parked();
5187 for (client, cx) in [(&client_a, &cx_a), (&client_b, &cx_b), (&client_c, &cx_c)] {
5188 client.user_store.read_with(*cx, |store, _| {
5189 assert_eq!(
5190 contacts(store),
5191 [
5192 ("user_a", true, vec![("a", vec!["user_b"])]),
5193 ("user_b", true, vec![]),
5194 ("user_c", true, vec![])
5195 ],
5196 "{} has the wrong contacts",
5197 client.username
5198 )
5199 });
5200 }
5201
5202 // Add a local project as client B
5203 let fs = FakeFs::new(cx_b.background());
5204 fs.create_dir(Path::new("/b")).await.unwrap();
5205 let (_project_b, _) = client_b.build_local_project(fs, "/b", cx_a).await;
5206
5207 deterministic.run_until_parked();
5208 for (client, cx) in [(&client_a, &cx_a), (&client_b, &cx_b), (&client_c, &cx_c)] {
5209 client.user_store.read_with(*cx, |store, _| {
5210 assert_eq!(
5211 contacts(store),
5212 [
5213 ("user_a", true, vec![("a", vec!["user_b"])]),
5214 ("user_b", true, vec![("b", vec![])]),
5215 ("user_c", true, vec![])
5216 ],
5217 "{} has the wrong contacts",
5218 client.username
5219 )
5220 });
5221 }
5222
5223 project_a
5224 .condition(&cx_a, |project, _| {
5225 project.collaborators().contains_key(&client_b.peer_id)
5226 })
5227 .await;
5228
5229 client_a.project.take();
5230 cx_a.update(move |_| drop(project_a));
5231 deterministic.run_until_parked();
5232 for (client, cx) in [(&client_a, &cx_a), (&client_b, &cx_b), (&client_c, &cx_c)] {
5233 client.user_store.read_with(*cx, |store, _| {
5234 assert_eq!(
5235 contacts(store),
5236 [
5237 ("user_a", true, vec![]),
5238 ("user_b", true, vec![("b", vec![])]),
5239 ("user_c", true, vec![])
5240 ],
5241 "{} has the wrong contacts",
5242 client.username
5243 )
5244 });
5245 }
5246
5247 server.disconnect_client(client_c.current_user_id(cx_c));
5248 server.forbid_connections();
5249 deterministic.advance_clock(rpc::RECEIVE_TIMEOUT);
5250 for (client, cx) in [(&client_a, &cx_a), (&client_b, &cx_b)] {
5251 client.user_store.read_with(*cx, |store, _| {
5252 assert_eq!(
5253 contacts(store),
5254 [
5255 ("user_a", true, vec![]),
5256 ("user_b", true, vec![("b", vec![])]),
5257 ("user_c", false, vec![])
5258 ],
5259 "{} has the wrong contacts",
5260 client.username
5261 )
5262 });
5263 }
5264 client_c
5265 .user_store
5266 .read_with(cx_c, |store, _| assert_eq!(contacts(store), []));
5267
5268 server.allow_connections();
5269 client_c
5270 .authenticate_and_connect(false, &cx_c.to_async())
5271 .await
5272 .unwrap();
5273
5274 deterministic.run_until_parked();
5275 for (client, cx) in [(&client_a, &cx_a), (&client_b, &cx_b), (&client_c, &cx_c)] {
5276 client.user_store.read_with(*cx, |store, _| {
5277 assert_eq!(
5278 contacts(store),
5279 [
5280 ("user_a", true, vec![]),
5281 ("user_b", true, vec![("b", vec![])]),
5282 ("user_c", true, vec![])
5283 ],
5284 "{} has the wrong contacts",
5285 client.username
5286 )
5287 });
5288 }
5289
5290 fn contacts(user_store: &UserStore) -> Vec<(&str, bool, Vec<(&str, Vec<&str>)>)> {
5291 user_store
5292 .contacts()
5293 .iter()
5294 .map(|contact| {
5295 let projects = contact
5296 .projects
5297 .iter()
5298 .map(|p| {
5299 (
5300 p.worktree_root_names[0].as_str(),
5301 p.guests.iter().map(|p| p.github_login.as_str()).collect(),
5302 )
5303 })
5304 .collect();
5305 (contact.user.github_login.as_str(), contact.online, projects)
5306 })
5307 .collect()
5308 }
5309 }
5310
5311 #[gpui::test(iterations = 10)]
5312 async fn test_contact_requests(
5313 executor: Arc<Deterministic>,
5314 cx_a: &mut TestAppContext,
5315 cx_a2: &mut TestAppContext,
5316 cx_b: &mut TestAppContext,
5317 cx_b2: &mut TestAppContext,
5318 cx_c: &mut TestAppContext,
5319 cx_c2: &mut TestAppContext,
5320 ) {
5321 cx_a.foreground().forbid_parking();
5322
5323 // Connect to a server as 3 clients.
5324 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
5325 let client_a = server.create_client(cx_a, "user_a").await;
5326 let client_a2 = server.create_client(cx_a2, "user_a").await;
5327 let client_b = server.create_client(cx_b, "user_b").await;
5328 let client_b2 = server.create_client(cx_b2, "user_b").await;
5329 let client_c = server.create_client(cx_c, "user_c").await;
5330 let client_c2 = server.create_client(cx_c2, "user_c").await;
5331
5332 assert_eq!(client_a.user_id().unwrap(), client_a2.user_id().unwrap());
5333 assert_eq!(client_b.user_id().unwrap(), client_b2.user_id().unwrap());
5334 assert_eq!(client_c.user_id().unwrap(), client_c2.user_id().unwrap());
5335
5336 // User A and User C request that user B become their contact.
5337 client_a
5338 .user_store
5339 .update(cx_a, |store, cx| {
5340 store.request_contact(client_b.user_id().unwrap(), cx)
5341 })
5342 .await
5343 .unwrap();
5344 client_c
5345 .user_store
5346 .update(cx_c, |store, cx| {
5347 store.request_contact(client_b.user_id().unwrap(), cx)
5348 })
5349 .await
5350 .unwrap();
5351 executor.run_until_parked();
5352
5353 // All users see the pending request appear in all their clients.
5354 assert_eq!(
5355 client_a.summarize_contacts(&cx_a).outgoing_requests,
5356 &["user_b"]
5357 );
5358 assert_eq!(
5359 client_a2.summarize_contacts(&cx_a2).outgoing_requests,
5360 &["user_b"]
5361 );
5362 assert_eq!(
5363 client_b.summarize_contacts(&cx_b).incoming_requests,
5364 &["user_a", "user_c"]
5365 );
5366 assert_eq!(
5367 client_b2.summarize_contacts(&cx_b2).incoming_requests,
5368 &["user_a", "user_c"]
5369 );
5370 assert_eq!(
5371 client_c.summarize_contacts(&cx_c).outgoing_requests,
5372 &["user_b"]
5373 );
5374 assert_eq!(
5375 client_c2.summarize_contacts(&cx_c2).outgoing_requests,
5376 &["user_b"]
5377 );
5378
5379 // Contact requests are present upon connecting (tested here via disconnect/reconnect)
5380 disconnect_and_reconnect(&client_a, cx_a).await;
5381 disconnect_and_reconnect(&client_b, cx_b).await;
5382 disconnect_and_reconnect(&client_c, cx_c).await;
5383 executor.run_until_parked();
5384 assert_eq!(
5385 client_a.summarize_contacts(&cx_a).outgoing_requests,
5386 &["user_b"]
5387 );
5388 assert_eq!(
5389 client_b.summarize_contacts(&cx_b).incoming_requests,
5390 &["user_a", "user_c"]
5391 );
5392 assert_eq!(
5393 client_c.summarize_contacts(&cx_c).outgoing_requests,
5394 &["user_b"]
5395 );
5396
5397 // User B accepts the request from user A.
5398 client_b
5399 .user_store
5400 .update(cx_b, |store, cx| {
5401 store.respond_to_contact_request(client_a.user_id().unwrap(), true, cx)
5402 })
5403 .await
5404 .unwrap();
5405
5406 executor.run_until_parked();
5407
5408 // User B sees user A as their contact now in all client, and the incoming request from them is removed.
5409 let contacts_b = client_b.summarize_contacts(&cx_b);
5410 assert_eq!(contacts_b.current, &["user_a", "user_b"]);
5411 assert_eq!(contacts_b.incoming_requests, &["user_c"]);
5412 let contacts_b2 = client_b2.summarize_contacts(&cx_b2);
5413 assert_eq!(contacts_b2.current, &["user_a", "user_b"]);
5414 assert_eq!(contacts_b2.incoming_requests, &["user_c"]);
5415
5416 // User A sees user B as their contact now in all clients, and the outgoing request to them is removed.
5417 let contacts_a = client_a.summarize_contacts(&cx_a);
5418 assert_eq!(contacts_a.current, &["user_a", "user_b"]);
5419 assert!(contacts_a.outgoing_requests.is_empty());
5420 let contacts_a2 = client_a2.summarize_contacts(&cx_a2);
5421 assert_eq!(contacts_a2.current, &["user_a", "user_b"]);
5422 assert!(contacts_a2.outgoing_requests.is_empty());
5423
5424 // Contacts are present upon connecting (tested here via disconnect/reconnect)
5425 disconnect_and_reconnect(&client_a, cx_a).await;
5426 disconnect_and_reconnect(&client_b, cx_b).await;
5427 disconnect_and_reconnect(&client_c, cx_c).await;
5428 executor.run_until_parked();
5429 assert_eq!(
5430 client_a.summarize_contacts(&cx_a).current,
5431 &["user_a", "user_b"]
5432 );
5433 assert_eq!(
5434 client_b.summarize_contacts(&cx_b).current,
5435 &["user_a", "user_b"]
5436 );
5437 assert_eq!(
5438 client_b.summarize_contacts(&cx_b).incoming_requests,
5439 &["user_c"]
5440 );
5441 assert_eq!(client_c.summarize_contacts(&cx_c).current, &["user_c"]);
5442 assert_eq!(
5443 client_c.summarize_contacts(&cx_c).outgoing_requests,
5444 &["user_b"]
5445 );
5446
5447 // User B rejects the request from user C.
5448 client_b
5449 .user_store
5450 .update(cx_b, |store, cx| {
5451 store.respond_to_contact_request(client_c.user_id().unwrap(), false, cx)
5452 })
5453 .await
5454 .unwrap();
5455
5456 executor.run_until_parked();
5457
5458 // User B doesn't see user C as their contact, and the incoming request from them is removed.
5459 let contacts_b = client_b.summarize_contacts(&cx_b);
5460 assert_eq!(contacts_b.current, &["user_a", "user_b"]);
5461 assert!(contacts_b.incoming_requests.is_empty());
5462 let contacts_b2 = client_b2.summarize_contacts(&cx_b2);
5463 assert_eq!(contacts_b2.current, &["user_a", "user_b"]);
5464 assert!(contacts_b2.incoming_requests.is_empty());
5465
5466 // User C doesn't see user B as their contact, and the outgoing request to them is removed.
5467 let contacts_c = client_c.summarize_contacts(&cx_c);
5468 assert_eq!(contacts_c.current, &["user_c"]);
5469 assert!(contacts_c.outgoing_requests.is_empty());
5470 let contacts_c2 = client_c2.summarize_contacts(&cx_c2);
5471 assert_eq!(contacts_c2.current, &["user_c"]);
5472 assert!(contacts_c2.outgoing_requests.is_empty());
5473
5474 // Incoming/outgoing requests are not present upon connecting (tested here via disconnect/reconnect)
5475 disconnect_and_reconnect(&client_a, cx_a).await;
5476 disconnect_and_reconnect(&client_b, cx_b).await;
5477 disconnect_and_reconnect(&client_c, cx_c).await;
5478 executor.run_until_parked();
5479 assert_eq!(
5480 client_a.summarize_contacts(&cx_a).current,
5481 &["user_a", "user_b"]
5482 );
5483 assert_eq!(
5484 client_b.summarize_contacts(&cx_b).current,
5485 &["user_a", "user_b"]
5486 );
5487 assert!(client_b
5488 .summarize_contacts(&cx_b)
5489 .incoming_requests
5490 .is_empty());
5491 assert_eq!(client_c.summarize_contacts(&cx_c).current, &["user_c"]);
5492 assert!(client_c
5493 .summarize_contacts(&cx_c)
5494 .outgoing_requests
5495 .is_empty());
5496
5497 async fn disconnect_and_reconnect(client: &TestClient, cx: &mut TestAppContext) {
5498 client.disconnect(&cx.to_async()).unwrap();
5499 client.clear_contacts(cx).await;
5500 client
5501 .authenticate_and_connect(false, &cx.to_async())
5502 .await
5503 .unwrap();
5504 }
5505 }
5506
5507 #[gpui::test(iterations = 10)]
5508 async fn test_following(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
5509 cx_a.foreground().forbid_parking();
5510 let fs = FakeFs::new(cx_a.background());
5511
5512 // 2 clients connect to a server.
5513 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
5514 let mut client_a = server.create_client(cx_a, "user_a").await;
5515 let mut client_b = server.create_client(cx_b, "user_b").await;
5516 server
5517 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
5518 .await;
5519 cx_a.update(editor::init);
5520 cx_b.update(editor::init);
5521
5522 // Client A shares a project.
5523 fs.insert_tree(
5524 "/a",
5525 json!({
5526 "1.txt": "one",
5527 "2.txt": "two",
5528 "3.txt": "three",
5529 }),
5530 )
5531 .await;
5532 let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
5533
5534 // Client B joins the project.
5535 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
5536
5537 // Client A opens some editors.
5538 let workspace_a = client_a.build_workspace(&project_a, cx_a);
5539 let pane_a = workspace_a.read_with(cx_a, |workspace, _| workspace.active_pane().clone());
5540 let editor_a1 = workspace_a
5541 .update(cx_a, |workspace, cx| {
5542 workspace.open_path((worktree_id, "1.txt"), true, cx)
5543 })
5544 .await
5545 .unwrap()
5546 .downcast::<Editor>()
5547 .unwrap();
5548 let editor_a2 = workspace_a
5549 .update(cx_a, |workspace, cx| {
5550 workspace.open_path((worktree_id, "2.txt"), true, cx)
5551 })
5552 .await
5553 .unwrap()
5554 .downcast::<Editor>()
5555 .unwrap();
5556
5557 // Client B opens an editor.
5558 let workspace_b = client_b.build_workspace(&project_b, cx_b);
5559 let editor_b1 = workspace_b
5560 .update(cx_b, |workspace, cx| {
5561 workspace.open_path((worktree_id, "1.txt"), true, cx)
5562 })
5563 .await
5564 .unwrap()
5565 .downcast::<Editor>()
5566 .unwrap();
5567
5568 let client_a_id = project_b.read_with(cx_b, |project, _| {
5569 project.collaborators().values().next().unwrap().peer_id
5570 });
5571 let client_b_id = project_a.read_with(cx_a, |project, _| {
5572 project.collaborators().values().next().unwrap().peer_id
5573 });
5574
5575 // When client B starts following client A, all visible view states are replicated to client B.
5576 editor_a1.update(cx_a, |editor, cx| {
5577 editor.change_selections(None, cx, |s| s.select_ranges([0..1]))
5578 });
5579 editor_a2.update(cx_a, |editor, cx| {
5580 editor.change_selections(None, cx, |s| s.select_ranges([2..3]))
5581 });
5582 workspace_b
5583 .update(cx_b, |workspace, cx| {
5584 workspace
5585 .toggle_follow(&ToggleFollow(client_a_id), cx)
5586 .unwrap()
5587 })
5588 .await
5589 .unwrap();
5590
5591 let editor_b2 = workspace_b.read_with(cx_b, |workspace, cx| {
5592 workspace
5593 .active_item(cx)
5594 .unwrap()
5595 .downcast::<Editor>()
5596 .unwrap()
5597 });
5598 assert!(cx_b.read(|cx| editor_b2.is_focused(cx)));
5599 assert_eq!(
5600 editor_b2.read_with(cx_b, |editor, cx| editor.project_path(cx)),
5601 Some((worktree_id, "2.txt").into())
5602 );
5603 assert_eq!(
5604 editor_b2.read_with(cx_b, |editor, cx| editor.selections.ranges(cx)),
5605 vec![2..3]
5606 );
5607 assert_eq!(
5608 editor_b1.read_with(cx_b, |editor, cx| editor.selections.ranges(cx)),
5609 vec![0..1]
5610 );
5611
5612 // When client A activates a different editor, client B does so as well.
5613 workspace_a.update(cx_a, |workspace, cx| {
5614 workspace.activate_item(&editor_a1, cx)
5615 });
5616 workspace_b
5617 .condition(cx_b, |workspace, cx| {
5618 workspace.active_item(cx).unwrap().id() == editor_b1.id()
5619 })
5620 .await;
5621
5622 // When client A navigates back and forth, client B does so as well.
5623 workspace_a
5624 .update(cx_a, |workspace, cx| {
5625 workspace::Pane::go_back(workspace, None, cx)
5626 })
5627 .await;
5628 workspace_b
5629 .condition(cx_b, |workspace, cx| {
5630 workspace.active_item(cx).unwrap().id() == editor_b2.id()
5631 })
5632 .await;
5633
5634 workspace_a
5635 .update(cx_a, |workspace, cx| {
5636 workspace::Pane::go_forward(workspace, None, cx)
5637 })
5638 .await;
5639 workspace_b
5640 .condition(cx_b, |workspace, cx| {
5641 workspace.active_item(cx).unwrap().id() == editor_b1.id()
5642 })
5643 .await;
5644
5645 // Changes to client A's editor are reflected on client B.
5646 editor_a1.update(cx_a, |editor, cx| {
5647 editor.change_selections(None, cx, |s| s.select_ranges([1..1, 2..2]));
5648 });
5649 editor_b1
5650 .condition(cx_b, |editor, cx| {
5651 editor.selections.ranges(cx) == vec![1..1, 2..2]
5652 })
5653 .await;
5654
5655 editor_a1.update(cx_a, |editor, cx| editor.set_text("TWO", cx));
5656 editor_b1
5657 .condition(cx_b, |editor, cx| editor.text(cx) == "TWO")
5658 .await;
5659
5660 editor_a1.update(cx_a, |editor, cx| {
5661 editor.change_selections(None, cx, |s| s.select_ranges([3..3]));
5662 editor.set_scroll_position(vec2f(0., 100.), cx);
5663 });
5664 editor_b1
5665 .condition(cx_b, |editor, cx| {
5666 editor.selections.ranges(cx) == vec![3..3]
5667 })
5668 .await;
5669
5670 // After unfollowing, client B stops receiving updates from client A.
5671 workspace_b.update(cx_b, |workspace, cx| {
5672 workspace.unfollow(&workspace.active_pane().clone(), cx)
5673 });
5674 workspace_a.update(cx_a, |workspace, cx| {
5675 workspace.activate_item(&editor_a2, cx)
5676 });
5677 cx_a.foreground().run_until_parked();
5678 assert_eq!(
5679 workspace_b.read_with(cx_b, |workspace, cx| workspace
5680 .active_item(cx)
5681 .unwrap()
5682 .id()),
5683 editor_b1.id()
5684 );
5685
5686 // Client A starts following client B.
5687 workspace_a
5688 .update(cx_a, |workspace, cx| {
5689 workspace
5690 .toggle_follow(&ToggleFollow(client_b_id), cx)
5691 .unwrap()
5692 })
5693 .await
5694 .unwrap();
5695 assert_eq!(
5696 workspace_a.read_with(cx_a, |workspace, _| workspace.leader_for_pane(&pane_a)),
5697 Some(client_b_id)
5698 );
5699 assert_eq!(
5700 workspace_a.read_with(cx_a, |workspace, cx| workspace
5701 .active_item(cx)
5702 .unwrap()
5703 .id()),
5704 editor_a1.id()
5705 );
5706
5707 // Following interrupts when client B disconnects.
5708 client_b.disconnect(&cx_b.to_async()).unwrap();
5709 cx_a.foreground().run_until_parked();
5710 assert_eq!(
5711 workspace_a.read_with(cx_a, |workspace, _| workspace.leader_for_pane(&pane_a)),
5712 None
5713 );
5714 }
5715
5716 #[gpui::test(iterations = 10)]
5717 async fn test_peers_following_each_other(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
5718 cx_a.foreground().forbid_parking();
5719 let fs = FakeFs::new(cx_a.background());
5720
5721 // 2 clients connect to a server.
5722 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
5723 let mut client_a = server.create_client(cx_a, "user_a").await;
5724 let mut client_b = server.create_client(cx_b, "user_b").await;
5725 server
5726 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
5727 .await;
5728 cx_a.update(editor::init);
5729 cx_b.update(editor::init);
5730
5731 // Client A shares a project.
5732 fs.insert_tree(
5733 "/a",
5734 json!({
5735 "1.txt": "one",
5736 "2.txt": "two",
5737 "3.txt": "three",
5738 "4.txt": "four",
5739 }),
5740 )
5741 .await;
5742 let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
5743
5744 // Client B joins the project.
5745 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
5746
5747 // Client A opens some editors.
5748 let workspace_a = client_a.build_workspace(&project_a, cx_a);
5749 let pane_a1 = workspace_a.read_with(cx_a, |workspace, _| workspace.active_pane().clone());
5750 let _editor_a1 = workspace_a
5751 .update(cx_a, |workspace, cx| {
5752 workspace.open_path((worktree_id, "1.txt"), true, cx)
5753 })
5754 .await
5755 .unwrap()
5756 .downcast::<Editor>()
5757 .unwrap();
5758
5759 // Client B opens an editor.
5760 let workspace_b = client_b.build_workspace(&project_b, cx_b);
5761 let pane_b1 = workspace_b.read_with(cx_b, |workspace, _| workspace.active_pane().clone());
5762 let _editor_b1 = workspace_b
5763 .update(cx_b, |workspace, cx| {
5764 workspace.open_path((worktree_id, "2.txt"), true, cx)
5765 })
5766 .await
5767 .unwrap()
5768 .downcast::<Editor>()
5769 .unwrap();
5770
5771 // Clients A and B follow each other in split panes
5772 workspace_a
5773 .update(cx_a, |workspace, cx| {
5774 workspace.split_pane(workspace.active_pane().clone(), SplitDirection::Right, cx);
5775 assert_ne!(*workspace.active_pane(), pane_a1);
5776 let leader_id = *project_a.read(cx).collaborators().keys().next().unwrap();
5777 workspace
5778 .toggle_follow(&workspace::ToggleFollow(leader_id), cx)
5779 .unwrap()
5780 })
5781 .await
5782 .unwrap();
5783 workspace_b
5784 .update(cx_b, |workspace, cx| {
5785 workspace.split_pane(workspace.active_pane().clone(), SplitDirection::Right, cx);
5786 assert_ne!(*workspace.active_pane(), pane_b1);
5787 let leader_id = *project_b.read(cx).collaborators().keys().next().unwrap();
5788 workspace
5789 .toggle_follow(&workspace::ToggleFollow(leader_id), cx)
5790 .unwrap()
5791 })
5792 .await
5793 .unwrap();
5794
5795 workspace_a
5796 .update(cx_a, |workspace, cx| {
5797 workspace.activate_next_pane(cx);
5798 assert_eq!(*workspace.active_pane(), pane_a1);
5799 workspace.open_path((worktree_id, "3.txt"), true, cx)
5800 })
5801 .await
5802 .unwrap();
5803 workspace_b
5804 .update(cx_b, |workspace, cx| {
5805 workspace.activate_next_pane(cx);
5806 assert_eq!(*workspace.active_pane(), pane_b1);
5807 workspace.open_path((worktree_id, "4.txt"), true, cx)
5808 })
5809 .await
5810 .unwrap();
5811 cx_a.foreground().run_until_parked();
5812
5813 // Ensure leader updates don't change the active pane of followers
5814 workspace_a.read_with(cx_a, |workspace, _| {
5815 assert_eq!(*workspace.active_pane(), pane_a1);
5816 });
5817 workspace_b.read_with(cx_b, |workspace, _| {
5818 assert_eq!(*workspace.active_pane(), pane_b1);
5819 });
5820
5821 // Ensure peers following each other doesn't cause an infinite loop.
5822 assert_eq!(
5823 workspace_a.read_with(cx_a, |workspace, cx| workspace
5824 .active_item(cx)
5825 .unwrap()
5826 .project_path(cx)),
5827 Some((worktree_id, "3.txt").into())
5828 );
5829 workspace_a.update(cx_a, |workspace, cx| {
5830 assert_eq!(
5831 workspace.active_item(cx).unwrap().project_path(cx),
5832 Some((worktree_id, "3.txt").into())
5833 );
5834 workspace.activate_next_pane(cx);
5835 assert_eq!(
5836 workspace.active_item(cx).unwrap().project_path(cx),
5837 Some((worktree_id, "4.txt").into())
5838 );
5839 });
5840 workspace_b.update(cx_b, |workspace, cx| {
5841 assert_eq!(
5842 workspace.active_item(cx).unwrap().project_path(cx),
5843 Some((worktree_id, "4.txt").into())
5844 );
5845 workspace.activate_next_pane(cx);
5846 assert_eq!(
5847 workspace.active_item(cx).unwrap().project_path(cx),
5848 Some((worktree_id, "3.txt").into())
5849 );
5850 });
5851 }
5852
5853 #[gpui::test(iterations = 10)]
5854 async fn test_auto_unfollowing(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
5855 cx_a.foreground().forbid_parking();
5856 let fs = FakeFs::new(cx_a.background());
5857
5858 // 2 clients connect to a server.
5859 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
5860 let mut client_a = server.create_client(cx_a, "user_a").await;
5861 let mut client_b = server.create_client(cx_b, "user_b").await;
5862 server
5863 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
5864 .await;
5865 cx_a.update(editor::init);
5866 cx_b.update(editor::init);
5867
5868 // Client A shares a project.
5869 fs.insert_tree(
5870 "/a",
5871 json!({
5872 "1.txt": "one",
5873 "2.txt": "two",
5874 "3.txt": "three",
5875 }),
5876 )
5877 .await;
5878 let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
5879
5880 // Client B joins the project.
5881 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
5882
5883 // Client A opens some editors.
5884 let workspace_a = client_a.build_workspace(&project_a, cx_a);
5885 let _editor_a1 = workspace_a
5886 .update(cx_a, |workspace, cx| {
5887 workspace.open_path((worktree_id, "1.txt"), true, cx)
5888 })
5889 .await
5890 .unwrap()
5891 .downcast::<Editor>()
5892 .unwrap();
5893
5894 // Client B starts following client A.
5895 let workspace_b = client_b.build_workspace(&project_b, cx_b);
5896 let pane_b = workspace_b.read_with(cx_b, |workspace, _| workspace.active_pane().clone());
5897 let leader_id = project_b.read_with(cx_b, |project, _| {
5898 project.collaborators().values().next().unwrap().peer_id
5899 });
5900 workspace_b
5901 .update(cx_b, |workspace, cx| {
5902 workspace
5903 .toggle_follow(&ToggleFollow(leader_id), cx)
5904 .unwrap()
5905 })
5906 .await
5907 .unwrap();
5908 assert_eq!(
5909 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5910 Some(leader_id)
5911 );
5912 let editor_b2 = workspace_b.read_with(cx_b, |workspace, cx| {
5913 workspace
5914 .active_item(cx)
5915 .unwrap()
5916 .downcast::<Editor>()
5917 .unwrap()
5918 });
5919
5920 // When client B moves, it automatically stops following client A.
5921 editor_b2.update(cx_b, |editor, cx| editor.move_right(&editor::MoveRight, cx));
5922 assert_eq!(
5923 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5924 None
5925 );
5926
5927 workspace_b
5928 .update(cx_b, |workspace, cx| {
5929 workspace
5930 .toggle_follow(&ToggleFollow(leader_id), cx)
5931 .unwrap()
5932 })
5933 .await
5934 .unwrap();
5935 assert_eq!(
5936 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5937 Some(leader_id)
5938 );
5939
5940 // When client B edits, it automatically stops following client A.
5941 editor_b2.update(cx_b, |editor, cx| editor.insert("X", cx));
5942 assert_eq!(
5943 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5944 None
5945 );
5946
5947 workspace_b
5948 .update(cx_b, |workspace, cx| {
5949 workspace
5950 .toggle_follow(&ToggleFollow(leader_id), cx)
5951 .unwrap()
5952 })
5953 .await
5954 .unwrap();
5955 assert_eq!(
5956 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5957 Some(leader_id)
5958 );
5959
5960 // When client B scrolls, it automatically stops following client A.
5961 editor_b2.update(cx_b, |editor, cx| {
5962 editor.set_scroll_position(vec2f(0., 3.), cx)
5963 });
5964 assert_eq!(
5965 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5966 None
5967 );
5968
5969 workspace_b
5970 .update(cx_b, |workspace, cx| {
5971 workspace
5972 .toggle_follow(&ToggleFollow(leader_id), cx)
5973 .unwrap()
5974 })
5975 .await
5976 .unwrap();
5977 assert_eq!(
5978 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5979 Some(leader_id)
5980 );
5981
5982 // When client B activates a different pane, it continues following client A in the original pane.
5983 workspace_b.update(cx_b, |workspace, cx| {
5984 workspace.split_pane(pane_b.clone(), SplitDirection::Right, cx)
5985 });
5986 assert_eq!(
5987 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5988 Some(leader_id)
5989 );
5990
5991 workspace_b.update(cx_b, |workspace, cx| workspace.activate_next_pane(cx));
5992 assert_eq!(
5993 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5994 Some(leader_id)
5995 );
5996
5997 // When client B activates a different item in the original pane, it automatically stops following client A.
5998 workspace_b
5999 .update(cx_b, |workspace, cx| {
6000 workspace.open_path((worktree_id, "2.txt"), true, cx)
6001 })
6002 .await
6003 .unwrap();
6004 assert_eq!(
6005 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
6006 None
6007 );
6008 }
6009
6010 #[gpui::test(iterations = 100)]
6011 async fn test_random_collaboration(
6012 cx: &mut TestAppContext,
6013 deterministic: Arc<Deterministic>,
6014 rng: StdRng,
6015 ) {
6016 cx.foreground().forbid_parking();
6017 let max_peers = env::var("MAX_PEERS")
6018 .map(|i| i.parse().expect("invalid `MAX_PEERS` variable"))
6019 .unwrap_or(5);
6020 assert!(max_peers <= 5);
6021
6022 let max_operations = env::var("OPERATIONS")
6023 .map(|i| i.parse().expect("invalid `OPERATIONS` variable"))
6024 .unwrap_or(10);
6025
6026 let rng = Arc::new(Mutex::new(rng));
6027
6028 let guest_lang_registry = Arc::new(LanguageRegistry::test());
6029 let host_language_registry = Arc::new(LanguageRegistry::test());
6030
6031 let fs = FakeFs::new(cx.background());
6032 fs.insert_tree("/_collab", json!({"init": ""})).await;
6033
6034 let mut server = TestServer::start(cx.foreground(), cx.background()).await;
6035 let db = server.app_state.db.clone();
6036 let host_user_id = db.create_user("host", false).await.unwrap();
6037 for username in ["guest-1", "guest-2", "guest-3", "guest-4"] {
6038 let guest_user_id = db.create_user(username, false).await.unwrap();
6039 server
6040 .app_state
6041 .db
6042 .send_contact_request(guest_user_id, host_user_id)
6043 .await
6044 .unwrap();
6045 server
6046 .app_state
6047 .db
6048 .respond_to_contact_request(host_user_id, guest_user_id, true)
6049 .await
6050 .unwrap();
6051 }
6052
6053 let mut clients = Vec::new();
6054 let mut user_ids = Vec::new();
6055 let mut op_start_signals = Vec::new();
6056
6057 let mut next_entity_id = 100000;
6058 let mut host_cx = TestAppContext::new(
6059 cx.foreground_platform(),
6060 cx.platform(),
6061 deterministic.build_foreground(next_entity_id),
6062 deterministic.build_background(),
6063 cx.font_cache(),
6064 cx.leak_detector(),
6065 next_entity_id,
6066 );
6067 let host = server.create_client(&mut host_cx, "host").await;
6068 let host_project = host_cx.update(|cx| {
6069 Project::local(
6070 host.client.clone(),
6071 host.user_store.clone(),
6072 host_language_registry.clone(),
6073 fs.clone(),
6074 cx,
6075 )
6076 });
6077 let host_project_id = host_project
6078 .update(&mut host_cx, |p, _| p.next_remote_id())
6079 .await;
6080
6081 let (collab_worktree, _) = host_project
6082 .update(&mut host_cx, |project, cx| {
6083 project.find_or_create_local_worktree("/_collab", true, cx)
6084 })
6085 .await
6086 .unwrap();
6087 collab_worktree
6088 .read_with(&host_cx, |tree, _| tree.as_local().unwrap().scan_complete())
6089 .await;
6090
6091 // Set up fake language servers.
6092 let mut language = Language::new(
6093 LanguageConfig {
6094 name: "Rust".into(),
6095 path_suffixes: vec!["rs".to_string()],
6096 ..Default::default()
6097 },
6098 None,
6099 );
6100 let _fake_servers = language.set_fake_lsp_adapter(FakeLspAdapter {
6101 name: "the-fake-language-server",
6102 capabilities: lsp::LanguageServer::full_capabilities(),
6103 initializer: Some(Box::new({
6104 let rng = rng.clone();
6105 let fs = fs.clone();
6106 let project = host_project.downgrade();
6107 move |fake_server: &mut FakeLanguageServer| {
6108 fake_server.handle_request::<lsp::request::Completion, _, _>(
6109 |_, _| async move {
6110 Ok(Some(lsp::CompletionResponse::Array(vec![
6111 lsp::CompletionItem {
6112 text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
6113 range: lsp::Range::new(
6114 lsp::Position::new(0, 0),
6115 lsp::Position::new(0, 0),
6116 ),
6117 new_text: "the-new-text".to_string(),
6118 })),
6119 ..Default::default()
6120 },
6121 ])))
6122 },
6123 );
6124
6125 fake_server.handle_request::<lsp::request::CodeActionRequest, _, _>(
6126 |_, _| async move {
6127 Ok(Some(vec![lsp::CodeActionOrCommand::CodeAction(
6128 lsp::CodeAction {
6129 title: "the-code-action".to_string(),
6130 ..Default::default()
6131 },
6132 )]))
6133 },
6134 );
6135
6136 fake_server.handle_request::<lsp::request::PrepareRenameRequest, _, _>(
6137 |params, _| async move {
6138 Ok(Some(lsp::PrepareRenameResponse::Range(lsp::Range::new(
6139 params.position,
6140 params.position,
6141 ))))
6142 },
6143 );
6144
6145 fake_server.handle_request::<lsp::request::GotoDefinition, _, _>({
6146 let fs = fs.clone();
6147 let rng = rng.clone();
6148 move |_, _| {
6149 let fs = fs.clone();
6150 let rng = rng.clone();
6151 async move {
6152 let files = fs.files().await;
6153 let mut rng = rng.lock();
6154 let count = rng.gen_range::<usize, _>(1..3);
6155 let files = (0..count)
6156 .map(|_| files.choose(&mut *rng).unwrap())
6157 .collect::<Vec<_>>();
6158 log::info!("LSP: Returning definitions in files {:?}", &files);
6159 Ok(Some(lsp::GotoDefinitionResponse::Array(
6160 files
6161 .into_iter()
6162 .map(|file| lsp::Location {
6163 uri: lsp::Url::from_file_path(file).unwrap(),
6164 range: Default::default(),
6165 })
6166 .collect(),
6167 )))
6168 }
6169 }
6170 });
6171
6172 fake_server.handle_request::<lsp::request::DocumentHighlightRequest, _, _>({
6173 let rng = rng.clone();
6174 let project = project.clone();
6175 move |params, mut cx| {
6176 let highlights = if let Some(project) = project.upgrade(&cx) {
6177 project.update(&mut cx, |project, cx| {
6178 let path = params
6179 .text_document_position_params
6180 .text_document
6181 .uri
6182 .to_file_path()
6183 .unwrap();
6184 let (worktree, relative_path) =
6185 project.find_local_worktree(&path, cx)?;
6186 let project_path =
6187 ProjectPath::from((worktree.read(cx).id(), relative_path));
6188 let buffer =
6189 project.get_open_buffer(&project_path, cx)?.read(cx);
6190
6191 let mut highlights = Vec::new();
6192 let highlight_count = rng.lock().gen_range(1..=5);
6193 let mut prev_end = 0;
6194 for _ in 0..highlight_count {
6195 let range =
6196 buffer.random_byte_range(prev_end, &mut *rng.lock());
6197
6198 highlights.push(lsp::DocumentHighlight {
6199 range: range_to_lsp(range.to_point_utf16(buffer)),
6200 kind: Some(lsp::DocumentHighlightKind::READ),
6201 });
6202 prev_end = range.end;
6203 }
6204 Some(highlights)
6205 })
6206 } else {
6207 None
6208 };
6209 async move { Ok(highlights) }
6210 }
6211 });
6212 }
6213 })),
6214 ..Default::default()
6215 });
6216 host_language_registry.add(Arc::new(language));
6217
6218 let op_start_signal = futures::channel::mpsc::unbounded();
6219 user_ids.push(host.current_user_id(&host_cx));
6220 op_start_signals.push(op_start_signal.0);
6221 clients.push(host_cx.foreground().spawn(host.simulate_host(
6222 host_project,
6223 op_start_signal.1,
6224 rng.clone(),
6225 host_cx,
6226 )));
6227
6228 let disconnect_host_at = if rng.lock().gen_bool(0.2) {
6229 rng.lock().gen_range(0..max_operations)
6230 } else {
6231 max_operations
6232 };
6233 let mut available_guests = vec![
6234 "guest-1".to_string(),
6235 "guest-2".to_string(),
6236 "guest-3".to_string(),
6237 "guest-4".to_string(),
6238 ];
6239 let mut operations = 0;
6240 while operations < max_operations {
6241 if operations == disconnect_host_at {
6242 server.disconnect_client(user_ids[0]);
6243 cx.foreground().advance_clock(RECEIVE_TIMEOUT);
6244 drop(op_start_signals);
6245 let mut clients = futures::future::join_all(clients).await;
6246 cx.foreground().run_until_parked();
6247
6248 let (host, mut host_cx, host_err) = clients.remove(0);
6249 if let Some(host_err) = host_err {
6250 log::error!("host error - {:?}", host_err);
6251 }
6252 host.project
6253 .as_ref()
6254 .unwrap()
6255 .read_with(&host_cx, |project, _| assert!(!project.is_shared()));
6256 for (guest, mut guest_cx, guest_err) in clients {
6257 if let Some(guest_err) = guest_err {
6258 log::error!("{} error - {:?}", guest.username, guest_err);
6259 }
6260
6261 let contacts = server
6262 .app_state
6263 .db
6264 .get_contacts(guest.current_user_id(&guest_cx))
6265 .await
6266 .unwrap();
6267 let contacts = server
6268 .store
6269 .read()
6270 .await
6271 .build_initial_contacts_update(contacts)
6272 .contacts;
6273 assert!(!contacts
6274 .iter()
6275 .flat_map(|contact| &contact.projects)
6276 .any(|project| project.id == host_project_id));
6277 guest
6278 .project
6279 .as_ref()
6280 .unwrap()
6281 .read_with(&guest_cx, |project, _| assert!(project.is_read_only()));
6282 guest_cx.update(|_| drop(guest));
6283 }
6284 host_cx.update(|_| drop(host));
6285
6286 return;
6287 }
6288
6289 let distribution = rng.lock().gen_range(0..100);
6290 match distribution {
6291 0..=19 if !available_guests.is_empty() => {
6292 let guest_ix = rng.lock().gen_range(0..available_guests.len());
6293 let guest_username = available_guests.remove(guest_ix);
6294 log::info!("Adding new connection for {}", guest_username);
6295 next_entity_id += 100000;
6296 let mut guest_cx = TestAppContext::new(
6297 cx.foreground_platform(),
6298 cx.platform(),
6299 deterministic.build_foreground(next_entity_id),
6300 deterministic.build_background(),
6301 cx.font_cache(),
6302 cx.leak_detector(),
6303 next_entity_id,
6304 );
6305 let guest = server.create_client(&mut guest_cx, &guest_username).await;
6306 let guest_project = Project::remote(
6307 host_project_id,
6308 guest.client.clone(),
6309 guest.user_store.clone(),
6310 guest_lang_registry.clone(),
6311 FakeFs::new(cx.background()),
6312 &mut guest_cx.to_async(),
6313 )
6314 .await
6315 .unwrap();
6316 let op_start_signal = futures::channel::mpsc::unbounded();
6317 user_ids.push(guest.current_user_id(&guest_cx));
6318 op_start_signals.push(op_start_signal.0);
6319 clients.push(guest_cx.foreground().spawn(guest.simulate_guest(
6320 guest_username.clone(),
6321 guest_project,
6322 op_start_signal.1,
6323 rng.clone(),
6324 guest_cx,
6325 )));
6326
6327 log::info!("Added connection for {}", guest_username);
6328 operations += 1;
6329 }
6330 20..=29 if clients.len() > 1 => {
6331 let guest_ix = rng.lock().gen_range(1..clients.len());
6332 log::info!("Removing guest {}", user_ids[guest_ix]);
6333 let removed_guest_id = user_ids.remove(guest_ix);
6334 let guest = clients.remove(guest_ix);
6335 op_start_signals.remove(guest_ix);
6336 server.forbid_connections();
6337 server.disconnect_client(removed_guest_id);
6338 cx.foreground().advance_clock(RECEIVE_TIMEOUT);
6339 let (guest, mut guest_cx, guest_err) = guest.await;
6340 server.allow_connections();
6341
6342 if let Some(guest_err) = guest_err {
6343 log::error!("{} error - {:?}", guest.username, guest_err);
6344 }
6345 guest
6346 .project
6347 .as_ref()
6348 .unwrap()
6349 .read_with(&guest_cx, |project, _| assert!(project.is_read_only()));
6350 for user_id in &user_ids {
6351 let contacts = server.app_state.db.get_contacts(*user_id).await.unwrap();
6352 let contacts = server
6353 .store
6354 .read()
6355 .await
6356 .build_initial_contacts_update(contacts)
6357 .contacts;
6358 for contact in contacts {
6359 if contact.online {
6360 assert_ne!(
6361 contact.user_id, removed_guest_id.0 as u64,
6362 "removed guest is still a contact of another peer"
6363 );
6364 }
6365 for project in contact.projects {
6366 for project_guest_id in project.guests {
6367 assert_ne!(
6368 project_guest_id, removed_guest_id.0 as u64,
6369 "removed guest appears as still participating on a project"
6370 );
6371 }
6372 }
6373 }
6374 }
6375
6376 log::info!("{} removed", guest.username);
6377 available_guests.push(guest.username.clone());
6378 guest_cx.update(|_| drop(guest));
6379
6380 operations += 1;
6381 }
6382 _ => {
6383 while operations < max_operations && rng.lock().gen_bool(0.7) {
6384 op_start_signals
6385 .choose(&mut *rng.lock())
6386 .unwrap()
6387 .unbounded_send(())
6388 .unwrap();
6389 operations += 1;
6390 }
6391
6392 if rng.lock().gen_bool(0.8) {
6393 cx.foreground().run_until_parked();
6394 }
6395 }
6396 }
6397 }
6398
6399 drop(op_start_signals);
6400 let mut clients = futures::future::join_all(clients).await;
6401 cx.foreground().run_until_parked();
6402
6403 let (host_client, mut host_cx, host_err) = clients.remove(0);
6404 if let Some(host_err) = host_err {
6405 panic!("host error - {:?}", host_err);
6406 }
6407 let host_project = host_client.project.as_ref().unwrap();
6408 let host_worktree_snapshots = host_project.read_with(&host_cx, |project, cx| {
6409 project
6410 .worktrees(cx)
6411 .map(|worktree| {
6412 let snapshot = worktree.read(cx).snapshot();
6413 (snapshot.id(), snapshot)
6414 })
6415 .collect::<BTreeMap<_, _>>()
6416 });
6417
6418 host_client
6419 .project
6420 .as_ref()
6421 .unwrap()
6422 .read_with(&host_cx, |project, cx| project.check_invariants(cx));
6423
6424 for (guest_client, mut guest_cx, guest_err) in clients.into_iter() {
6425 if let Some(guest_err) = guest_err {
6426 panic!("{} error - {:?}", guest_client.username, guest_err);
6427 }
6428 let worktree_snapshots =
6429 guest_client
6430 .project
6431 .as_ref()
6432 .unwrap()
6433 .read_with(&guest_cx, |project, cx| {
6434 project
6435 .worktrees(cx)
6436 .map(|worktree| {
6437 let worktree = worktree.read(cx);
6438 (worktree.id(), worktree.snapshot())
6439 })
6440 .collect::<BTreeMap<_, _>>()
6441 });
6442
6443 assert_eq!(
6444 worktree_snapshots.keys().collect::<Vec<_>>(),
6445 host_worktree_snapshots.keys().collect::<Vec<_>>(),
6446 "{} has different worktrees than the host",
6447 guest_client.username
6448 );
6449 for (id, host_snapshot) in &host_worktree_snapshots {
6450 let guest_snapshot = &worktree_snapshots[id];
6451 assert_eq!(
6452 guest_snapshot.root_name(),
6453 host_snapshot.root_name(),
6454 "{} has different root name than the host for worktree {}",
6455 guest_client.username,
6456 id
6457 );
6458 assert_eq!(
6459 guest_snapshot.entries(false).collect::<Vec<_>>(),
6460 host_snapshot.entries(false).collect::<Vec<_>>(),
6461 "{} has different snapshot than the host for worktree {}",
6462 guest_client.username,
6463 id
6464 );
6465 assert_eq!(guest_snapshot.scan_id(), host_snapshot.scan_id());
6466 }
6467
6468 guest_client
6469 .project
6470 .as_ref()
6471 .unwrap()
6472 .read_with(&guest_cx, |project, cx| project.check_invariants(cx));
6473
6474 for guest_buffer in &guest_client.buffers {
6475 let buffer_id = guest_buffer.read_with(&guest_cx, |buffer, _| buffer.remote_id());
6476 let host_buffer = host_project.read_with(&host_cx, |project, cx| {
6477 project.buffer_for_id(buffer_id, cx).expect(&format!(
6478 "host does not have buffer for guest:{}, peer:{}, id:{}",
6479 guest_client.username, guest_client.peer_id, buffer_id
6480 ))
6481 });
6482 let path = host_buffer
6483 .read_with(&host_cx, |buffer, cx| buffer.file().unwrap().full_path(cx));
6484
6485 assert_eq!(
6486 guest_buffer.read_with(&guest_cx, |buffer, _| buffer.deferred_ops_len()),
6487 0,
6488 "{}, buffer {}, path {:?} has deferred operations",
6489 guest_client.username,
6490 buffer_id,
6491 path,
6492 );
6493 assert_eq!(
6494 guest_buffer.read_with(&guest_cx, |buffer, _| buffer.text()),
6495 host_buffer.read_with(&host_cx, |buffer, _| buffer.text()),
6496 "{}, buffer {}, path {:?}, differs from the host's buffer",
6497 guest_client.username,
6498 buffer_id,
6499 path
6500 );
6501 }
6502
6503 guest_cx.update(|_| drop(guest_client));
6504 }
6505
6506 host_cx.update(|_| drop(host_client));
6507 }
6508
6509 struct TestServer {
6510 peer: Arc<Peer>,
6511 app_state: Arc<AppState>,
6512 server: Arc<Server>,
6513 foreground: Rc<executor::Foreground>,
6514 notifications: mpsc::UnboundedReceiver<()>,
6515 connection_killers: Arc<Mutex<HashMap<UserId, Arc<AtomicBool>>>>,
6516 forbid_connections: Arc<AtomicBool>,
6517 _test_db: TestDb,
6518 }
6519
6520 impl TestServer {
6521 async fn start(
6522 foreground: Rc<executor::Foreground>,
6523 background: Arc<executor::Background>,
6524 ) -> Self {
6525 let test_db = TestDb::fake(background);
6526 let app_state = Self::build_app_state(&test_db).await;
6527 let peer = Peer::new();
6528 let notifications = mpsc::unbounded();
6529 let server = Server::new(app_state.clone(), Some(notifications.0));
6530 Self {
6531 peer,
6532 app_state,
6533 server,
6534 foreground,
6535 notifications: notifications.1,
6536 connection_killers: Default::default(),
6537 forbid_connections: Default::default(),
6538 _test_db: test_db,
6539 }
6540 }
6541
6542 async fn create_client(&mut self, cx: &mut TestAppContext, name: &str) -> TestClient {
6543 cx.update(|cx| {
6544 let settings = Settings::test(cx);
6545 cx.set_global(settings);
6546 });
6547
6548 let http = FakeHttpClient::with_404_response();
6549 let user_id =
6550 if let Ok(Some(user)) = self.app_state.db.get_user_by_github_login(name).await {
6551 user.id
6552 } else {
6553 self.app_state.db.create_user(name, false).await.unwrap()
6554 };
6555 let client_name = name.to_string();
6556 let mut client = Client::new(http.clone());
6557 let server = self.server.clone();
6558 let db = self.app_state.db.clone();
6559 let connection_killers = self.connection_killers.clone();
6560 let forbid_connections = self.forbid_connections.clone();
6561 let (connection_id_tx, mut connection_id_rx) = mpsc::channel(16);
6562
6563 Arc::get_mut(&mut client)
6564 .unwrap()
6565 .override_authenticate(move |cx| {
6566 cx.spawn(|_| async move {
6567 let access_token = "the-token".to_string();
6568 Ok(Credentials {
6569 user_id: user_id.0 as u64,
6570 access_token,
6571 })
6572 })
6573 })
6574 .override_establish_connection(move |credentials, cx| {
6575 assert_eq!(credentials.user_id, user_id.0 as u64);
6576 assert_eq!(credentials.access_token, "the-token");
6577
6578 let server = server.clone();
6579 let db = db.clone();
6580 let connection_killers = connection_killers.clone();
6581 let forbid_connections = forbid_connections.clone();
6582 let client_name = client_name.clone();
6583 let connection_id_tx = connection_id_tx.clone();
6584 cx.spawn(move |cx| async move {
6585 if forbid_connections.load(SeqCst) {
6586 Err(EstablishConnectionError::other(anyhow!(
6587 "server is forbidding connections"
6588 )))
6589 } else {
6590 let (client_conn, server_conn, killed) =
6591 Connection::in_memory(cx.background());
6592 connection_killers.lock().insert(user_id, killed);
6593 let user = db.get_user_by_id(user_id).await.unwrap().unwrap();
6594 cx.background()
6595 .spawn(server.handle_connection(
6596 server_conn,
6597 client_name,
6598 user,
6599 Some(connection_id_tx),
6600 cx.background(),
6601 ))
6602 .detach();
6603 Ok(client_conn)
6604 }
6605 })
6606 });
6607
6608 let user_store = cx.add_model(|cx| UserStore::new(client.clone(), http, cx));
6609 let app_state = Arc::new(workspace::AppState {
6610 client: client.clone(),
6611 user_store: user_store.clone(),
6612 languages: Arc::new(LanguageRegistry::new(Task::ready(()))),
6613 themes: ThemeRegistry::new((), cx.font_cache()),
6614 fs: FakeFs::new(cx.background()),
6615 build_window_options: || Default::default(),
6616 initialize_workspace: |_, _, _| unimplemented!(),
6617 });
6618
6619 Channel::init(&client);
6620 Project::init(&client);
6621 cx.update(|cx| workspace::init(app_state.clone(), cx));
6622
6623 client
6624 .authenticate_and_connect(false, &cx.to_async())
6625 .await
6626 .unwrap();
6627 let peer_id = PeerId(connection_id_rx.next().await.unwrap().0);
6628
6629 let client = TestClient {
6630 client,
6631 peer_id,
6632 username: name.to_string(),
6633 user_store,
6634 language_registry: Arc::new(LanguageRegistry::test()),
6635 project: Default::default(),
6636 buffers: Default::default(),
6637 };
6638 client.wait_for_current_user(cx).await;
6639 client
6640 }
6641
6642 fn disconnect_client(&self, user_id: UserId) {
6643 self.connection_killers
6644 .lock()
6645 .remove(&user_id)
6646 .unwrap()
6647 .store(true, SeqCst);
6648 }
6649
6650 fn forbid_connections(&self) {
6651 self.forbid_connections.store(true, SeqCst);
6652 }
6653
6654 fn allow_connections(&self) {
6655 self.forbid_connections.store(false, SeqCst);
6656 }
6657
6658 async fn make_contacts(&self, mut clients: Vec<(&TestClient, &mut TestAppContext)>) {
6659 while let Some((client_a, cx_a)) = clients.pop() {
6660 for (client_b, cx_b) in &mut clients {
6661 client_a
6662 .user_store
6663 .update(cx_a, |store, cx| {
6664 store.request_contact(client_b.user_id().unwrap(), cx)
6665 })
6666 .await
6667 .unwrap();
6668 cx_a.foreground().run_until_parked();
6669 client_b
6670 .user_store
6671 .update(*cx_b, |store, cx| {
6672 store.respond_to_contact_request(client_a.user_id().unwrap(), true, cx)
6673 })
6674 .await
6675 .unwrap();
6676 }
6677 }
6678 }
6679
6680 async fn build_app_state(test_db: &TestDb) -> Arc<AppState> {
6681 Arc::new(AppState {
6682 db: test_db.db().clone(),
6683 api_token: Default::default(),
6684 })
6685 }
6686
6687 async fn state<'a>(&'a self) -> RwLockReadGuard<'a, Store> {
6688 self.server.store.read().await
6689 }
6690
6691 async fn condition<F>(&mut self, mut predicate: F)
6692 where
6693 F: FnMut(&Store) -> bool,
6694 {
6695 assert!(
6696 self.foreground.parking_forbidden(),
6697 "you must call forbid_parking to use server conditions so we don't block indefinitely"
6698 );
6699 while !(predicate)(&*self.server.store.read().await) {
6700 self.foreground.start_waiting();
6701 self.notifications.next().await;
6702 self.foreground.finish_waiting();
6703 }
6704 }
6705 }
6706
6707 impl Deref for TestServer {
6708 type Target = Server;
6709
6710 fn deref(&self) -> &Self::Target {
6711 &self.server
6712 }
6713 }
6714
6715 impl Drop for TestServer {
6716 fn drop(&mut self) {
6717 self.peer.reset();
6718 }
6719 }
6720
6721 struct TestClient {
6722 client: Arc<Client>,
6723 username: String,
6724 pub peer_id: PeerId,
6725 pub user_store: ModelHandle<UserStore>,
6726 language_registry: Arc<LanguageRegistry>,
6727 project: Option<ModelHandle<Project>>,
6728 buffers: HashSet<ModelHandle<language::Buffer>>,
6729 }
6730
6731 impl Deref for TestClient {
6732 type Target = Arc<Client>;
6733
6734 fn deref(&self) -> &Self::Target {
6735 &self.client
6736 }
6737 }
6738
6739 struct ContactsSummary {
6740 pub current: Vec<String>,
6741 pub outgoing_requests: Vec<String>,
6742 pub incoming_requests: Vec<String>,
6743 }
6744
6745 impl TestClient {
6746 pub fn current_user_id(&self, cx: &TestAppContext) -> UserId {
6747 UserId::from_proto(
6748 self.user_store
6749 .read_with(cx, |user_store, _| user_store.current_user().unwrap().id),
6750 )
6751 }
6752
6753 async fn wait_for_current_user(&self, cx: &TestAppContext) {
6754 let mut authed_user = self
6755 .user_store
6756 .read_with(cx, |user_store, _| user_store.watch_current_user());
6757 while authed_user.next().await.unwrap().is_none() {}
6758 }
6759
6760 async fn clear_contacts(&self, cx: &mut TestAppContext) {
6761 self.user_store
6762 .update(cx, |store, _| store.clear_contacts())
6763 .await;
6764 }
6765
6766 fn summarize_contacts(&self, cx: &TestAppContext) -> ContactsSummary {
6767 self.user_store.read_with(cx, |store, _| ContactsSummary {
6768 current: store
6769 .contacts()
6770 .iter()
6771 .map(|contact| contact.user.github_login.clone())
6772 .collect(),
6773 outgoing_requests: store
6774 .outgoing_contact_requests()
6775 .iter()
6776 .map(|user| user.github_login.clone())
6777 .collect(),
6778 incoming_requests: store
6779 .incoming_contact_requests()
6780 .iter()
6781 .map(|user| user.github_login.clone())
6782 .collect(),
6783 })
6784 }
6785
6786 async fn build_local_project(
6787 &mut self,
6788 fs: Arc<FakeFs>,
6789 root_path: impl AsRef<Path>,
6790 cx: &mut TestAppContext,
6791 ) -> (ModelHandle<Project>, WorktreeId) {
6792 let project = cx.update(|cx| {
6793 Project::local(
6794 self.client.clone(),
6795 self.user_store.clone(),
6796 self.language_registry.clone(),
6797 fs,
6798 cx,
6799 )
6800 });
6801 self.project = Some(project.clone());
6802 let (worktree, _) = project
6803 .update(cx, |p, cx| {
6804 p.find_or_create_local_worktree(root_path, true, cx)
6805 })
6806 .await
6807 .unwrap();
6808 worktree
6809 .read_with(cx, |tree, _| tree.as_local().unwrap().scan_complete())
6810 .await;
6811 project
6812 .update(cx, |project, _| project.next_remote_id())
6813 .await;
6814 (project, worktree.read_with(cx, |tree, _| tree.id()))
6815 }
6816
6817 async fn build_remote_project(
6818 &mut self,
6819 host_project: &ModelHandle<Project>,
6820 host_cx: &mut TestAppContext,
6821 guest_cx: &mut TestAppContext,
6822 ) -> ModelHandle<Project> {
6823 let host_project_id = host_project
6824 .read_with(host_cx, |project, _| project.next_remote_id())
6825 .await;
6826 let guest_user_id = self.user_id().unwrap();
6827 let languages =
6828 host_project.read_with(host_cx, |project, _| project.languages().clone());
6829 let project_b = guest_cx.spawn(|mut cx| {
6830 let user_store = self.user_store.clone();
6831 let guest_client = self.client.clone();
6832 async move {
6833 Project::remote(
6834 host_project_id,
6835 guest_client,
6836 user_store.clone(),
6837 languages,
6838 FakeFs::new(cx.background()),
6839 &mut cx,
6840 )
6841 .await
6842 .unwrap()
6843 }
6844 });
6845 host_cx.foreground().run_until_parked();
6846 host_project.update(host_cx, |project, cx| {
6847 project.respond_to_join_request(guest_user_id, true, cx)
6848 });
6849 let project = project_b.await;
6850 self.project = Some(project.clone());
6851 project
6852 }
6853
6854 fn build_workspace(
6855 &self,
6856 project: &ModelHandle<Project>,
6857 cx: &mut TestAppContext,
6858 ) -> ViewHandle<Workspace> {
6859 let (window_id, _) = cx.add_window(|_| EmptyView);
6860 cx.add_view(window_id, |cx| Workspace::new(project.clone(), cx))
6861 }
6862
6863 async fn simulate_host(
6864 mut self,
6865 project: ModelHandle<Project>,
6866 op_start_signal: futures::channel::mpsc::UnboundedReceiver<()>,
6867 rng: Arc<Mutex<StdRng>>,
6868 mut cx: TestAppContext,
6869 ) -> (Self, TestAppContext, Option<anyhow::Error>) {
6870 async fn simulate_host_internal(
6871 client: &mut TestClient,
6872 project: ModelHandle<Project>,
6873 mut op_start_signal: futures::channel::mpsc::UnboundedReceiver<()>,
6874 rng: Arc<Mutex<StdRng>>,
6875 cx: &mut TestAppContext,
6876 ) -> anyhow::Result<()> {
6877 let fs = project.read_with(cx, |project, _| project.fs().clone());
6878
6879 cx.update(|cx| {
6880 cx.subscribe(&project, move |project, event, cx| {
6881 if let project::Event::ContactRequestedJoin(user) = event {
6882 log::info!("Host: accepting join request from {}", user.github_login);
6883 project.update(cx, |project, cx| {
6884 project.respond_to_join_request(user.id, true, cx)
6885 });
6886 }
6887 })
6888 .detach();
6889 });
6890
6891 while op_start_signal.next().await.is_some() {
6892 let distribution = rng.lock().gen_range::<usize, _>(0..100);
6893 let files = fs.as_fake().files().await;
6894 match distribution {
6895 0..=19 if !files.is_empty() => {
6896 let path = files.choose(&mut *rng.lock()).unwrap();
6897 let mut path = path.as_path();
6898 while let Some(parent_path) = path.parent() {
6899 path = parent_path;
6900 if rng.lock().gen() {
6901 break;
6902 }
6903 }
6904
6905 log::info!("Host: find/create local worktree {:?}", path);
6906 let find_or_create_worktree = project.update(cx, |project, cx| {
6907 project.find_or_create_local_worktree(path, true, cx)
6908 });
6909 if rng.lock().gen() {
6910 cx.background().spawn(find_or_create_worktree).detach();
6911 } else {
6912 find_or_create_worktree.await?;
6913 }
6914 }
6915 20..=79 if !files.is_empty() => {
6916 let buffer = if client.buffers.is_empty() || rng.lock().gen() {
6917 let file = files.choose(&mut *rng.lock()).unwrap();
6918 let (worktree, path) = project
6919 .update(cx, |project, cx| {
6920 project.find_or_create_local_worktree(
6921 file.clone(),
6922 true,
6923 cx,
6924 )
6925 })
6926 .await?;
6927 let project_path =
6928 worktree.read_with(cx, |worktree, _| (worktree.id(), path));
6929 log::info!(
6930 "Host: opening path {:?}, worktree {}, relative_path {:?}",
6931 file,
6932 project_path.0,
6933 project_path.1
6934 );
6935 let buffer = project
6936 .update(cx, |project, cx| project.open_buffer(project_path, cx))
6937 .await
6938 .unwrap();
6939 client.buffers.insert(buffer.clone());
6940 buffer
6941 } else {
6942 client
6943 .buffers
6944 .iter()
6945 .choose(&mut *rng.lock())
6946 .unwrap()
6947 .clone()
6948 };
6949
6950 if rng.lock().gen_bool(0.1) {
6951 cx.update(|cx| {
6952 log::info!(
6953 "Host: dropping buffer {:?}",
6954 buffer.read(cx).file().unwrap().full_path(cx)
6955 );
6956 client.buffers.remove(&buffer);
6957 drop(buffer);
6958 });
6959 } else {
6960 buffer.update(cx, |buffer, cx| {
6961 log::info!(
6962 "Host: updating buffer {:?} ({})",
6963 buffer.file().unwrap().full_path(cx),
6964 buffer.remote_id()
6965 );
6966
6967 if rng.lock().gen_bool(0.7) {
6968 buffer.randomly_edit(&mut *rng.lock(), 5, cx);
6969 } else {
6970 buffer.randomly_undo_redo(&mut *rng.lock(), cx);
6971 }
6972 });
6973 }
6974 }
6975 _ => loop {
6976 let path_component_count = rng.lock().gen_range::<usize, _>(1..=5);
6977 let mut path = PathBuf::new();
6978 path.push("/");
6979 for _ in 0..path_component_count {
6980 let letter = rng.lock().gen_range(b'a'..=b'z');
6981 path.push(std::str::from_utf8(&[letter]).unwrap());
6982 }
6983 path.set_extension("rs");
6984 let parent_path = path.parent().unwrap();
6985
6986 log::info!("Host: creating file {:?}", path,);
6987
6988 if fs.create_dir(&parent_path).await.is_ok()
6989 && fs.create_file(&path, Default::default()).await.is_ok()
6990 {
6991 break;
6992 } else {
6993 log::info!("Host: cannot create file");
6994 }
6995 },
6996 }
6997
6998 cx.background().simulate_random_delay().await;
6999 }
7000
7001 Ok(())
7002 }
7003
7004 let result =
7005 simulate_host_internal(&mut self, project.clone(), op_start_signal, rng, &mut cx)
7006 .await;
7007 log::info!("Host done");
7008 self.project = Some(project);
7009 (self, cx, result.err())
7010 }
7011
7012 pub async fn simulate_guest(
7013 mut self,
7014 guest_username: String,
7015 project: ModelHandle<Project>,
7016 op_start_signal: futures::channel::mpsc::UnboundedReceiver<()>,
7017 rng: Arc<Mutex<StdRng>>,
7018 mut cx: TestAppContext,
7019 ) -> (Self, TestAppContext, Option<anyhow::Error>) {
7020 async fn simulate_guest_internal(
7021 client: &mut TestClient,
7022 guest_username: &str,
7023 project: ModelHandle<Project>,
7024 mut op_start_signal: futures::channel::mpsc::UnboundedReceiver<()>,
7025 rng: Arc<Mutex<StdRng>>,
7026 cx: &mut TestAppContext,
7027 ) -> anyhow::Result<()> {
7028 while op_start_signal.next().await.is_some() {
7029 let buffer = if client.buffers.is_empty() || rng.lock().gen() {
7030 let worktree = if let Some(worktree) =
7031 project.read_with(cx, |project, cx| {
7032 project
7033 .worktrees(&cx)
7034 .filter(|worktree| {
7035 let worktree = worktree.read(cx);
7036 worktree.is_visible()
7037 && worktree.entries(false).any(|e| e.is_file())
7038 })
7039 .choose(&mut *rng.lock())
7040 }) {
7041 worktree
7042 } else {
7043 cx.background().simulate_random_delay().await;
7044 continue;
7045 };
7046
7047 let (worktree_root_name, project_path) =
7048 worktree.read_with(cx, |worktree, _| {
7049 let entry = worktree
7050 .entries(false)
7051 .filter(|e| e.is_file())
7052 .choose(&mut *rng.lock())
7053 .unwrap();
7054 (
7055 worktree.root_name().to_string(),
7056 (worktree.id(), entry.path.clone()),
7057 )
7058 });
7059 log::info!(
7060 "{}: opening path {:?} in worktree {} ({})",
7061 guest_username,
7062 project_path.1,
7063 project_path.0,
7064 worktree_root_name,
7065 );
7066 let buffer = project
7067 .update(cx, |project, cx| {
7068 project.open_buffer(project_path.clone(), cx)
7069 })
7070 .await?;
7071 log::info!(
7072 "{}: opened path {:?} in worktree {} ({}) with buffer id {}",
7073 guest_username,
7074 project_path.1,
7075 project_path.0,
7076 worktree_root_name,
7077 buffer.read_with(cx, |buffer, _| buffer.remote_id())
7078 );
7079 client.buffers.insert(buffer.clone());
7080 buffer
7081 } else {
7082 client
7083 .buffers
7084 .iter()
7085 .choose(&mut *rng.lock())
7086 .unwrap()
7087 .clone()
7088 };
7089
7090 let choice = rng.lock().gen_range(0..100);
7091 match choice {
7092 0..=9 => {
7093 cx.update(|cx| {
7094 log::info!(
7095 "{}: dropping buffer {:?}",
7096 guest_username,
7097 buffer.read(cx).file().unwrap().full_path(cx)
7098 );
7099 client.buffers.remove(&buffer);
7100 drop(buffer);
7101 });
7102 }
7103 10..=19 => {
7104 let completions = project.update(cx, |project, cx| {
7105 log::info!(
7106 "{}: requesting completions for buffer {} ({:?})",
7107 guest_username,
7108 buffer.read(cx).remote_id(),
7109 buffer.read(cx).file().unwrap().full_path(cx)
7110 );
7111 let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
7112 project.completions(&buffer, offset, cx)
7113 });
7114 let completions = cx.background().spawn(async move {
7115 completions
7116 .await
7117 .map_err(|err| anyhow!("completions request failed: {:?}", err))
7118 });
7119 if rng.lock().gen_bool(0.3) {
7120 log::info!("{}: detaching completions request", guest_username);
7121 cx.update(|cx| completions.detach_and_log_err(cx));
7122 } else {
7123 completions.await?;
7124 }
7125 }
7126 20..=29 => {
7127 let code_actions = project.update(cx, |project, cx| {
7128 log::info!(
7129 "{}: requesting code actions for buffer {} ({:?})",
7130 guest_username,
7131 buffer.read(cx).remote_id(),
7132 buffer.read(cx).file().unwrap().full_path(cx)
7133 );
7134 let range = buffer.read(cx).random_byte_range(0, &mut *rng.lock());
7135 project.code_actions(&buffer, range, cx)
7136 });
7137 let code_actions = cx.background().spawn(async move {
7138 code_actions.await.map_err(|err| {
7139 anyhow!("code actions request failed: {:?}", err)
7140 })
7141 });
7142 if rng.lock().gen_bool(0.3) {
7143 log::info!("{}: detaching code actions request", guest_username);
7144 cx.update(|cx| code_actions.detach_and_log_err(cx));
7145 } else {
7146 code_actions.await?;
7147 }
7148 }
7149 30..=39 if buffer.read_with(cx, |buffer, _| buffer.is_dirty()) => {
7150 let (requested_version, save) = buffer.update(cx, |buffer, cx| {
7151 log::info!(
7152 "{}: saving buffer {} ({:?})",
7153 guest_username,
7154 buffer.remote_id(),
7155 buffer.file().unwrap().full_path(cx)
7156 );
7157 (buffer.version(), buffer.save(cx))
7158 });
7159 let save = cx.background().spawn(async move {
7160 let (saved_version, _) = save
7161 .await
7162 .map_err(|err| anyhow!("save request failed: {:?}", err))?;
7163 assert!(saved_version.observed_all(&requested_version));
7164 Ok::<_, anyhow::Error>(())
7165 });
7166 if rng.lock().gen_bool(0.3) {
7167 log::info!("{}: detaching save request", guest_username);
7168 cx.update(|cx| save.detach_and_log_err(cx));
7169 } else {
7170 save.await?;
7171 }
7172 }
7173 40..=44 => {
7174 let prepare_rename = project.update(cx, |project, cx| {
7175 log::info!(
7176 "{}: preparing rename for buffer {} ({:?})",
7177 guest_username,
7178 buffer.read(cx).remote_id(),
7179 buffer.read(cx).file().unwrap().full_path(cx)
7180 );
7181 let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
7182 project.prepare_rename(buffer, offset, cx)
7183 });
7184 let prepare_rename = cx.background().spawn(async move {
7185 prepare_rename.await.map_err(|err| {
7186 anyhow!("prepare rename request failed: {:?}", err)
7187 })
7188 });
7189 if rng.lock().gen_bool(0.3) {
7190 log::info!("{}: detaching prepare rename request", guest_username);
7191 cx.update(|cx| prepare_rename.detach_and_log_err(cx));
7192 } else {
7193 prepare_rename.await?;
7194 }
7195 }
7196 45..=49 => {
7197 let definitions = project.update(cx, |project, cx| {
7198 log::info!(
7199 "{}: requesting definitions for buffer {} ({:?})",
7200 guest_username,
7201 buffer.read(cx).remote_id(),
7202 buffer.read(cx).file().unwrap().full_path(cx)
7203 );
7204 let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
7205 project.definition(&buffer, offset, cx)
7206 });
7207 let definitions = cx.background().spawn(async move {
7208 definitions
7209 .await
7210 .map_err(|err| anyhow!("definitions request failed: {:?}", err))
7211 });
7212 if rng.lock().gen_bool(0.3) {
7213 log::info!("{}: detaching definitions request", guest_username);
7214 cx.update(|cx| definitions.detach_and_log_err(cx));
7215 } else {
7216 client
7217 .buffers
7218 .extend(definitions.await?.into_iter().map(|loc| loc.buffer));
7219 }
7220 }
7221 50..=54 => {
7222 let highlights = project.update(cx, |project, cx| {
7223 log::info!(
7224 "{}: requesting highlights for buffer {} ({:?})",
7225 guest_username,
7226 buffer.read(cx).remote_id(),
7227 buffer.read(cx).file().unwrap().full_path(cx)
7228 );
7229 let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
7230 project.document_highlights(&buffer, offset, cx)
7231 });
7232 let highlights = cx.background().spawn(async move {
7233 highlights
7234 .await
7235 .map_err(|err| anyhow!("highlights request failed: {:?}", err))
7236 });
7237 if rng.lock().gen_bool(0.3) {
7238 log::info!("{}: detaching highlights request", guest_username);
7239 cx.update(|cx| highlights.detach_and_log_err(cx));
7240 } else {
7241 highlights.await?;
7242 }
7243 }
7244 55..=59 => {
7245 let search = project.update(cx, |project, cx| {
7246 let query = rng.lock().gen_range('a'..='z');
7247 log::info!("{}: project-wide search {:?}", guest_username, query);
7248 project.search(SearchQuery::text(query, false, false), cx)
7249 });
7250 let search = cx.background().spawn(async move {
7251 search
7252 .await
7253 .map_err(|err| anyhow!("search request failed: {:?}", err))
7254 });
7255 if rng.lock().gen_bool(0.3) {
7256 log::info!("{}: detaching search request", guest_username);
7257 cx.update(|cx| search.detach_and_log_err(cx));
7258 } else {
7259 client.buffers.extend(search.await?.into_keys());
7260 }
7261 }
7262 60..=69 => {
7263 let worktree = project
7264 .read_with(cx, |project, cx| {
7265 project
7266 .worktrees(&cx)
7267 .filter(|worktree| {
7268 let worktree = worktree.read(cx);
7269 worktree.is_visible()
7270 && worktree.entries(false).any(|e| e.is_file())
7271 && worktree
7272 .root_entry()
7273 .map_or(false, |e| e.is_dir())
7274 })
7275 .choose(&mut *rng.lock())
7276 })
7277 .unwrap();
7278 let (worktree_id, worktree_root_name) = worktree
7279 .read_with(cx, |worktree, _| {
7280 (worktree.id(), worktree.root_name().to_string())
7281 });
7282
7283 let mut new_name = String::new();
7284 for _ in 0..10 {
7285 let letter = rng.lock().gen_range('a'..='z');
7286 new_name.push(letter);
7287 }
7288 let mut new_path = PathBuf::new();
7289 new_path.push(new_name);
7290 new_path.set_extension("rs");
7291 log::info!(
7292 "{}: creating {:?} in worktree {} ({})",
7293 guest_username,
7294 new_path,
7295 worktree_id,
7296 worktree_root_name,
7297 );
7298 project
7299 .update(cx, |project, cx| {
7300 project.create_entry((worktree_id, new_path), false, cx)
7301 })
7302 .unwrap()
7303 .await?;
7304 }
7305 _ => {
7306 buffer.update(cx, |buffer, cx| {
7307 log::info!(
7308 "{}: updating buffer {} ({:?})",
7309 guest_username,
7310 buffer.remote_id(),
7311 buffer.file().unwrap().full_path(cx)
7312 );
7313 if rng.lock().gen_bool(0.7) {
7314 buffer.randomly_edit(&mut *rng.lock(), 5, cx);
7315 } else {
7316 buffer.randomly_undo_redo(&mut *rng.lock(), cx);
7317 }
7318 });
7319 }
7320 }
7321 cx.background().simulate_random_delay().await;
7322 }
7323 Ok(())
7324 }
7325
7326 let result = simulate_guest_internal(
7327 &mut self,
7328 &guest_username,
7329 project.clone(),
7330 op_start_signal,
7331 rng,
7332 &mut cx,
7333 )
7334 .await;
7335 log::info!("{}: done", guest_username);
7336
7337 self.project = Some(project);
7338 (self, cx, result.err())
7339 }
7340 }
7341
7342 impl Drop for TestClient {
7343 fn drop(&mut self) {
7344 self.client.tear_down();
7345 }
7346 }
7347
7348 impl Executor for Arc<gpui::executor::Background> {
7349 type Sleep = gpui::executor::Timer;
7350
7351 fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F) {
7352 self.spawn(future).detach();
7353 }
7354
7355 fn sleep(&self, duration: Duration) -> Self::Sleep {
7356 self.as_ref().timer(duration)
7357 }
7358 }
7359
7360 fn channel_messages(channel: &Channel) -> Vec<(String, String, bool)> {
7361 channel
7362 .messages()
7363 .cursor::<()>()
7364 .map(|m| {
7365 (
7366 m.sender.github_login.clone(),
7367 m.body.clone(),
7368 m.is_pending(),
7369 )
7370 })
7371 .collect()
7372 }
7373
7374 struct EmptyView;
7375
7376 impl gpui::Entity for EmptyView {
7377 type Event = ();
7378 }
7379
7380 impl gpui::View for EmptyView {
7381 fn ui_name() -> &'static str {
7382 "empty view"
7383 }
7384
7385 fn render(&mut self, _: &mut gpui::RenderContext<Self>) -> gpui::ElementBox {
7386 gpui::Element::boxed(gpui::elements::Empty::new())
7387 }
7388 }
7389}