1use crate::call::Call;
2
3use super::{http::HttpClient, proto, Client, Status, TypedEnvelope};
4use anyhow::{anyhow, Context, Result};
5use collections::{hash_map::Entry, BTreeSet, HashMap, HashSet};
6use futures::{channel::mpsc, future, AsyncReadExt, Future, Stream, StreamExt};
7use gpui::{AsyncAppContext, Entity, ImageData, ModelContext, ModelHandle, Task};
8use postage::{sink::Sink, watch};
9use rpc::proto::{RequestMessage, UsersResponse};
10use std::sync::{Arc, Weak};
11use util::TryFutureExt as _;
12
13#[derive(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 projects: Vec<ProjectMetadata>,
45}
46
47#[derive(Clone, Debug, PartialEq)]
48pub struct ProjectMetadata {
49 pub id: u64,
50 pub visible_worktree_root_names: Vec<String>,
51 pub guests: BTreeSet<Arc<User>>,
52}
53
54#[derive(Debug, Clone, Copy, PartialEq, Eq)]
55pub enum ContactRequestStatus {
56 None,
57 RequestSent,
58 RequestReceived,
59 RequestAccepted,
60}
61
62pub struct UserStore {
63 users: HashMap<u64, Arc<User>>,
64 update_contacts_tx: mpsc::UnboundedSender<UpdateContacts>,
65 current_user: watch::Receiver<Option<Arc<User>>>,
66 contacts: Vec<Arc<Contact>>,
67 incoming_contact_requests: Vec<Arc<User>>,
68 outgoing_contact_requests: Vec<Arc<User>>,
69 pending_contact_requests: HashMap<u64, usize>,
70 invite_info: Option<InviteInfo>,
71 incoming_calls: Vec<mpsc::UnboundedSender<Call>>,
72 client: Weak<Client>,
73 http: Arc<dyn HttpClient>,
74 _maintain_contacts: Task<()>,
75 _maintain_current_user: Task<()>,
76}
77
78#[derive(Clone)]
79pub struct InviteInfo {
80 pub count: u32,
81 pub url: Arc<str>,
82}
83
84pub enum Event {
85 Contact {
86 user: Arc<User>,
87 kind: ContactEventKind,
88 },
89 ShowContacts,
90}
91
92#[derive(Clone, Copy)]
93pub enum ContactEventKind {
94 Requested,
95 Accepted,
96 Cancelled,
97}
98
99impl Entity for UserStore {
100 type Event = Event;
101}
102
103enum UpdateContacts {
104 Update(proto::UpdateContacts),
105 Wait(postage::barrier::Sender),
106 Clear(postage::barrier::Sender),
107}
108
109impl UserStore {
110 pub fn new(
111 client: Arc<Client>,
112 http: Arc<dyn HttpClient>,
113 cx: &mut ModelContext<Self>,
114 ) -> Self {
115 let (mut current_user_tx, current_user_rx) = watch::channel();
116 let (update_contacts_tx, mut update_contacts_rx) = mpsc::unbounded();
117 let rpc_subscriptions = vec![
118 client.add_message_handler(cx.handle(), Self::handle_update_contacts),
119 client.add_message_handler(cx.handle(), Self::handle_update_invite_info),
120 client.add_message_handler(cx.handle(), Self::handle_show_contacts),
121 client.add_request_handler(cx.handle(), Self::handle_incoming_call),
122 ];
123 Self {
124 users: Default::default(),
125 current_user: current_user_rx,
126 contacts: Default::default(),
127 incoming_contact_requests: Default::default(),
128 outgoing_contact_requests: Default::default(),
129 invite_info: None,
130 incoming_calls: Default::default(),
131 client: Arc::downgrade(&client),
132 update_contacts_tx,
133 http,
134 _maintain_contacts: cx.spawn_weak(|this, mut cx| async move {
135 let _subscriptions = rpc_subscriptions;
136 while let Some(message) = update_contacts_rx.next().await {
137 if let Some(this) = this.upgrade(&cx) {
138 this.update(&mut cx, |this, cx| this.update_contacts(message, cx))
139 .log_err()
140 .await;
141 }
142 }
143 }),
144 _maintain_current_user: cx.spawn_weak(|this, mut cx| async move {
145 let mut status = client.status();
146 while let Some(status) = status.next().await {
147 match status {
148 Status::Connected { .. } => {
149 if let Some((this, user_id)) = this.upgrade(&cx).zip(client.user_id()) {
150 let user = this
151 .update(&mut cx, |this, cx| this.get_user(user_id, cx))
152 .log_err()
153 .await;
154 current_user_tx.send(user).await.ok();
155 }
156 }
157 Status::SignedOut => {
158 current_user_tx.send(None).await.ok();
159 if let Some(this) = this.upgrade(&cx) {
160 this.update(&mut cx, |this, _| this.clear_contacts()).await;
161 }
162 }
163 Status::ConnectionLost => {
164 if let Some(this) = this.upgrade(&cx) {
165 this.update(&mut cx, |this, _| this.clear_contacts()).await;
166 }
167 }
168 _ => {}
169 }
170 }
171 }),
172 pending_contact_requests: Default::default(),
173 }
174 }
175
176 async fn handle_update_invite_info(
177 this: ModelHandle<Self>,
178 message: TypedEnvelope<proto::UpdateInviteInfo>,
179 _: Arc<Client>,
180 mut cx: AsyncAppContext,
181 ) -> Result<()> {
182 this.update(&mut cx, |this, cx| {
183 this.invite_info = Some(InviteInfo {
184 url: Arc::from(message.payload.url),
185 count: message.payload.count,
186 });
187 cx.notify();
188 });
189 Ok(())
190 }
191
192 async fn handle_show_contacts(
193 this: ModelHandle<Self>,
194 _: TypedEnvelope<proto::ShowContacts>,
195 _: Arc<Client>,
196 mut cx: AsyncAppContext,
197 ) -> Result<()> {
198 this.update(&mut cx, |_, cx| cx.emit(Event::ShowContacts));
199 Ok(())
200 }
201
202 async fn handle_incoming_call(
203 this: ModelHandle<Self>,
204 envelope: TypedEnvelope<proto::IncomingCall>,
205 _: Arc<Client>,
206 mut cx: AsyncAppContext,
207 ) -> Result<proto::Ack> {
208 let call = Call {
209 room_id: envelope.payload.room_id,
210 participants: this
211 .update(&mut cx, |this, cx| {
212 this.get_users(envelope.payload.participant_user_ids, cx)
213 })
214 .await?,
215 from: this
216 .update(&mut cx, |this, cx| {
217 this.get_user(envelope.payload.from_user_id, cx)
218 })
219 .await?,
220 };
221 this.update(&mut cx, |this, _| {
222 this.incoming_calls
223 .retain(|tx| tx.unbounded_send(call.clone()).is_ok());
224 });
225
226 Ok(proto::Ack {})
227 }
228
229 pub fn invite_info(&self) -> Option<&InviteInfo> {
230 self.invite_info.as_ref()
231 }
232
233 pub fn incoming_calls(&mut self) -> impl 'static + Stream<Item = Call> {
234 let (tx, rx) = mpsc::unbounded();
235 self.incoming_calls.push(tx);
236 rx
237 }
238
239 async fn handle_update_contacts(
240 this: ModelHandle<Self>,
241 message: TypedEnvelope<proto::UpdateContacts>,
242 _: Arc<Client>,
243 mut cx: AsyncAppContext,
244 ) -> Result<()> {
245 this.update(&mut cx, |this, _| {
246 this.update_contacts_tx
247 .unbounded_send(UpdateContacts::Update(message.payload))
248 .unwrap();
249 });
250 Ok(())
251 }
252
253 fn update_contacts(
254 &mut self,
255 message: UpdateContacts,
256 cx: &mut ModelContext<Self>,
257 ) -> Task<Result<()>> {
258 match message {
259 UpdateContacts::Wait(barrier) => {
260 drop(barrier);
261 Task::ready(Ok(()))
262 }
263 UpdateContacts::Clear(barrier) => {
264 self.contacts.clear();
265 self.incoming_contact_requests.clear();
266 self.outgoing_contact_requests.clear();
267 drop(barrier);
268 Task::ready(Ok(()))
269 }
270 UpdateContacts::Update(message) => {
271 let mut user_ids = HashSet::default();
272 for contact in &message.contacts {
273 user_ids.insert(contact.user_id);
274 user_ids.extend(contact.projects.iter().flat_map(|w| &w.guests).copied());
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 for contact in message.contacts {
287 let should_notify = contact.should_notify;
288 updated_contacts.push((
289 Arc::new(Contact::from_proto(contact, &this, &mut cx).await?),
290 should_notify,
291 ));
292 }
293
294 let mut incoming_requests = Vec::new();
295 for request in message.incoming_requests {
296 incoming_requests.push({
297 let user = this
298 .update(&mut cx, |this, cx| this.get_user(request.requester_id, cx))
299 .await?;
300 (user, request.should_notify)
301 });
302 }
303
304 let mut outgoing_requests = Vec::new();
305 for requested_user_id in message.outgoing_requests {
306 outgoing_requests.push(
307 this.update(&mut cx, |this, cx| this.get_user(requested_user_id, cx))
308 .await?,
309 );
310 }
311
312 let removed_contacts =
313 HashSet::<u64>::from_iter(message.remove_contacts.iter().copied());
314 let removed_incoming_requests =
315 HashSet::<u64>::from_iter(message.remove_incoming_requests.iter().copied());
316 let removed_outgoing_requests =
317 HashSet::<u64>::from_iter(message.remove_outgoing_requests.iter().copied());
318
319 this.update(&mut cx, |this, cx| {
320 // Remove contacts
321 this.contacts
322 .retain(|contact| !removed_contacts.contains(&contact.user.id));
323 // Update existing contacts and insert new ones
324 for (updated_contact, should_notify) in updated_contacts {
325 if should_notify {
326 cx.emit(Event::Contact {
327 user: updated_contact.user.clone(),
328 kind: ContactEventKind::Accepted,
329 });
330 }
331 match this.contacts.binary_search_by_key(
332 &&updated_contact.user.github_login,
333 |contact| &contact.user.github_login,
334 ) {
335 Ok(ix) => this.contacts[ix] = updated_contact,
336 Err(ix) => this.contacts.insert(ix, updated_contact),
337 }
338 }
339
340 // Remove incoming contact requests
341 this.incoming_contact_requests.retain(|user| {
342 if removed_incoming_requests.contains(&user.id) {
343 cx.emit(Event::Contact {
344 user: user.clone(),
345 kind: ContactEventKind::Cancelled,
346 });
347 false
348 } else {
349 true
350 }
351 });
352 // Update existing incoming requests and insert new ones
353 for (user, should_notify) in incoming_requests {
354 if should_notify {
355 cx.emit(Event::Contact {
356 user: user.clone(),
357 kind: ContactEventKind::Requested,
358 });
359 }
360
361 match this
362 .incoming_contact_requests
363 .binary_search_by_key(&&user.github_login, |contact| {
364 &contact.github_login
365 }) {
366 Ok(ix) => this.incoming_contact_requests[ix] = user,
367 Err(ix) => this.incoming_contact_requests.insert(ix, user),
368 }
369 }
370
371 // Remove outgoing contact requests
372 this.outgoing_contact_requests
373 .retain(|user| !removed_outgoing_requests.contains(&user.id));
374 // Update existing incoming requests and insert new ones
375 for request in outgoing_requests {
376 match this
377 .outgoing_contact_requests
378 .binary_search_by_key(&&request.github_login, |contact| {
379 &contact.github_login
380 }) {
381 Ok(ix) => this.outgoing_contact_requests[ix] = request,
382 Err(ix) => this.outgoing_contact_requests.insert(ix, request),
383 }
384 }
385
386 cx.notify();
387 });
388
389 Ok(())
390 })
391 }
392 }
393 }
394
395 pub fn contacts(&self) -> &[Arc<Contact>] {
396 &self.contacts
397 }
398
399 pub fn has_contact(&self, user: &Arc<User>) -> bool {
400 self.contacts
401 .binary_search_by_key(&&user.github_login, |contact| &contact.user.github_login)
402 .is_ok()
403 }
404
405 pub fn incoming_contact_requests(&self) -> &[Arc<User>] {
406 &self.incoming_contact_requests
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_user(
590 &mut self,
591 user_id: u64,
592 cx: &mut ModelContext<Self>,
593 ) -> Task<Result<Arc<User>>> {
594 if let Some(user) = self.users.get(&user_id).cloned() {
595 return cx.foreground().spawn(async move { Ok(user) });
596 }
597
598 let load_users = self.get_users(vec![user_id], cx);
599 cx.spawn(|this, mut cx| async move {
600 load_users.await?;
601 this.update(&mut cx, |this, _| {
602 this.users
603 .get(&user_id)
604 .cloned()
605 .ok_or_else(|| anyhow!("server responded with no users"))
606 })
607 })
608 }
609
610 pub fn current_user(&self) -> Option<Arc<User>> {
611 self.current_user.borrow().clone()
612 }
613
614 pub fn watch_current_user(&self) -> watch::Receiver<Option<Arc<User>>> {
615 self.current_user.clone()
616 }
617
618 fn load_users(
619 &mut self,
620 request: impl RequestMessage<Response = UsersResponse>,
621 cx: &mut ModelContext<Self>,
622 ) -> Task<Result<Vec<Arc<User>>>> {
623 let client = self.client.clone();
624 let http = self.http.clone();
625 cx.spawn_weak(|this, mut cx| async move {
626 if let Some(rpc) = client.upgrade() {
627 let response = rpc.request(request).await.context("error loading users")?;
628 let users = future::join_all(
629 response
630 .users
631 .into_iter()
632 .map(|user| User::new(user, http.as_ref())),
633 )
634 .await;
635
636 if let Some(this) = this.upgrade(&cx) {
637 this.update(&mut cx, |this, _| {
638 for user in &users {
639 this.users.insert(user.id, user.clone());
640 }
641 });
642 }
643 Ok(users)
644 } else {
645 Ok(Vec::new())
646 }
647 })
648 }
649}
650
651impl User {
652 async fn new(message: proto::User, http: &dyn HttpClient) -> Arc<Self> {
653 Arc::new(User {
654 id: message.id,
655 github_login: message.github_login,
656 avatar: fetch_avatar(http, &message.avatar_url).warn_on_err().await,
657 })
658 }
659}
660
661impl Contact {
662 async fn from_proto(
663 contact: proto::Contact,
664 user_store: &ModelHandle<UserStore>,
665 cx: &mut AsyncAppContext,
666 ) -> Result<Self> {
667 let user = user_store
668 .update(cx, |user_store, cx| {
669 user_store.get_user(contact.user_id, cx)
670 })
671 .await?;
672 let mut projects = Vec::new();
673 for project in contact.projects {
674 let mut guests = BTreeSet::new();
675 for participant_id in project.guests {
676 guests.insert(
677 user_store
678 .update(cx, |user_store, cx| user_store.get_user(participant_id, cx))
679 .await?,
680 );
681 }
682 projects.push(ProjectMetadata {
683 id: project.id,
684 visible_worktree_root_names: project.visible_worktree_root_names.clone(),
685 guests,
686 });
687 }
688 Ok(Self {
689 user,
690 online: contact.online,
691 projects,
692 })
693 }
694
695 pub fn non_empty_projects(&self) -> impl Iterator<Item = &ProjectMetadata> {
696 self.projects
697 .iter()
698 .filter(|project| !project.visible_worktree_root_names.is_empty())
699 }
700}
701
702async fn fetch_avatar(http: &dyn HttpClient, url: &str) -> Result<Arc<ImageData>> {
703 let mut response = http
704 .get(url, Default::default(), true)
705 .await
706 .map_err(|e| anyhow!("failed to send user avatar request: {}", e))?;
707
708 if !response.status().is_success() {
709 return Err(anyhow!("avatar request failed {:?}", response.status()));
710 }
711
712 let mut body = Vec::new();
713 response
714 .body_mut()
715 .read_to_end(&mut body)
716 .await
717 .map_err(|e| anyhow!("failed to read user avatar response body: {}", e))?;
718 let format = image::guess_format(&body)?;
719 let image = image::load_from_memory_with_format(&body, format)?.into_bgra8();
720 Ok(ImageData::new(image))
721}