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