1use super::{http::HttpClient, proto, Client, Status, TypedEnvelope};
2use anyhow::{anyhow, Result};
3use futures::{future, AsyncReadExt, Future};
4use gpui::{AsyncAppContext, Entity, ImageData, ModelContext, ModelHandle, Task};
5use postage::{prelude::Stream, sink::Sink, watch};
6use rpc::proto::{RequestMessage, UsersResponse};
7use std::{
8 collections::{HashMap, HashSet},
9 sync::{Arc, Weak},
10};
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
20#[derive(Debug)]
21pub struct Contact {
22 pub user: Arc<User>,
23 pub projects: Vec<ProjectMetadata>,
24}
25
26#[derive(Debug)]
27pub struct ProjectMetadata {
28 pub id: u64,
29 pub is_shared: bool,
30 pub worktree_root_names: Vec<String>,
31 pub guests: Vec<Arc<User>>,
32}
33
34pub struct UserStore {
35 users: HashMap<u64, Arc<User>>,
36 update_contacts_tx: watch::Sender<Option<proto::UpdateContacts>>,
37 current_user: watch::Receiver<Option<Arc<User>>>,
38 contacts: Vec<Arc<Contact>>,
39 incoming_contact_requests: Vec<Arc<User>>,
40 outgoing_contact_requests: Vec<Arc<User>>,
41 client: Weak<Client>,
42 http: Arc<dyn HttpClient>,
43 _maintain_contacts: Task<()>,
44 _maintain_current_user: Task<()>,
45}
46
47pub enum Event {}
48
49impl Entity for UserStore {
50 type Event = Event;
51}
52
53impl UserStore {
54 pub fn new(
55 client: Arc<Client>,
56 http: Arc<dyn HttpClient>,
57 cx: &mut ModelContext<Self>,
58 ) -> Self {
59 let (mut current_user_tx, current_user_rx) = watch::channel();
60 let (update_contacts_tx, mut update_contacts_rx) =
61 watch::channel::<Option<proto::UpdateContacts>>();
62 let rpc_subscription =
63 client.add_message_handler(cx.handle(), Self::handle_update_contacts);
64 Self {
65 users: Default::default(),
66 current_user: current_user_rx,
67 contacts: Default::default(),
68 incoming_contact_requests: Default::default(),
69 outgoing_contact_requests: Default::default(),
70 client: Arc::downgrade(&client),
71 update_contacts_tx,
72 http,
73 _maintain_contacts: cx.spawn_weak(|this, mut cx| async move {
74 let _subscription = rpc_subscription;
75 while let Some(message) = update_contacts_rx.recv().await {
76 if let Some((message, this)) = message.zip(this.upgrade(&cx)) {
77 this.update(&mut cx, |this, cx| this.update_contacts(message, cx))
78 .log_err()
79 .await;
80 }
81 }
82 }),
83 _maintain_current_user: cx.spawn_weak(|this, mut cx| async move {
84 let mut status = client.status();
85 while let Some(status) = status.recv().await {
86 match status {
87 Status::Connected { .. } => {
88 if let Some((this, user_id)) = this.upgrade(&cx).zip(client.user_id()) {
89 let user = this
90 .update(&mut cx, |this, cx| this.fetch_user(user_id, cx))
91 .log_err()
92 .await;
93 current_user_tx.send(user).await.ok();
94 }
95 }
96 Status::SignedOut => {
97 current_user_tx.send(None).await.ok();
98 }
99 _ => {}
100 }
101 }
102 }),
103 }
104 }
105
106 async fn handle_update_contacts(
107 this: ModelHandle<Self>,
108 msg: TypedEnvelope<proto::UpdateContacts>,
109 _: Arc<Client>,
110 mut cx: AsyncAppContext,
111 ) -> Result<()> {
112 this.update(&mut cx, |this, _| {
113 *this.update_contacts_tx.borrow_mut() = Some(msg.payload);
114 });
115 Ok(())
116 }
117
118 fn update_contacts(
119 &mut self,
120 message: proto::UpdateContacts,
121 cx: &mut ModelContext<Self>,
122 ) -> Task<Result<()>> {
123 let mut user_ids = HashSet::new();
124 for contact in &message.contacts {
125 user_ids.insert(contact.user_id);
126 user_ids.extend(contact.projects.iter().flat_map(|w| &w.guests).copied());
127 }
128 user_ids.extend(message.incoming_requests.iter().map(|req| req.user_id));
129 user_ids.extend(message.outgoing_requests.iter());
130
131 let load_users = self.get_users(user_ids.into_iter().collect(), cx);
132 cx.spawn(|this, mut cx| async move {
133 load_users.await?;
134
135 // Users are fetched in parallel above and cached in call to get_users
136 // No need to paralellize here
137 let mut updated_contacts = Vec::new();
138 for contact in message.contacts {
139 updated_contacts.push(Arc::new(
140 Contact::from_proto(contact, &this, &mut cx).await?,
141 ));
142 }
143
144 let mut incoming_requests = Vec::new();
145 for request in message.incoming_requests {
146 incoming_requests.push(
147 this.update(&mut cx, |this, cx| this.fetch_user(request.user_id, cx))
148 .await?,
149 );
150 }
151
152 let mut outgoing_requests = Vec::new();
153 for requested_user_id in message.outgoing_requests {
154 outgoing_requests.push(
155 this.update(&mut cx, |this, cx| this.fetch_user(requested_user_id, cx))
156 .await?,
157 );
158 }
159
160 let removed_contacts =
161 HashSet::<u64>::from_iter(message.remove_contacts.iter().copied());
162 let removed_incoming_requests =
163 HashSet::<u64>::from_iter(message.remove_incoming_requests.iter().copied());
164 let removed_outgoing_requests =
165 HashSet::<u64>::from_iter(message.remove_outgoing_requests.iter().copied());
166
167 this.update(&mut cx, |this, cx| {
168 this.contacts
169 .retain(|contact| !removed_contacts.contains(&contact.user.id));
170 this.contacts.extend(updated_contacts);
171 this.contacts
172 .sort_by(|a, b| a.user.github_login.cmp(&b.user.github_login));
173 cx.notify();
174
175 this.incoming_contact_requests
176 .retain(|user| !removed_incoming_requests.contains(&user.id));
177 this.incoming_contact_requests.extend(incoming_requests);
178 this.incoming_contact_requests
179 .sort_by(|a, b| a.github_login.cmp(&b.github_login));
180
181 this.outgoing_contact_requests
182 .retain(|user| !removed_outgoing_requests.contains(&user.id));
183 this.outgoing_contact_requests.extend(outgoing_requests);
184 this.outgoing_contact_requests
185 .sort_by(|a, b| a.github_login.cmp(&b.github_login));
186 });
187
188 Ok(())
189 })
190 }
191
192 pub fn contacts(&self) -> &[Arc<Contact>] {
193 &self.contacts
194 }
195
196 pub fn has_contact(&self, user: &Arc<User>) -> bool {
197 self.contacts
198 .binary_search_by_key(&&user.github_login, |contact| &contact.user.github_login)
199 .is_ok()
200 }
201
202 pub fn request_contact(&self, to_user_id: u64) -> impl Future<Output = Result<()>> {
203 let client = self.client.upgrade();
204 async move {
205 client
206 .ok_or_else(|| anyhow!("not logged in"))?
207 .request(proto::RequestContact { to_user_id })
208 .await?;
209 Ok(())
210 }
211 }
212
213 pub fn respond_to_contact_request(
214 &self,
215 from_user_id: u64,
216 accept: bool,
217 ) -> impl Future<Output = Result<()>> {
218 let client = self.client.upgrade();
219 async move {
220 client
221 .ok_or_else(|| anyhow!("not logged in"))?
222 .request(proto::RespondToContactRequest {
223 requesting_user_id: from_user_id,
224 response: if accept {
225 proto::ContactRequestResponse::Accept
226 } else {
227 proto::ContactRequestResponse::Reject
228 } as i32,
229 })
230 .await?;
231 Ok(())
232 }
233 }
234
235 pub fn get_users(
236 &mut self,
237 mut user_ids: Vec<u64>,
238 cx: &mut ModelContext<Self>,
239 ) -> Task<Result<Vec<Arc<User>>>> {
240 user_ids.retain(|id| !self.users.contains_key(id));
241 self.load_users(proto::GetUsers { user_ids }, cx)
242 }
243
244 pub fn fuzzy_search_users(
245 &mut self,
246 query: String,
247 cx: &mut ModelContext<Self>,
248 ) -> Task<Result<Vec<Arc<User>>>> {
249 self.load_users(proto::FuzzySearchUsers { query }, cx)
250 }
251
252 pub fn fetch_user(
253 &mut self,
254 user_id: u64,
255 cx: &mut ModelContext<Self>,
256 ) -> Task<Result<Arc<User>>> {
257 if let Some(user) = self.users.get(&user_id).cloned() {
258 return cx.foreground().spawn(async move { Ok(user) });
259 }
260
261 let load_users = self.get_users(vec![user_id], cx);
262 cx.spawn(|this, mut cx| async move {
263 load_users.await?;
264 this.update(&mut cx, |this, _| {
265 this.users
266 .get(&user_id)
267 .cloned()
268 .ok_or_else(|| anyhow!("server responded with no users"))
269 })
270 })
271 }
272
273 pub fn current_user(&self) -> Option<Arc<User>> {
274 self.current_user.borrow().clone()
275 }
276
277 pub fn watch_current_user(&self) -> watch::Receiver<Option<Arc<User>>> {
278 self.current_user.clone()
279 }
280
281 fn load_users(
282 &mut self,
283 request: impl RequestMessage<Response = UsersResponse>,
284 cx: &mut ModelContext<Self>,
285 ) -> Task<Result<Vec<Arc<User>>>> {
286 let client = self.client.clone();
287 let http = self.http.clone();
288 cx.spawn_weak(|this, mut cx| async move {
289 if let Some(rpc) = client.upgrade() {
290 let response = rpc.request(request).await?;
291 let users = future::join_all(
292 response
293 .users
294 .into_iter()
295 .map(|user| User::new(user, http.as_ref())),
296 )
297 .await;
298
299 if let Some(this) = this.upgrade(&cx) {
300 this.update(&mut cx, |this, _| {
301 for user in &users {
302 this.users.insert(user.id, user.clone());
303 }
304 });
305 }
306 Ok(users)
307 } else {
308 Ok(Vec::new())
309 }
310 })
311 }
312}
313
314impl User {
315 async fn new(message: proto::User, http: &dyn HttpClient) -> Arc<Self> {
316 Arc::new(User {
317 id: message.id,
318 github_login: message.github_login,
319 avatar: fetch_avatar(http, &message.avatar_url).warn_on_err().await,
320 })
321 }
322}
323
324impl Contact {
325 async fn from_proto(
326 contact: proto::Contact,
327 user_store: &ModelHandle<UserStore>,
328 cx: &mut AsyncAppContext,
329 ) -> Result<Self> {
330 let user = user_store
331 .update(cx, |user_store, cx| {
332 user_store.fetch_user(contact.user_id, cx)
333 })
334 .await?;
335 let mut projects = Vec::new();
336 for project in contact.projects {
337 let mut guests = Vec::new();
338 for participant_id in project.guests {
339 guests.push(
340 user_store
341 .update(cx, |user_store, cx| {
342 user_store.fetch_user(participant_id, cx)
343 })
344 .await?,
345 );
346 }
347 projects.push(ProjectMetadata {
348 id: project.id,
349 worktree_root_names: project.worktree_root_names.clone(),
350 is_shared: project.is_shared,
351 guests,
352 });
353 }
354 Ok(Self { user, projects })
355 }
356}
357
358async fn fetch_avatar(http: &dyn HttpClient, url: &str) -> Result<Arc<ImageData>> {
359 let mut response = http
360 .get(url, Default::default(), true)
361 .await
362 .map_err(|e| anyhow!("failed to send user avatar request: {}", e))?;
363
364 if !response.status().is_success() {
365 return Err(anyhow!("avatar request failed {:?}", response.status()));
366 }
367
368 let mut body = Vec::new();
369 response
370 .body_mut()
371 .read_to_end(&mut body)
372 .await
373 .map_err(|e| anyhow!("failed to read user avatar response body: {}", e))?;
374 let format = image::guess_format(&body)?;
375 let image = image::load_from_memory_with_format(&body, format)?.into_bgra8();
376 Ok(ImageData::new(image))
377}