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