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