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, AsyncReadExt, Future, StreamExt};
6use gpui::{AsyncAppContext, Entity, ImageData, ModelContext, ModelHandle, Task};
7use postage::{sink::Sink, watch};
8use rpc::proto::{RequestMessage, UsersResponse};
9use std::sync::{Arc, Weak};
10use text::ReplicaId;
11use util::http::HttpClient;
12use util::TryFutureExt as _;
13
14pub type UserId = u64;
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub struct ParticipantIndex(pub u32);
18
19#[derive(Default, Debug)]
20pub struct User {
21 pub id: UserId,
22 pub github_login: String,
23 pub avatar: Option<Arc<ImageData>>,
24}
25
26#[derive(Clone, Debug, PartialEq, Eq)]
27pub struct Collaborator {
28 pub peer_id: proto::PeerId,
29 pub replica_id: ReplicaId,
30 pub user_id: UserId,
31}
32
33impl PartialOrd for User {
34 fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
35 Some(self.cmp(other))
36 }
37}
38
39impl Ord for User {
40 fn cmp(&self, other: &Self) -> std::cmp::Ordering {
41 self.github_login.cmp(&other.github_login)
42 }
43}
44
45impl PartialEq for User {
46 fn eq(&self, other: &Self) -> bool {
47 self.id == other.id && self.github_login == other.github_login
48 }
49}
50
51impl Eq for User {}
52
53#[derive(Debug, PartialEq)]
54pub struct Contact {
55 pub user: Arc<User>,
56 pub online: bool,
57 pub busy: bool,
58}
59
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
61pub enum ContactRequestStatus {
62 None,
63 RequestSent,
64 RequestReceived,
65 RequestAccepted,
66}
67
68pub struct UserStore {
69 users: HashMap<u64, Arc<User>>,
70 participant_indices: HashMap<u64, ParticipantIndex>,
71 update_contacts_tx: mpsc::UnboundedSender<UpdateContacts>,
72 current_user: watch::Receiver<Option<Arc<User>>>,
73 contacts: Vec<Arc<Contact>>,
74 incoming_contact_requests: Vec<Arc<User>>,
75 outgoing_contact_requests: Vec<Arc<User>>,
76 pending_contact_requests: HashMap<u64, usize>,
77 invite_info: Option<InviteInfo>,
78 client: Weak<Client>,
79 http: Arc<dyn HttpClient>,
80 _maintain_contacts: Task<()>,
81 _maintain_current_user: Task<()>,
82}
83
84#[derive(Clone)]
85pub struct InviteInfo {
86 pub count: u32,
87 pub url: Arc<str>,
88}
89
90pub enum Event {
91 Contact {
92 user: Arc<User>,
93 kind: ContactEventKind,
94 },
95 ShowContacts,
96 ParticipantIndicesChanged,
97}
98
99#[derive(Clone, Copy)]
100pub enum ContactEventKind {
101 Requested,
102 Accepted,
103 Cancelled,
104}
105
106impl Entity for UserStore {
107 type Event = Event;
108}
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(
118 client: Arc<Client>,
119 http: Arc<dyn HttpClient>,
120 cx: &mut ModelContext<Self>,
121 ) -> Self {
122 let (mut current_user_tx, current_user_rx) = watch::channel();
123 let (update_contacts_tx, mut update_contacts_rx) = mpsc::unbounded();
124 let rpc_subscriptions = vec![
125 client.add_message_handler(cx.handle(), Self::handle_update_contacts),
126 client.add_message_handler(cx.handle(), Self::handle_update_invite_info),
127 client.add_message_handler(cx.handle(), Self::handle_show_contacts),
128 ];
129 Self {
130 users: Default::default(),
131 current_user: current_user_rx,
132 contacts: Default::default(),
133 incoming_contact_requests: Default::default(),
134 participant_indices: Default::default(),
135 outgoing_contact_requests: Default::default(),
136 invite_info: None,
137 client: Arc::downgrade(&client),
138 update_contacts_tx,
139 http,
140 _maintain_contacts: cx.spawn_weak(|this, mut cx| async move {
141 let _subscriptions = rpc_subscriptions;
142 while let Some(message) = update_contacts_rx.next().await {
143 if let Some(this) = this.upgrade(&cx) {
144 this.update(&mut cx, |this, cx| this.update_contacts(message, cx))
145 .log_err()
146 .await;
147 }
148 }
149 }),
150 _maintain_current_user: cx.spawn_weak(|this, mut cx| async move {
151 let mut status = client.status();
152 while let Some(status) = status.next().await {
153 match status {
154 Status::Connected { .. } => {
155 if let Some((this, user_id)) = this.upgrade(&cx).zip(client.user_id()) {
156 let fetch_user = this
157 .update(&mut cx, |this, cx| this.get_user(user_id, cx))
158 .log_err();
159 let fetch_metrics_id =
160 client.request(proto::GetPrivateUserInfo {}).log_err();
161 let (user, info) = futures::join!(fetch_user, fetch_metrics_id);
162
163 if let Some(info) = info {
164 cx.update(|cx| {
165 cx.update_flags(info.staff, info.flags);
166 client.telemetry.set_authenticated_user_info(
167 Some(info.metrics_id.clone()),
168 info.staff,
169 cx,
170 )
171 });
172 } else {
173 cx.read(|cx| {
174 client
175 .telemetry
176 .set_authenticated_user_info(None, false, cx)
177 });
178 }
179
180 current_user_tx.send(user).await.ok();
181
182 this.update(&mut cx, |_, cx| {
183 cx.notify();
184 });
185 }
186 }
187 Status::SignedOut => {
188 current_user_tx.send(None).await.ok();
189 if let Some(this) = this.upgrade(&cx) {
190 this.update(&mut cx, |this, cx| {
191 cx.notify();
192 this.clear_contacts()
193 })
194 .await;
195 }
196 }
197 Status::ConnectionLost => {
198 if let Some(this) = this.upgrade(&cx) {
199 this.update(&mut cx, |this, cx| {
200 cx.notify();
201 this.clear_contacts()
202 })
203 .await;
204 }
205 }
206 _ => {}
207 }
208 }
209 }),
210 pending_contact_requests: Default::default(),
211 }
212 }
213
214 #[cfg(feature = "test-support")]
215 pub fn clear_cache(&mut self) {
216 self.users.clear();
217 }
218
219 async fn handle_update_invite_info(
220 this: ModelHandle<Self>,
221 message: TypedEnvelope<proto::UpdateInviteInfo>,
222 _: Arc<Client>,
223 mut cx: AsyncAppContext,
224 ) -> Result<()> {
225 this.update(&mut cx, |this, cx| {
226 this.invite_info = Some(InviteInfo {
227 url: Arc::from(message.payload.url),
228 count: message.payload.count,
229 });
230 cx.notify();
231 });
232 Ok(())
233 }
234
235 async fn handle_show_contacts(
236 this: ModelHandle<Self>,
237 _: TypedEnvelope<proto::ShowContacts>,
238 _: Arc<Client>,
239 mut cx: AsyncAppContext,
240 ) -> Result<()> {
241 this.update(&mut cx, |_, cx| cx.emit(Event::ShowContacts));
242 Ok(())
243 }
244
245 pub fn invite_info(&self) -> Option<&InviteInfo> {
246 self.invite_info.as_ref()
247 }
248
249 async fn handle_update_contacts(
250 this: ModelHandle<Self>,
251 message: TypedEnvelope<proto::UpdateContacts>,
252 _: Arc<Client>,
253 mut cx: AsyncAppContext,
254 ) -> Result<()> {
255 this.update(&mut cx, |this, _| {
256 this.update_contacts_tx
257 .unbounded_send(UpdateContacts::Update(message.payload))
258 .unwrap();
259 });
260 Ok(())
261 }
262
263 fn update_contacts(
264 &mut self,
265 message: UpdateContacts,
266 cx: &mut ModelContext<Self>,
267 ) -> Task<Result<()>> {
268 match message {
269 UpdateContacts::Wait(barrier) => {
270 drop(barrier);
271 Task::ready(Ok(()))
272 }
273 UpdateContacts::Clear(barrier) => {
274 self.contacts.clear();
275 self.incoming_contact_requests.clear();
276 self.outgoing_contact_requests.clear();
277 drop(barrier);
278 Task::ready(Ok(()))
279 }
280 UpdateContacts::Update(message) => {
281 let mut user_ids = HashSet::default();
282 for contact in &message.contacts {
283 user_ids.insert(contact.user_id);
284 }
285 user_ids.extend(message.incoming_requests.iter().map(|req| req.requester_id));
286 user_ids.extend(message.outgoing_requests.iter());
287
288 let load_users = self.get_users(user_ids.into_iter().collect(), cx);
289 cx.spawn(|this, mut cx| async move {
290 load_users.await?;
291
292 // Users are fetched in parallel above and cached in call to get_users
293 // No need to paralellize here
294 let mut updated_contacts = Vec::new();
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 has_incoming_contact_request(&self, user_id: u64) -> bool {
404 self.incoming_contact_requests
405 .iter()
406 .any(|user| user.id == user_id)
407 }
408
409 pub fn outgoing_contact_requests(&self) -> &[Arc<User>] {
410 &self.outgoing_contact_requests
411 }
412
413 pub fn is_contact_request_pending(&self, user: &User) -> bool {
414 self.pending_contact_requests.contains_key(&user.id)
415 }
416
417 pub fn contact_request_status(&self, user: &User) -> ContactRequestStatus {
418 if self
419 .contacts
420 .binary_search_by_key(&&user.github_login, |contact| &contact.user.github_login)
421 .is_ok()
422 {
423 ContactRequestStatus::RequestAccepted
424 } else if self
425 .outgoing_contact_requests
426 .binary_search_by_key(&&user.github_login, |user| &user.github_login)
427 .is_ok()
428 {
429 ContactRequestStatus::RequestSent
430 } else if self
431 .incoming_contact_requests
432 .binary_search_by_key(&&user.github_login, |user| &user.github_login)
433 .is_ok()
434 {
435 ContactRequestStatus::RequestReceived
436 } else {
437 ContactRequestStatus::None
438 }
439 }
440
441 pub fn request_contact(
442 &mut self,
443 responder_id: u64,
444 cx: &mut ModelContext<Self>,
445 ) -> Task<Result<()>> {
446 self.perform_contact_request(responder_id, proto::RequestContact { responder_id }, cx)
447 }
448
449 pub fn remove_contact(
450 &mut self,
451 user_id: u64,
452 cx: &mut ModelContext<Self>,
453 ) -> Task<Result<()>> {
454 self.perform_contact_request(user_id, proto::RemoveContact { user_id }, cx)
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_weak(|_, _| 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(|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.read_with(&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(
594 &mut self,
595 user_id: u64,
596 cx: &mut ModelContext<Self>,
597 ) -> Task<Result<Arc<User>>> {
598 if let Some(user) = self.users.get(&user_id).cloned() {
599 return cx.foreground().spawn(async move { Ok(user) });
600 }
601
602 let load_users = self.get_users(vec![user_id], cx);
603 cx.spawn(|this, mut cx| async move {
604 load_users.await?;
605 this.update(&mut cx, |this, _| {
606 this.users
607 .get(&user_id)
608 .cloned()
609 .ok_or_else(|| anyhow!("server responded with no users"))
610 })
611 })
612 }
613
614 pub fn current_user(&self) -> Option<Arc<User>> {
615 self.current_user.borrow().clone()
616 }
617
618 pub fn watch_current_user(&self) -> watch::Receiver<Option<Arc<User>>> {
619 self.current_user.clone()
620 }
621
622 fn load_users(
623 &mut self,
624 request: impl RequestMessage<Response = UsersResponse>,
625 cx: &mut ModelContext<Self>,
626 ) -> Task<Result<Vec<Arc<User>>>> {
627 let client = self.client.clone();
628 let http = self.http.clone();
629 cx.spawn_weak(|this, mut cx| async move {
630 if let Some(rpc) = client.upgrade() {
631 let response = rpc.request(request).await.context("error loading users")?;
632 let users = future::join_all(
633 response
634 .users
635 .into_iter()
636 .map(|user| User::new(user, http.as_ref())),
637 )
638 .await;
639
640 if let Some(this) = this.upgrade(&cx) {
641 this.update(&mut cx, |this, _| {
642 for user in &users {
643 this.users.insert(user.id, user.clone());
644 }
645 });
646 }
647 Ok(users)
648 } else {
649 Ok(Vec::new())
650 }
651 })
652 }
653
654 pub fn set_participant_indices(
655 &mut self,
656 participant_indices: HashMap<u64, ParticipantIndex>,
657 cx: &mut ModelContext<Self>,
658 ) {
659 if participant_indices != self.participant_indices {
660 self.participant_indices = participant_indices;
661 cx.emit(Event::ParticipantIndicesChanged);
662 }
663 }
664
665 pub fn participant_indices(&self) -> &HashMap<u64, ParticipantIndex> {
666 &self.participant_indices
667 }
668}
669
670impl User {
671 async fn new(message: proto::User, http: &dyn HttpClient) -> Arc<Self> {
672 Arc::new(User {
673 id: message.id,
674 github_login: message.github_login,
675 avatar: fetch_avatar(http, &message.avatar_url).warn_on_err().await,
676 })
677 }
678}
679
680impl Contact {
681 async fn from_proto(
682 contact: proto::Contact,
683 user_store: &ModelHandle<UserStore>,
684 cx: &mut AsyncAppContext,
685 ) -> Result<Self> {
686 let user = user_store
687 .update(cx, |user_store, cx| {
688 user_store.get_user(contact.user_id, cx)
689 })
690 .await?;
691 Ok(Self {
692 user,
693 online: contact.online,
694 busy: contact.busy,
695 })
696 }
697}
698
699impl Collaborator {
700 pub fn from_proto(message: proto::Collaborator) -> Result<Self> {
701 Ok(Self {
702 peer_id: message.peer_id.ok_or_else(|| anyhow!("invalid peer id"))?,
703 replica_id: message.replica_id as ReplicaId,
704 user_id: message.user_id as UserId,
705 })
706 }
707}
708
709async fn fetch_avatar(http: &dyn HttpClient, url: &str) -> Result<Arc<ImageData>> {
710 let mut response = http
711 .get(url, Default::default(), true)
712 .await
713 .map_err(|e| anyhow!("failed to send user avatar request: {}", e))?;
714
715 if !response.status().is_success() {
716 return Err(anyhow!("avatar request failed {:?}", response.status()));
717 }
718
719 let mut body = Vec::new();
720 response
721 .body_mut()
722 .read_to_end(&mut body)
723 .await
724 .map_err(|e| anyhow!("failed to read user avatar response body: {}", e))?;
725 let format = image::guess_format(&body)?;
726 let image = image::load_from_memory_with_format(&body, format)?.into_bgra8();
727 Ok(ImageData::new(image))
728}