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