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