1use super::{Client, Status, TypedEnvelope, proto};
2use anyhow::{Context as _, Result};
3use chrono::{DateTime, Utc};
4use cloud_api_client::websocket_protocol::MessageToClient;
5use cloud_api_client::{GetAuthenticatedUserResponse, Plan, PlanInfo};
6use cloud_llm_client::{
7 EDIT_PREDICTIONS_USAGE_AMOUNT_HEADER_NAME, EDIT_PREDICTIONS_USAGE_LIMIT_HEADER_NAME, UsageLimit,
8};
9use collections::{HashMap, HashSet, hash_map::Entry};
10use derive_more::Deref;
11use feature_flags::FeatureFlagAppExt;
12use futures::{Future, StreamExt, channel::mpsc};
13use gpui::{
14 App, AsyncApp, Context, Entity, EventEmitter, SharedString, SharedUri, Task, WeakEntity,
15};
16use http_client::http::{HeaderMap, HeaderValue};
17use postage::{sink::Sink, watch};
18use rpc::proto::{RequestMessage, UsersResponse};
19use std::{
20 str::FromStr as _,
21 sync::{Arc, Weak},
22};
23use text::ReplicaId;
24use util::{ResultExt, TryFutureExt as _};
25
26pub type UserId = u64;
27
28#[derive(
29 Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Copy, serde::Serialize, serde::Deserialize,
30)]
31pub struct ChannelId(pub u64);
32
33impl std::fmt::Display for ChannelId {
34 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
35 self.0.fmt(f)
36 }
37}
38
39#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Copy)]
40pub struct ProjectId(pub u64);
41
42impl ProjectId {
43 pub fn to_proto(self) -> u64 {
44 self.0
45 }
46}
47
48#[derive(Debug, Clone, Copy, PartialEq, Eq)]
49pub struct ParticipantIndex(pub u32);
50
51#[derive(Default, Debug)]
52pub struct User {
53 pub id: UserId,
54 pub github_login: SharedString,
55 pub avatar_uri: SharedUri,
56 pub name: Option<String>,
57}
58
59#[derive(Clone, Debug, PartialEq, Eq)]
60pub struct Collaborator {
61 pub peer_id: proto::PeerId,
62 pub replica_id: ReplicaId,
63 pub user_id: UserId,
64 pub is_host: bool,
65 pub committer_name: Option<String>,
66 pub committer_email: Option<String>,
67}
68
69impl PartialOrd for User {
70 fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
71 Some(self.cmp(other))
72 }
73}
74
75impl Ord for User {
76 fn cmp(&self, other: &Self) -> std::cmp::Ordering {
77 self.github_login.cmp(&other.github_login)
78 }
79}
80
81impl PartialEq for User {
82 fn eq(&self, other: &Self) -> bool {
83 self.id == other.id && self.github_login == other.github_login
84 }
85}
86
87impl Eq for User {}
88
89#[derive(Debug, PartialEq)]
90pub struct Contact {
91 pub user: Arc<User>,
92 pub online: bool,
93 pub busy: bool,
94}
95
96#[derive(Debug, Clone, Copy, PartialEq, Eq)]
97pub enum ContactRequestStatus {
98 None,
99 RequestSent,
100 RequestReceived,
101 RequestAccepted,
102}
103
104pub struct UserStore {
105 users: HashMap<u64, Arc<User>>,
106 by_github_login: HashMap<SharedString, u64>,
107 participant_indices: HashMap<u64, ParticipantIndex>,
108 update_contacts_tx: mpsc::UnboundedSender<UpdateContacts>,
109 edit_prediction_usage: Option<EditPredictionUsage>,
110 plan_info: Option<PlanInfo>,
111 current_user: watch::Receiver<Option<Arc<User>>>,
112 contacts: Vec<Arc<Contact>>,
113 incoming_contact_requests: Vec<Arc<User>>,
114 outgoing_contact_requests: Vec<Arc<User>>,
115 pending_contact_requests: HashMap<u64, usize>,
116 client: Weak<Client>,
117 _maintain_contacts: Task<()>,
118 _maintain_current_user: Task<Result<()>>,
119 weak_self: WeakEntity<Self>,
120}
121
122#[derive(Clone)]
123pub struct InviteInfo {
124 pub count: u32,
125 pub url: Arc<str>,
126}
127
128pub enum Event {
129 Contact {
130 user: Arc<User>,
131 kind: ContactEventKind,
132 },
133 ShowContacts,
134 ParticipantIndicesChanged,
135 PrivateUserInfoUpdated,
136 PlanUpdated,
137}
138
139#[derive(Clone, Copy)]
140pub enum ContactEventKind {
141 Requested,
142 Accepted,
143 Cancelled,
144}
145
146impl EventEmitter<Event> for UserStore {}
147
148enum UpdateContacts {
149 Update(proto::UpdateContacts),
150 Wait(postage::barrier::Sender),
151 Clear(postage::barrier::Sender),
152}
153
154#[derive(Debug, Clone, Copy, Deref)]
155pub struct EditPredictionUsage(pub RequestUsage);
156
157#[derive(Debug, Clone, Copy)]
158pub struct RequestUsage {
159 pub limit: UsageLimit,
160 pub amount: i32,
161}
162
163impl UserStore {
164 pub fn new(client: Arc<Client>, cx: &Context<Self>) -> Self {
165 let (mut current_user_tx, current_user_rx) = watch::channel();
166 let (update_contacts_tx, mut update_contacts_rx) = mpsc::unbounded();
167 let rpc_subscriptions = vec![
168 client.add_message_handler(cx.weak_entity(), Self::handle_update_contacts),
169 client.add_message_handler(cx.weak_entity(), Self::handle_show_contacts),
170 ];
171
172 client.add_message_to_client_handler({
173 let this = cx.weak_entity();
174 move |message, cx| Self::handle_message_to_client(this.clone(), message, cx)
175 });
176
177 Self {
178 users: Default::default(),
179 by_github_login: Default::default(),
180 current_user: current_user_rx,
181 plan_info: None,
182 edit_prediction_usage: None,
183 contacts: Default::default(),
184 incoming_contact_requests: Default::default(),
185 participant_indices: Default::default(),
186 outgoing_contact_requests: Default::default(),
187 client: Arc::downgrade(&client),
188 update_contacts_tx,
189 _maintain_contacts: cx.spawn(async move |this, cx| {
190 let _subscriptions = rpc_subscriptions;
191 while let Some(message) = update_contacts_rx.next().await {
192 if let Ok(task) = this.update(cx, |this, cx| this.update_contacts(message, cx))
193 {
194 task.log_err().await;
195 } else {
196 break;
197 }
198 }
199 }),
200 _maintain_current_user: cx.spawn(async move |this, cx| {
201 let mut status = client.status();
202 let weak = Arc::downgrade(&client);
203 drop(client);
204 while let Some(status) = status.next().await {
205 // if the client is dropped, the app is shutting down.
206 let Some(client) = weak.upgrade() else {
207 return Ok(());
208 };
209 match status {
210 Status::Authenticated
211 | Status::Reauthenticated
212 | Status::Connected { .. } => {
213 if let Some(user_id) = client.user_id() {
214 let response = client
215 .cloud_client()
216 .get_authenticated_user()
217 .await
218 .log_err();
219
220 let current_user_and_response = if let Some(response) = response {
221 let user = Arc::new(User {
222 id: user_id,
223 github_login: response.user.github_login.clone().into(),
224 avatar_uri: response.user.avatar_url.clone().into(),
225 name: response.user.name.clone(),
226 });
227
228 Some((user, response))
229 } else {
230 None
231 };
232 current_user_tx
233 .send(
234 current_user_and_response
235 .as_ref()
236 .map(|(user, _)| user.clone()),
237 )
238 .await
239 .ok();
240
241 cx.update(|cx| {
242 if let Some((user, response)) = current_user_and_response {
243 this.update(cx, |this, cx| {
244 this.by_github_login
245 .insert(user.github_login.clone(), user_id);
246 this.users.insert(user_id, user);
247 this.update_authenticated_user(response, cx)
248 })
249 } else {
250 anyhow::Ok(())
251 }
252 })?;
253
254 this.update(cx, |_, cx| cx.notify())?;
255 }
256 }
257 Status::SignedOut => {
258 current_user_tx.send(None).await.ok();
259 this.update(cx, |this, cx| {
260 this.clear_plan_and_usage();
261 cx.emit(Event::PrivateUserInfoUpdated);
262 cx.notify();
263 this.clear_contacts()
264 })?
265 .await;
266 }
267 Status::ConnectionLost => {
268 this.update(cx, |this, cx| {
269 cx.notify();
270 this.clear_contacts()
271 })?
272 .await;
273 }
274 _ => {}
275 }
276 }
277 Ok(())
278 }),
279 pending_contact_requests: Default::default(),
280 weak_self: cx.weak_entity(),
281 }
282 }
283
284 #[cfg(feature = "test-support")]
285 pub fn clear_cache(&mut self) {
286 self.users.clear();
287 self.by_github_login.clear();
288 }
289
290 async fn handle_show_contacts(
291 this: Entity<Self>,
292 _: TypedEnvelope<proto::ShowContacts>,
293 mut cx: AsyncApp,
294 ) -> Result<()> {
295 this.update(&mut cx, |_, cx| cx.emit(Event::ShowContacts));
296 Ok(())
297 }
298
299 async fn handle_update_contacts(
300 this: Entity<Self>,
301 message: TypedEnvelope<proto::UpdateContacts>,
302 cx: AsyncApp,
303 ) -> Result<()> {
304 this.read_with(&cx, |this, _| {
305 this.update_contacts_tx
306 .unbounded_send(UpdateContacts::Update(message.payload))
307 .unwrap();
308 });
309 Ok(())
310 }
311
312 fn update_contacts(&mut self, message: UpdateContacts, cx: &Context<Self>) -> Task<Result<()>> {
313 match message {
314 UpdateContacts::Wait(barrier) => {
315 drop(barrier);
316 Task::ready(Ok(()))
317 }
318 UpdateContacts::Clear(barrier) => {
319 self.contacts.clear();
320 self.incoming_contact_requests.clear();
321 self.outgoing_contact_requests.clear();
322 drop(barrier);
323 Task::ready(Ok(()))
324 }
325 UpdateContacts::Update(message) => {
326 let mut user_ids = HashSet::default();
327 for contact in &message.contacts {
328 user_ids.insert(contact.user_id);
329 }
330 user_ids.extend(message.incoming_requests.iter().map(|req| req.requester_id));
331 user_ids.extend(message.outgoing_requests.iter());
332
333 let load_users = self.get_users(user_ids.into_iter().collect(), cx);
334 cx.spawn(async move |this, cx| {
335 load_users.await?;
336
337 // Users are fetched in parallel above and cached in call to get_users
338 // No need to parallelize here
339 let mut updated_contacts = Vec::new();
340 let this = this.upgrade().context("can't upgrade user store handle")?;
341 for contact in message.contacts {
342 updated_contacts
343 .push(Arc::new(Contact::from_proto(contact, &this, cx).await?));
344 }
345
346 let mut incoming_requests = Vec::new();
347 for request in message.incoming_requests {
348 incoming_requests.push({
349 this.update(cx, |this, cx| this.get_user(request.requester_id, cx))
350 .await?
351 });
352 }
353
354 let mut outgoing_requests = Vec::new();
355 for requested_user_id in message.outgoing_requests {
356 outgoing_requests.push(
357 this.update(cx, |this, cx| this.get_user(requested_user_id, cx))
358 .await?,
359 );
360 }
361
362 let removed_contacts =
363 HashSet::<u64>::from_iter(message.remove_contacts.iter().copied());
364 let removed_incoming_requests =
365 HashSet::<u64>::from_iter(message.remove_incoming_requests.iter().copied());
366 let removed_outgoing_requests =
367 HashSet::<u64>::from_iter(message.remove_outgoing_requests.iter().copied());
368
369 this.update(cx, |this, cx| {
370 // Remove contacts
371 this.contacts
372 .retain(|contact| !removed_contacts.contains(&contact.user.id));
373 // Update existing contacts and insert new ones
374 for updated_contact in updated_contacts {
375 match this.contacts.binary_search_by_key(
376 &&updated_contact.user.github_login,
377 |contact| &contact.user.github_login,
378 ) {
379 Ok(ix) => this.contacts[ix] = updated_contact,
380 Err(ix) => this.contacts.insert(ix, updated_contact),
381 }
382 }
383
384 // Remove incoming contact requests
385 this.incoming_contact_requests.retain(|user| {
386 if removed_incoming_requests.contains(&user.id) {
387 cx.emit(Event::Contact {
388 user: user.clone(),
389 kind: ContactEventKind::Cancelled,
390 });
391 false
392 } else {
393 true
394 }
395 });
396 // Update existing incoming requests and insert new ones
397 for user in incoming_requests {
398 match this
399 .incoming_contact_requests
400 .binary_search_by_key(&&user.github_login, |contact| {
401 &contact.github_login
402 }) {
403 Ok(ix) => this.incoming_contact_requests[ix] = user,
404 Err(ix) => this.incoming_contact_requests.insert(ix, user),
405 }
406 }
407
408 // Remove outgoing contact requests
409 this.outgoing_contact_requests
410 .retain(|user| !removed_outgoing_requests.contains(&user.id));
411 // Update existing incoming requests and insert new ones
412 for request in outgoing_requests {
413 match this
414 .outgoing_contact_requests
415 .binary_search_by_key(&&request.github_login, |contact| {
416 &contact.github_login
417 }) {
418 Ok(ix) => this.outgoing_contact_requests[ix] = request,
419 Err(ix) => this.outgoing_contact_requests.insert(ix, request),
420 }
421 }
422
423 cx.notify();
424 });
425
426 Ok(())
427 })
428 }
429 }
430 }
431
432 pub fn contacts(&self) -> &[Arc<Contact>] {
433 &self.contacts
434 }
435
436 pub fn has_contact(&self, user: &Arc<User>) -> bool {
437 self.contacts
438 .binary_search_by_key(&&user.github_login, |contact| &contact.user.github_login)
439 .is_ok()
440 }
441
442 pub fn incoming_contact_requests(&self) -> &[Arc<User>] {
443 &self.incoming_contact_requests
444 }
445
446 pub fn outgoing_contact_requests(&self) -> &[Arc<User>] {
447 &self.outgoing_contact_requests
448 }
449
450 pub fn is_contact_request_pending(&self, user: &User) -> bool {
451 self.pending_contact_requests.contains_key(&user.id)
452 }
453
454 pub fn contact_request_status(&self, user: &User) -> ContactRequestStatus {
455 if self
456 .contacts
457 .binary_search_by_key(&&user.github_login, |contact| &contact.user.github_login)
458 .is_ok()
459 {
460 ContactRequestStatus::RequestAccepted
461 } else if self
462 .outgoing_contact_requests
463 .binary_search_by_key(&&user.github_login, |user| &user.github_login)
464 .is_ok()
465 {
466 ContactRequestStatus::RequestSent
467 } else if self
468 .incoming_contact_requests
469 .binary_search_by_key(&&user.github_login, |user| &user.github_login)
470 .is_ok()
471 {
472 ContactRequestStatus::RequestReceived
473 } else {
474 ContactRequestStatus::None
475 }
476 }
477
478 pub fn request_contact(
479 &mut self,
480 responder_id: u64,
481 cx: &mut Context<Self>,
482 ) -> Task<Result<()>> {
483 self.perform_contact_request(responder_id, proto::RequestContact { responder_id }, cx)
484 }
485
486 pub fn remove_contact(&mut self, user_id: u64, cx: &mut Context<Self>) -> Task<Result<()>> {
487 self.perform_contact_request(user_id, proto::RemoveContact { user_id }, cx)
488 }
489
490 pub fn has_incoming_contact_request(&self, user_id: u64) -> bool {
491 self.incoming_contact_requests
492 .iter()
493 .any(|user| user.id == user_id)
494 }
495
496 pub fn respond_to_contact_request(
497 &mut self,
498 requester_id: u64,
499 accept: bool,
500 cx: &mut Context<Self>,
501 ) -> Task<Result<()>> {
502 self.perform_contact_request(
503 requester_id,
504 proto::RespondToContactRequest {
505 requester_id,
506 response: if accept {
507 proto::ContactRequestResponse::Accept
508 } else {
509 proto::ContactRequestResponse::Decline
510 } as i32,
511 },
512 cx,
513 )
514 }
515
516 pub fn dismiss_contact_request(
517 &self,
518 requester_id: u64,
519 cx: &Context<Self>,
520 ) -> Task<Result<()>> {
521 let client = self.client.upgrade();
522 cx.spawn(async move |_, _| {
523 client
524 .context("can't upgrade client reference")?
525 .request(proto::RespondToContactRequest {
526 requester_id,
527 response: proto::ContactRequestResponse::Dismiss as i32,
528 })
529 .await?;
530 Ok(())
531 })
532 }
533
534 fn perform_contact_request<T: RequestMessage>(
535 &mut self,
536 user_id: u64,
537 request: T,
538 cx: &mut Context<Self>,
539 ) -> Task<Result<()>> {
540 let client = self.client.upgrade();
541 *self.pending_contact_requests.entry(user_id).or_insert(0) += 1;
542 cx.notify();
543
544 cx.spawn(async move |this, cx| {
545 let response = client
546 .context("can't upgrade client reference")?
547 .request(request)
548 .await;
549 this.update(cx, |this, cx| {
550 if let Entry::Occupied(mut request_count) =
551 this.pending_contact_requests.entry(user_id)
552 {
553 *request_count.get_mut() -= 1;
554 if *request_count.get() == 0 {
555 request_count.remove();
556 }
557 }
558 cx.notify();
559 })?;
560 response?;
561 Ok(())
562 })
563 }
564
565 pub fn clear_contacts(&self) -> impl Future<Output = ()> + use<> {
566 let (tx, mut rx) = postage::barrier::channel();
567 self.update_contacts_tx
568 .unbounded_send(UpdateContacts::Clear(tx))
569 .unwrap();
570 async move {
571 rx.next().await;
572 }
573 }
574
575 pub fn contact_updates_done(&self) -> impl Future<Output = ()> {
576 let (tx, mut rx) = postage::barrier::channel();
577 self.update_contacts_tx
578 .unbounded_send(UpdateContacts::Wait(tx))
579 .unwrap();
580 async move {
581 rx.next().await;
582 }
583 }
584
585 pub fn get_users(
586 &self,
587 user_ids: Vec<u64>,
588 cx: &Context<Self>,
589 ) -> Task<Result<Vec<Arc<User>>>> {
590 let mut user_ids_to_fetch = user_ids.clone();
591 user_ids_to_fetch.retain(|id| !self.users.contains_key(id));
592
593 cx.spawn(async move |this, cx| {
594 if !user_ids_to_fetch.is_empty() {
595 this.update(cx, |this, cx| {
596 this.load_users(
597 proto::GetUsers {
598 user_ids: user_ids_to_fetch,
599 },
600 cx,
601 )
602 })?
603 .await?;
604 }
605
606 this.read_with(cx, |this, _| {
607 user_ids
608 .iter()
609 .map(|user_id| {
610 this.users
611 .get(user_id)
612 .cloned()
613 .with_context(|| format!("user {user_id} not found"))
614 })
615 .collect()
616 })?
617 })
618 }
619
620 pub fn fuzzy_search_users(
621 &self,
622 query: String,
623 cx: &Context<Self>,
624 ) -> Task<Result<Vec<Arc<User>>>> {
625 self.load_users(proto::FuzzySearchUsers { query }, cx)
626 }
627
628 pub fn get_cached_user(&self, user_id: u64) -> Option<Arc<User>> {
629 self.users.get(&user_id).cloned()
630 }
631
632 pub fn get_user_optimistic(&self, user_id: u64, cx: &Context<Self>) -> Option<Arc<User>> {
633 if let Some(user) = self.users.get(&user_id).cloned() {
634 return Some(user);
635 }
636
637 self.get_user(user_id, cx).detach_and_log_err(cx);
638 None
639 }
640
641 pub fn get_user(&self, user_id: u64, cx: &Context<Self>) -> Task<Result<Arc<User>>> {
642 if let Some(user) = self.users.get(&user_id).cloned() {
643 return Task::ready(Ok(user));
644 }
645
646 let load_users = self.get_users(vec![user_id], cx);
647 cx.spawn(async move |this, cx| {
648 load_users.await?;
649 this.read_with(cx, |this, _| {
650 this.users
651 .get(&user_id)
652 .cloned()
653 .context("server responded with no users")
654 })?
655 })
656 }
657
658 pub fn cached_user_by_github_login(&self, github_login: &str) -> Option<Arc<User>> {
659 self.by_github_login
660 .get(github_login)
661 .and_then(|id| self.users.get(id).cloned())
662 }
663
664 pub fn current_user(&self) -> Option<Arc<User>> {
665 self.current_user.borrow().clone()
666 }
667
668 pub fn plan(&self) -> Option<Plan> {
669 #[cfg(debug_assertions)]
670 if let Ok(plan) = std::env::var("ZED_SIMULATE_PLAN").as_ref() {
671 use cloud_api_client::PlanV2;
672
673 return match plan.as_str() {
674 "free" => Some(Plan::V2(PlanV2::ZedFree)),
675 "trial" => Some(Plan::V2(PlanV2::ZedProTrial)),
676 "pro" => Some(Plan::V2(PlanV2::ZedPro)),
677 _ => {
678 panic!("ZED_SIMULATE_PLAN must be one of 'free', 'trial', or 'pro'");
679 }
680 };
681 }
682
683 self.plan_info.as_ref().map(|info| info.plan())
684 }
685
686 pub fn subscription_period(&self) -> Option<(DateTime<Utc>, DateTime<Utc>)> {
687 self.plan_info
688 .as_ref()
689 .and_then(|plan| plan.subscription_period)
690 .map(|subscription_period| {
691 (
692 subscription_period.started_at.0,
693 subscription_period.ended_at.0,
694 )
695 })
696 }
697
698 pub fn trial_started_at(&self) -> Option<DateTime<Utc>> {
699 self.plan_info
700 .as_ref()
701 .and_then(|plan| plan.trial_started_at)
702 .map(|trial_started_at| trial_started_at.0)
703 }
704
705 /// Returns whether the user's account is too new to use the service.
706 pub fn account_too_young(&self) -> bool {
707 self.plan_info
708 .as_ref()
709 .map(|plan| plan.is_account_too_young)
710 .unwrap_or_default()
711 }
712
713 /// Returns whether the current user has overdue invoices and usage should be blocked.
714 pub fn has_overdue_invoices(&self) -> bool {
715 self.plan_info
716 .as_ref()
717 .map(|plan| plan.has_overdue_invoices)
718 .unwrap_or_default()
719 }
720
721 pub fn edit_prediction_usage(&self) -> Option<EditPredictionUsage> {
722 self.edit_prediction_usage
723 }
724
725 pub fn update_edit_prediction_usage(
726 &mut self,
727 usage: EditPredictionUsage,
728 cx: &mut Context<Self>,
729 ) {
730 self.edit_prediction_usage = Some(usage);
731 cx.notify();
732 }
733
734 pub fn clear_plan_and_usage(&mut self) {
735 self.plan_info = None;
736 self.edit_prediction_usage = None;
737 }
738
739 fn update_authenticated_user(
740 &mut self,
741 response: GetAuthenticatedUserResponse,
742 cx: &mut Context<Self>,
743 ) {
744 let staff = response.user.is_staff && !*feature_flags::ZED_DISABLE_STAFF;
745 cx.update_flags(staff, response.feature_flags);
746 if let Some(client) = self.client.upgrade() {
747 client
748 .telemetry
749 .set_authenticated_user_info(Some(response.user.metrics_id.clone()), staff);
750 }
751
752 self.edit_prediction_usage = Some(EditPredictionUsage(RequestUsage {
753 limit: response.plan.usage.edit_predictions.limit,
754 amount: response.plan.usage.edit_predictions.used as i32,
755 }));
756 self.plan_info = Some(response.plan);
757 cx.emit(Event::PrivateUserInfoUpdated);
758 }
759
760 fn handle_message_to_client(this: WeakEntity<Self>, message: &MessageToClient, cx: &App) {
761 cx.spawn(async move |cx| {
762 match message {
763 MessageToClient::UserUpdated => {
764 let cloud_client = cx
765 .update(|cx| {
766 this.read_with(cx, |this, _cx| {
767 this.client.upgrade().map(|client| client.cloud_client())
768 })
769 })?
770 .ok_or(anyhow::anyhow!("Failed to get Cloud client"))?;
771
772 let response = cloud_client.get_authenticated_user().await?;
773 cx.update(|cx| {
774 this.update(cx, |this, cx| {
775 this.update_authenticated_user(response, cx);
776 })
777 })?;
778 }
779 }
780
781 anyhow::Ok(())
782 })
783 .detach_and_log_err(cx);
784 }
785
786 pub fn watch_current_user(&self) -> watch::Receiver<Option<Arc<User>>> {
787 self.current_user.clone()
788 }
789
790 fn load_users(
791 &self,
792 request: impl RequestMessage<Response = UsersResponse>,
793 cx: &Context<Self>,
794 ) -> Task<Result<Vec<Arc<User>>>> {
795 let client = self.client.clone();
796 cx.spawn(async move |this, cx| {
797 if let Some(rpc) = client.upgrade() {
798 let response = rpc.request(request).await.context("error loading users")?;
799 let users = response.users;
800
801 this.update(cx, |this, _| this.insert(users))
802 } else {
803 Ok(Vec::new())
804 }
805 })
806 }
807
808 pub fn insert(&mut self, users: Vec<proto::User>) -> Vec<Arc<User>> {
809 let mut ret = Vec::with_capacity(users.len());
810 for user in users {
811 let user = User::new(user);
812 if let Some(old) = self.users.insert(user.id, user.clone())
813 && old.github_login != user.github_login
814 {
815 self.by_github_login.remove(&old.github_login);
816 }
817 self.by_github_login
818 .insert(user.github_login.clone(), user.id);
819 ret.push(user)
820 }
821 ret
822 }
823
824 pub fn set_participant_indices(
825 &mut self,
826 participant_indices: HashMap<u64, ParticipantIndex>,
827 cx: &mut Context<Self>,
828 ) {
829 if participant_indices != self.participant_indices {
830 self.participant_indices = participant_indices;
831 cx.emit(Event::ParticipantIndicesChanged);
832 }
833 }
834
835 pub fn participant_indices(&self) -> &HashMap<u64, ParticipantIndex> {
836 &self.participant_indices
837 }
838
839 pub fn participant_names(
840 &self,
841 user_ids: impl Iterator<Item = u64>,
842 cx: &App,
843 ) -> HashMap<u64, SharedString> {
844 let mut ret = HashMap::default();
845 let mut missing_user_ids = Vec::new();
846 for id in user_ids {
847 if let Some(github_login) = self.get_cached_user(id).map(|u| u.github_login.clone()) {
848 ret.insert(id, github_login);
849 } else {
850 missing_user_ids.push(id)
851 }
852 }
853 if !missing_user_ids.is_empty() {
854 let this = self.weak_self.clone();
855 cx.spawn(async move |cx| {
856 this.update(cx, |this, cx| this.get_users(missing_user_ids, cx))?
857 .await
858 })
859 .detach_and_log_err(cx);
860 }
861 ret
862 }
863}
864
865impl User {
866 fn new(message: proto::User) -> Arc<Self> {
867 Arc::new(User {
868 id: message.id,
869 github_login: message.github_login.into(),
870 avatar_uri: message.avatar_url.into(),
871 name: message.name,
872 })
873 }
874}
875
876impl Contact {
877 async fn from_proto(
878 contact: proto::Contact,
879 user_store: &Entity<UserStore>,
880 cx: &mut AsyncApp,
881 ) -> Result<Self> {
882 let user = user_store
883 .update(cx, |user_store, cx| {
884 user_store.get_user(contact.user_id, cx)
885 })
886 .await?;
887 Ok(Self {
888 user,
889 online: contact.online,
890 busy: contact.busy,
891 })
892 }
893}
894
895impl Collaborator {
896 pub fn from_proto(message: proto::Collaborator) -> Result<Self> {
897 Ok(Self {
898 peer_id: message.peer_id.context("invalid peer id")?,
899 replica_id: ReplicaId::new(message.replica_id as u16),
900 user_id: message.user_id as UserId,
901 is_host: message.is_host,
902 committer_name: message.committer_name,
903 committer_email: message.committer_email,
904 })
905 }
906}
907
908impl RequestUsage {
909 pub fn over_limit(&self) -> bool {
910 match self.limit {
911 UsageLimit::Limited(limit) => self.amount >= limit,
912 UsageLimit::Unlimited => false,
913 }
914 }
915
916 fn from_headers(
917 limit_name: &str,
918 amount_name: &str,
919 headers: &HeaderMap<HeaderValue>,
920 ) -> Result<Self> {
921 let limit = headers
922 .get(limit_name)
923 .with_context(|| format!("missing {limit_name:?} header"))?;
924 let limit = UsageLimit::from_str(limit.to_str()?)?;
925
926 let amount = headers
927 .get(amount_name)
928 .with_context(|| format!("missing {amount_name:?} header"))?;
929 let amount = amount.to_str()?.parse::<i32>()?;
930
931 Ok(Self { limit, amount })
932 }
933}
934
935impl EditPredictionUsage {
936 pub fn from_headers(headers: &HeaderMap<HeaderValue>) -> Result<Self> {
937 Ok(Self(RequestUsage::from_headers(
938 EDIT_PREDICTIONS_USAGE_LIMIT_HEADER_NAME,
939 EDIT_PREDICTIONS_USAGE_AMOUNT_HEADER_NAME,
940 headers,
941 )?))
942 }
943}