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