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