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