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