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