1use super::{proto, Client, Status, TypedEnvelope};
2use anyhow::{anyhow, Context, Result};
3use collections::{hash_map::Entry, HashMap, HashSet};
4use feature_flags::FeatureFlagAppExt;
5use futures::{channel::mpsc, Future, StreamExt};
6use gpui::{
7 AppContext, AsyncAppContext, EventEmitter, Model, ModelContext, SharedString, SharedUri, Task,
8 WeakModel,
9};
10use postage::{sink::Sink, watch};
11use rpc::proto::{RequestMessage, UsersResponse};
12use std::sync::{Arc, Weak};
13use text::ReplicaId;
14use util::TryFutureExt as _;
15
16pub type UserId = u64;
17
18#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Copy)]
19pub struct ChannelId(pub u64);
20
21impl std::fmt::Display for ChannelId {
22 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
23 self.0.fmt(f)
24 }
25}
26
27#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Copy)]
28pub struct ProjectId(pub u64);
29
30#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Copy)]
31pub struct DevServerId(pub u64);
32
33#[derive(
34 Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Copy, serde::Serialize, serde::Deserialize,
35)]
36pub struct DevServerProjectId(pub u64);
37
38#[derive(Debug, Clone, Copy, PartialEq, Eq)]
39pub struct ParticipantIndex(pub u32);
40
41#[derive(Default, Debug)]
42pub struct User {
43 pub id: UserId,
44 pub github_login: String,
45 pub avatar_uri: SharedUri,
46}
47
48#[derive(Clone, Debug, PartialEq, Eq)]
49pub struct Collaborator {
50 pub peer_id: proto::PeerId,
51 pub replica_id: ReplicaId,
52 pub user_id: UserId,
53}
54
55impl PartialOrd for User {
56 fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
57 Some(self.cmp(other))
58 }
59}
60
61impl Ord for User {
62 fn cmp(&self, other: &Self) -> std::cmp::Ordering {
63 self.github_login.cmp(&other.github_login)
64 }
65}
66
67impl PartialEq for User {
68 fn eq(&self, other: &Self) -> bool {
69 self.id == other.id && self.github_login == other.github_login
70 }
71}
72
73impl Eq for User {}
74
75#[derive(Debug, PartialEq)]
76pub struct Contact {
77 pub user: Arc<User>,
78 pub online: bool,
79 pub busy: bool,
80}
81
82#[derive(Debug, Clone, Copy, PartialEq, Eq)]
83pub enum ContactRequestStatus {
84 None,
85 RequestSent,
86 RequestReceived,
87 RequestAccepted,
88}
89
90pub struct UserStore {
91 users: HashMap<u64, Arc<User>>,
92 by_github_login: HashMap<String, u64>,
93 participant_indices: HashMap<u64, ParticipantIndex>,
94 update_contacts_tx: mpsc::UnboundedSender<UpdateContacts>,
95 current_user: watch::Receiver<Option<Arc<User>>>,
96 contacts: Vec<Arc<Contact>>,
97 incoming_contact_requests: Vec<Arc<User>>,
98 outgoing_contact_requests: Vec<Arc<User>>,
99 pending_contact_requests: HashMap<u64, usize>,
100 invite_info: Option<InviteInfo>,
101 client: Weak<Client>,
102 _maintain_contacts: Task<()>,
103 _maintain_current_user: Task<Result<()>>,
104 weak_self: WeakModel<Self>,
105}
106
107#[derive(Clone)]
108pub struct InviteInfo {
109 pub count: u32,
110 pub url: Arc<str>,
111}
112
113pub enum Event {
114 Contact {
115 user: Arc<User>,
116 kind: ContactEventKind,
117 },
118 ShowContacts,
119 ParticipantIndicesChanged,
120}
121
122#[derive(Clone, Copy)]
123pub enum ContactEventKind {
124 Requested,
125 Accepted,
126 Cancelled,
127}
128
129impl EventEmitter<Event> for UserStore {}
130
131enum UpdateContacts {
132 Update(proto::UpdateContacts),
133 Wait(postage::barrier::Sender),
134 Clear(postage::barrier::Sender),
135}
136
137impl UserStore {
138 pub fn new(client: Arc<Client>, cx: &mut ModelContext<Self>) -> Self {
139 let (mut current_user_tx, current_user_rx) = watch::channel();
140 let (update_contacts_tx, mut update_contacts_rx) = mpsc::unbounded();
141 let rpc_subscriptions = vec![
142 client.add_message_handler(cx.weak_model(), Self::handle_update_contacts),
143 client.add_message_handler(cx.weak_model(), Self::handle_update_invite_info),
144 client.add_message_handler(cx.weak_model(), Self::handle_show_contacts),
145 ];
146 Self {
147 users: Default::default(),
148 by_github_login: Default::default(),
149 current_user: current_user_rx,
150 contacts: Default::default(),
151 incoming_contact_requests: Default::default(),
152 participant_indices: Default::default(),
153 outgoing_contact_requests: Default::default(),
154 invite_info: None,
155 client: Arc::downgrade(&client),
156 update_contacts_tx,
157 _maintain_contacts: cx.spawn(|this, mut cx| async move {
158 let _subscriptions = rpc_subscriptions;
159 while let Some(message) = update_contacts_rx.next().await {
160 if let Ok(task) =
161 this.update(&mut cx, |this, cx| this.update_contacts(message, cx))
162 {
163 task.log_err().await;
164 } else {
165 break;
166 }
167 }
168 }),
169 _maintain_current_user: cx.spawn(|this, mut cx| async move {
170 let mut status = client.status();
171 let weak = Arc::downgrade(&client);
172 drop(client);
173 while let Some(status) = status.next().await {
174 // if the client is dropped, the app is shutting down.
175 let Some(client) = weak.upgrade() else {
176 return Ok(());
177 };
178 match status {
179 Status::Connected { .. } => {
180 if let Some(user_id) = client.user_id() {
181 let fetch_user = if let Ok(fetch_user) = this
182 .update(&mut cx, |this, cx| {
183 this.get_user(user_id, cx).log_err()
184 }) {
185 fetch_user
186 } else {
187 break;
188 };
189 let fetch_metrics_id =
190 client.request(proto::GetPrivateUserInfo {}).log_err();
191 let (user, info) = futures::join!(fetch_user, fetch_metrics_id);
192
193 cx.update(|cx| {
194 if let Some(info) = info {
195 let disable_staff = std::env::var("ZED_DISABLE_STAFF")
196 .map_or(false, |v| v != "" && v != "0");
197 let staff = info.staff && !disable_staff;
198 cx.update_flags(staff, info.flags);
199 client.telemetry.set_authenticated_user_info(
200 Some(info.metrics_id.clone()),
201 staff,
202 )
203 }
204 })?;
205
206 current_user_tx.send(user).await.ok();
207
208 this.update(&mut cx, |_, cx| cx.notify())?;
209 }
210 }
211 Status::SignedOut => {
212 current_user_tx.send(None).await.ok();
213 this.update(&mut cx, |this, cx| {
214 cx.notify();
215 this.clear_contacts()
216 })?
217 .await;
218 }
219 Status::ConnectionLost => {
220 this.update(&mut cx, |this, cx| {
221 cx.notify();
222 this.clear_contacts()
223 })?
224 .await;
225 }
226 _ => {}
227 }
228 }
229 Ok(())
230 }),
231 pending_contact_requests: Default::default(),
232 weak_self: cx.weak_model(),
233 }
234 }
235
236 #[cfg(feature = "test-support")]
237 pub fn clear_cache(&mut self) {
238 self.users.clear();
239 self.by_github_login.clear();
240 }
241
242 async fn handle_update_invite_info(
243 this: Model<Self>,
244 message: TypedEnvelope<proto::UpdateInviteInfo>,
245 mut cx: AsyncAppContext,
246 ) -> Result<()> {
247 this.update(&mut cx, |this, cx| {
248 this.invite_info = Some(InviteInfo {
249 url: Arc::from(message.payload.url),
250 count: message.payload.count,
251 });
252 cx.notify();
253 })?;
254 Ok(())
255 }
256
257 async fn handle_show_contacts(
258 this: Model<Self>,
259 _: TypedEnvelope<proto::ShowContacts>,
260 mut cx: AsyncAppContext,
261 ) -> Result<()> {
262 this.update(&mut cx, |_, cx| cx.emit(Event::ShowContacts))?;
263 Ok(())
264 }
265
266 pub fn invite_info(&self) -> Option<&InviteInfo> {
267 self.invite_info.as_ref()
268 }
269
270 async fn handle_update_contacts(
271 this: Model<Self>,
272 message: TypedEnvelope<proto::UpdateContacts>,
273 mut cx: AsyncAppContext,
274 ) -> Result<()> {
275 this.update(&mut cx, |this, _| {
276 this.update_contacts_tx
277 .unbounded_send(UpdateContacts::Update(message.payload))
278 .unwrap();
279 })?;
280 Ok(())
281 }
282
283 fn update_contacts(
284 &mut self,
285 message: UpdateContacts,
286 cx: &mut ModelContext<Self>,
287 ) -> Task<Result<()>> {
288 match message {
289 UpdateContacts::Wait(barrier) => {
290 drop(barrier);
291 Task::ready(Ok(()))
292 }
293 UpdateContacts::Clear(barrier) => {
294 self.contacts.clear();
295 self.incoming_contact_requests.clear();
296 self.outgoing_contact_requests.clear();
297 drop(barrier);
298 Task::ready(Ok(()))
299 }
300 UpdateContacts::Update(message) => {
301 let mut user_ids = HashSet::default();
302 for contact in &message.contacts {
303 user_ids.insert(contact.user_id);
304 }
305 user_ids.extend(message.incoming_requests.iter().map(|req| req.requester_id));
306 user_ids.extend(message.outgoing_requests.iter());
307
308 let load_users = self.get_users(user_ids.into_iter().collect(), cx);
309 cx.spawn(|this, mut cx| async move {
310 load_users.await?;
311
312 // Users are fetched in parallel above and cached in call to get_users
313 // No need to parallelize here
314 let mut updated_contacts = Vec::new();
315 let this = this
316 .upgrade()
317 .ok_or_else(|| anyhow!("can't upgrade user store handle"))?;
318 for contact in message.contacts {
319 updated_contacts.push(Arc::new(
320 Contact::from_proto(contact, &this, &mut cx).await?,
321 ));
322 }
323
324 let mut incoming_requests = Vec::new();
325 for request in message.incoming_requests {
326 incoming_requests.push({
327 this.update(&mut cx, |this, cx| {
328 this.get_user(request.requester_id, cx)
329 })?
330 .await?
331 });
332 }
333
334 let mut outgoing_requests = Vec::new();
335 for requested_user_id in message.outgoing_requests {
336 outgoing_requests.push(
337 this.update(&mut cx, |this, cx| this.get_user(requested_user_id, cx))?
338 .await?,
339 );
340 }
341
342 let removed_contacts =
343 HashSet::<u64>::from_iter(message.remove_contacts.iter().copied());
344 let removed_incoming_requests =
345 HashSet::<u64>::from_iter(message.remove_incoming_requests.iter().copied());
346 let removed_outgoing_requests =
347 HashSet::<u64>::from_iter(message.remove_outgoing_requests.iter().copied());
348
349 this.update(&mut cx, |this, cx| {
350 // Remove contacts
351 this.contacts
352 .retain(|contact| !removed_contacts.contains(&contact.user.id));
353 // Update existing contacts and insert new ones
354 for updated_contact in updated_contacts {
355 match this.contacts.binary_search_by_key(
356 &&updated_contact.user.github_login,
357 |contact| &contact.user.github_login,
358 ) {
359 Ok(ix) => this.contacts[ix] = updated_contact,
360 Err(ix) => this.contacts.insert(ix, updated_contact),
361 }
362 }
363
364 // Remove incoming contact requests
365 this.incoming_contact_requests.retain(|user| {
366 if removed_incoming_requests.contains(&user.id) {
367 cx.emit(Event::Contact {
368 user: user.clone(),
369 kind: ContactEventKind::Cancelled,
370 });
371 false
372 } else {
373 true
374 }
375 });
376 // Update existing incoming requests and insert new ones
377 for user in incoming_requests {
378 match this
379 .incoming_contact_requests
380 .binary_search_by_key(&&user.github_login, |contact| {
381 &contact.github_login
382 }) {
383 Ok(ix) => this.incoming_contact_requests[ix] = user,
384 Err(ix) => this.incoming_contact_requests.insert(ix, user),
385 }
386 }
387
388 // Remove outgoing contact requests
389 this.outgoing_contact_requests
390 .retain(|user| !removed_outgoing_requests.contains(&user.id));
391 // Update existing incoming requests and insert new ones
392 for request in outgoing_requests {
393 match this
394 .outgoing_contact_requests
395 .binary_search_by_key(&&request.github_login, |contact| {
396 &contact.github_login
397 }) {
398 Ok(ix) => this.outgoing_contact_requests[ix] = request,
399 Err(ix) => this.outgoing_contact_requests.insert(ix, request),
400 }
401 }
402
403 cx.notify();
404 })?;
405
406 Ok(())
407 })
408 }
409 }
410 }
411
412 pub fn contacts(&self) -> &[Arc<Contact>] {
413 &self.contacts
414 }
415
416 pub fn has_contact(&self, user: &Arc<User>) -> bool {
417 self.contacts
418 .binary_search_by_key(&&user.github_login, |contact| &contact.user.github_login)
419 .is_ok()
420 }
421
422 pub fn incoming_contact_requests(&self) -> &[Arc<User>] {
423 &self.incoming_contact_requests
424 }
425
426 pub fn outgoing_contact_requests(&self) -> &[Arc<User>] {
427 &self.outgoing_contact_requests
428 }
429
430 pub fn is_contact_request_pending(&self, user: &User) -> bool {
431 self.pending_contact_requests.contains_key(&user.id)
432 }
433
434 pub fn contact_request_status(&self, user: &User) -> ContactRequestStatus {
435 if self
436 .contacts
437 .binary_search_by_key(&&user.github_login, |contact| &contact.user.github_login)
438 .is_ok()
439 {
440 ContactRequestStatus::RequestAccepted
441 } else if self
442 .outgoing_contact_requests
443 .binary_search_by_key(&&user.github_login, |user| &user.github_login)
444 .is_ok()
445 {
446 ContactRequestStatus::RequestSent
447 } else if self
448 .incoming_contact_requests
449 .binary_search_by_key(&&user.github_login, |user| &user.github_login)
450 .is_ok()
451 {
452 ContactRequestStatus::RequestReceived
453 } else {
454 ContactRequestStatus::None
455 }
456 }
457
458 pub fn request_contact(
459 &mut self,
460 responder_id: u64,
461 cx: &mut ModelContext<Self>,
462 ) -> Task<Result<()>> {
463 self.perform_contact_request(responder_id, proto::RequestContact { responder_id }, cx)
464 }
465
466 pub fn remove_contact(
467 &mut self,
468 user_id: u64,
469 cx: &mut ModelContext<Self>,
470 ) -> Task<Result<()>> {
471 self.perform_contact_request(user_id, proto::RemoveContact { user_id }, cx)
472 }
473
474 pub fn has_incoming_contact_request(&self, user_id: u64) -> bool {
475 self.incoming_contact_requests
476 .iter()
477 .any(|user| user.id == user_id)
478 }
479
480 pub fn respond_to_contact_request(
481 &mut self,
482 requester_id: u64,
483 accept: bool,
484 cx: &mut ModelContext<Self>,
485 ) -> Task<Result<()>> {
486 self.perform_contact_request(
487 requester_id,
488 proto::RespondToContactRequest {
489 requester_id,
490 response: if accept {
491 proto::ContactRequestResponse::Accept
492 } else {
493 proto::ContactRequestResponse::Decline
494 } as i32,
495 },
496 cx,
497 )
498 }
499
500 pub fn dismiss_contact_request(
501 &mut self,
502 requester_id: u64,
503 cx: &mut ModelContext<Self>,
504 ) -> Task<Result<()>> {
505 let client = self.client.upgrade();
506 cx.spawn(move |_, _| async move {
507 client
508 .ok_or_else(|| anyhow!("can't upgrade client reference"))?
509 .request(proto::RespondToContactRequest {
510 requester_id,
511 response: proto::ContactRequestResponse::Dismiss as i32,
512 })
513 .await?;
514 Ok(())
515 })
516 }
517
518 fn perform_contact_request<T: RequestMessage>(
519 &mut self,
520 user_id: u64,
521 request: T,
522 cx: &mut ModelContext<Self>,
523 ) -> Task<Result<()>> {
524 let client = self.client.upgrade();
525 *self.pending_contact_requests.entry(user_id).or_insert(0) += 1;
526 cx.notify();
527
528 cx.spawn(move |this, mut cx| async move {
529 let response = client
530 .ok_or_else(|| anyhow!("can't upgrade client reference"))?
531 .request(request)
532 .await;
533 this.update(&mut cx, |this, cx| {
534 if let Entry::Occupied(mut request_count) =
535 this.pending_contact_requests.entry(user_id)
536 {
537 *request_count.get_mut() -= 1;
538 if *request_count.get() == 0 {
539 request_count.remove();
540 }
541 }
542 cx.notify();
543 })?;
544 response?;
545 Ok(())
546 })
547 }
548
549 pub fn clear_contacts(&mut self) -> impl Future<Output = ()> {
550 let (tx, mut rx) = postage::barrier::channel();
551 self.update_contacts_tx
552 .unbounded_send(UpdateContacts::Clear(tx))
553 .unwrap();
554 async move {
555 rx.next().await;
556 }
557 }
558
559 pub fn contact_updates_done(&mut self) -> impl Future<Output = ()> {
560 let (tx, mut rx) = postage::barrier::channel();
561 self.update_contacts_tx
562 .unbounded_send(UpdateContacts::Wait(tx))
563 .unwrap();
564 async move {
565 rx.next().await;
566 }
567 }
568
569 pub fn get_users(
570 &mut self,
571 user_ids: Vec<u64>,
572 cx: &mut ModelContext<Self>,
573 ) -> Task<Result<Vec<Arc<User>>>> {
574 let mut user_ids_to_fetch = user_ids.clone();
575 user_ids_to_fetch.retain(|id| !self.users.contains_key(id));
576
577 cx.spawn(|this, mut cx| async move {
578 if !user_ids_to_fetch.is_empty() {
579 this.update(&mut cx, |this, cx| {
580 this.load_users(
581 proto::GetUsers {
582 user_ids: user_ids_to_fetch,
583 },
584 cx,
585 )
586 })?
587 .await?;
588 }
589
590 this.update(&mut cx, |this, _| {
591 user_ids
592 .iter()
593 .map(|user_id| {
594 this.users
595 .get(user_id)
596 .cloned()
597 .ok_or_else(|| anyhow!("user {} not found", user_id))
598 })
599 .collect()
600 })?
601 })
602 }
603
604 pub fn fuzzy_search_users(
605 &mut self,
606 query: String,
607 cx: &mut ModelContext<Self>,
608 ) -> Task<Result<Vec<Arc<User>>>> {
609 self.load_users(proto::FuzzySearchUsers { query }, cx)
610 }
611
612 pub fn get_cached_user(&self, user_id: u64) -> Option<Arc<User>> {
613 self.users.get(&user_id).cloned()
614 }
615
616 pub fn get_user_optimistic(
617 &mut self,
618 user_id: u64,
619 cx: &mut ModelContext<Self>,
620 ) -> Option<Arc<User>> {
621 if let Some(user) = self.users.get(&user_id).cloned() {
622 return Some(user);
623 }
624
625 self.get_user(user_id, cx).detach_and_log_err(cx);
626 None
627 }
628
629 pub fn get_user(
630 &mut self,
631 user_id: u64,
632 cx: &mut ModelContext<Self>,
633 ) -> Task<Result<Arc<User>>> {
634 if let Some(user) = self.users.get(&user_id).cloned() {
635 return Task::ready(Ok(user));
636 }
637
638 let load_users = self.get_users(vec![user_id], cx);
639 cx.spawn(move |this, mut cx| async move {
640 load_users.await?;
641 this.update(&mut cx, |this, _| {
642 this.users
643 .get(&user_id)
644 .cloned()
645 .ok_or_else(|| anyhow!("server responded with no users"))
646 })?
647 })
648 }
649
650 pub fn cached_user_by_github_login(&self, github_login: &str) -> Option<Arc<User>> {
651 self.by_github_login
652 .get(github_login)
653 .and_then(|id| self.users.get(id).cloned())
654 }
655
656 pub fn current_user(&self) -> Option<Arc<User>> {
657 self.current_user.borrow().clone()
658 }
659
660 pub fn watch_current_user(&self) -> watch::Receiver<Option<Arc<User>>> {
661 self.current_user.clone()
662 }
663
664 fn load_users(
665 &mut self,
666 request: impl RequestMessage<Response = UsersResponse>,
667 cx: &mut ModelContext<Self>,
668 ) -> Task<Result<Vec<Arc<User>>>> {
669 let client = self.client.clone();
670 cx.spawn(|this, mut cx| async move {
671 if let Some(rpc) = client.upgrade() {
672 let response = rpc.request(request).await.context("error loading users")?;
673 let users = response.users;
674
675 this.update(&mut cx, |this, _| this.insert(users))
676 } else {
677 Ok(Vec::new())
678 }
679 })
680 }
681
682 pub fn insert(&mut self, users: Vec<proto::User>) -> Vec<Arc<User>> {
683 let mut ret = Vec::with_capacity(users.len());
684 for user in users {
685 let user = User::new(user);
686 if let Some(old) = self.users.insert(user.id, user.clone()) {
687 if old.github_login != user.github_login {
688 self.by_github_login.remove(&old.github_login);
689 }
690 }
691 self.by_github_login
692 .insert(user.github_login.clone(), user.id);
693 ret.push(user)
694 }
695 ret
696 }
697
698 pub fn set_participant_indices(
699 &mut self,
700 participant_indices: HashMap<u64, ParticipantIndex>,
701 cx: &mut ModelContext<Self>,
702 ) {
703 if participant_indices != self.participant_indices {
704 self.participant_indices = participant_indices;
705 cx.emit(Event::ParticipantIndicesChanged);
706 }
707 }
708
709 pub fn participant_indices(&self) -> &HashMap<u64, ParticipantIndex> {
710 &self.participant_indices
711 }
712
713 pub fn participant_names(
714 &self,
715 user_ids: impl Iterator<Item = u64>,
716 cx: &AppContext,
717 ) -> HashMap<u64, SharedString> {
718 let mut ret = HashMap::default();
719 let mut missing_user_ids = Vec::new();
720 for id in user_ids {
721 if let Some(github_login) = self.get_cached_user(id).map(|u| u.github_login.clone()) {
722 ret.insert(id, github_login.into());
723 } else {
724 missing_user_ids.push(id)
725 }
726 }
727 if !missing_user_ids.is_empty() {
728 let this = self.weak_self.clone();
729 cx.spawn(|mut cx| async move {
730 this.update(&mut cx, |this, cx| this.get_users(missing_user_ids, cx))?
731 .await
732 })
733 .detach_and_log_err(cx);
734 }
735 ret
736 }
737}
738
739impl User {
740 fn new(message: proto::User) -> Arc<Self> {
741 Arc::new(User {
742 id: message.id,
743 github_login: message.github_login,
744 avatar_uri: message.avatar_url.into(),
745 })
746 }
747}
748
749impl Contact {
750 async fn from_proto(
751 contact: proto::Contact,
752 user_store: &Model<UserStore>,
753 cx: &mut AsyncAppContext,
754 ) -> Result<Self> {
755 let user = user_store
756 .update(cx, |user_store, cx| {
757 user_store.get_user(contact.user_id, cx)
758 })?
759 .await?;
760 Ok(Self {
761 user,
762 online: contact.online,
763 busy: contact.busy,
764 })
765 }
766}
767
768impl Collaborator {
769 pub fn from_proto(message: proto::Collaborator) -> Result<Self> {
770 Ok(Self {
771 peer_id: message.peer_id.ok_or_else(|| anyhow!("invalid peer id"))?,
772 replica_id: message.replica_id as ReplicaId,
773 user_id: message.user_id as UserId,
774 })
775 }
776}