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