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