1use super::{
2 http::{HttpClient, Method, Request, Url},
3 proto, Client, Status, TypedEnvelope,
4};
5use anyhow::{anyhow, Context, Result};
6use futures::future;
7use gpui::{AsyncAppContext, Entity, ImageData, ModelContext, ModelHandle, Task};
8use postage::{prelude::Stream, sink::Sink, watch};
9use std::{
10 collections::{HashMap, HashSet},
11 sync::{Arc, Weak},
12};
13use util::TryFutureExt as _;
14
15#[derive(Debug)]
16pub struct User {
17 pub id: u64,
18 pub github_login: String,
19 pub avatar: Option<Arc<ImageData>>,
20}
21
22#[derive(Debug)]
23pub struct Contact {
24 pub user: Arc<User>,
25 pub projects: Vec<ProjectMetadata>,
26}
27
28#[derive(Debug)]
29pub struct ProjectMetadata {
30 pub id: u64,
31 pub is_shared: bool,
32 pub worktree_root_names: Vec<String>,
33 pub guests: Vec<Arc<User>>,
34}
35
36pub struct UserStore {
37 users: HashMap<u64, Arc<User>>,
38 update_contacts_tx: watch::Sender<Option<proto::UpdateContacts>>,
39 current_user: watch::Receiver<Option<Arc<User>>>,
40 contacts: Arc<[Contact]>,
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: Arc::from([]),
68 client: Arc::downgrade(&client),
69 update_contacts_tx,
70 http,
71 _maintain_contacts: cx.spawn_weak(|this, mut cx| async move {
72 let _subscription = rpc_subscription;
73 while let Some(message) = update_contacts_rx.recv().await {
74 if let Some((message, this)) = message.zip(this.upgrade(&cx)) {
75 this.update(&mut cx, |this, cx| this.update_contacts(message, cx))
76 .log_err()
77 .await;
78 }
79 }
80 }),
81 _maintain_current_user: cx.spawn_weak(|this, mut cx| async move {
82 let mut status = client.status();
83 while let Some(status) = status.recv().await {
84 match status {
85 Status::Connected { .. } => {
86 if let Some((this, user_id)) = this.upgrade(&cx).zip(client.user_id()) {
87 let user = this
88 .update(&mut cx, |this, cx| this.fetch_user(user_id, cx))
89 .log_err()
90 .await;
91 current_user_tx.send(user).await.ok();
92 }
93 }
94 Status::SignedOut => {
95 current_user_tx.send(None).await.ok();
96 }
97 _ => {}
98 }
99 }
100 }),
101 }
102 }
103
104 async fn handle_update_contacts(
105 this: ModelHandle<Self>,
106 msg: TypedEnvelope<proto::UpdateContacts>,
107 _: Arc<Client>,
108 mut cx: AsyncAppContext,
109 ) -> Result<()> {
110 this.update(&mut cx, |this, _| {
111 *this.update_contacts_tx.borrow_mut() = Some(msg.payload);
112 });
113 Ok(())
114 }
115
116 fn update_contacts(
117 &mut self,
118 message: proto::UpdateContacts,
119 cx: &mut ModelContext<Self>,
120 ) -> Task<Result<()>> {
121 let mut user_ids = HashSet::new();
122 for contact in &message.contacts {
123 user_ids.insert(contact.user_id);
124 user_ids.extend(contact.projects.iter().flat_map(|w| &w.guests).copied());
125 }
126
127 let load_users = self.load_users(user_ids.into_iter().collect(), cx);
128 cx.spawn(|this, mut cx| async move {
129 load_users.await?;
130
131 let mut contacts = Vec::new();
132 for contact in message.contacts {
133 contacts.push(Contact::from_proto(contact, &this, &mut cx).await?);
134 }
135
136 this.update(&mut cx, |this, cx| {
137 contacts.sort_by(|a, b| a.user.github_login.cmp(&b.user.github_login));
138 this.contacts = contacts.into();
139 cx.notify();
140 });
141
142 Ok(())
143 })
144 }
145
146 pub fn contacts(&self) -> &Arc<[Contact]> {
147 &self.contacts
148 }
149
150 pub fn load_users(
151 &mut self,
152 mut user_ids: Vec<u64>,
153 cx: &mut ModelContext<Self>,
154 ) -> Task<Result<()>> {
155 let rpc = self.client.clone();
156 let http = self.http.clone();
157 user_ids.retain(|id| !self.users.contains_key(id));
158 cx.spawn_weak(|this, mut cx| async move {
159 if let Some(rpc) = rpc.upgrade() {
160 if !user_ids.is_empty() {
161 let response = rpc.request(proto::GetUsers { user_ids }).await?;
162 let new_users = future::join_all(
163 response
164 .users
165 .into_iter()
166 .map(|user| User::new(user, http.as_ref())),
167 )
168 .await;
169
170 if let Some(this) = this.upgrade(&cx) {
171 this.update(&mut cx, |this, _| {
172 for user in new_users {
173 this.users.insert(user.id, Arc::new(user));
174 }
175 });
176 }
177 }
178 }
179 Ok(())
180 })
181 }
182
183 pub fn fetch_user(
184 &mut self,
185 user_id: u64,
186 cx: &mut ModelContext<Self>,
187 ) -> Task<Result<Arc<User>>> {
188 if let Some(user) = self.users.get(&user_id).cloned() {
189 return cx.spawn_weak(|_, _| async move { Ok(user) });
190 }
191
192 let load_users = self.load_users(vec![user_id], cx);
193 cx.spawn(|this, mut cx| async move {
194 load_users.await?;
195 this.update(&mut cx, |this, _| {
196 this.users
197 .get(&user_id)
198 .cloned()
199 .ok_or_else(|| anyhow!("server responded with no users"))
200 })
201 })
202 }
203
204 pub fn current_user(&self) -> Option<Arc<User>> {
205 self.current_user.borrow().clone()
206 }
207
208 pub fn watch_current_user(&self) -> watch::Receiver<Option<Arc<User>>> {
209 self.current_user.clone()
210 }
211}
212
213impl User {
214 async fn new(message: proto::User, http: &dyn HttpClient) -> Self {
215 User {
216 id: message.id,
217 github_login: message.github_login,
218 avatar: fetch_avatar(http, &message.avatar_url).warn_on_err().await,
219 }
220 }
221}
222
223impl Contact {
224 async fn from_proto(
225 contact: proto::Contact,
226 user_store: &ModelHandle<UserStore>,
227 cx: &mut AsyncAppContext,
228 ) -> Result<Self> {
229 let user = user_store
230 .update(cx, |user_store, cx| {
231 user_store.fetch_user(contact.user_id, cx)
232 })
233 .await?;
234 let mut projects = Vec::new();
235 for project in contact.projects {
236 let mut guests = Vec::new();
237 for participant_id in project.guests {
238 guests.push(
239 user_store
240 .update(cx, |user_store, cx| {
241 user_store.fetch_user(participant_id, cx)
242 })
243 .await?,
244 );
245 }
246 projects.push(ProjectMetadata {
247 id: project.id,
248 worktree_root_names: project.worktree_root_names.clone(),
249 is_shared: project.is_shared,
250 guests,
251 });
252 }
253 Ok(Self { user, projects })
254 }
255}
256
257async fn fetch_avatar(http: &dyn HttpClient, url: &str) -> Result<Arc<ImageData>> {
258 let url = Url::parse(url).with_context(|| format!("failed to parse avatar url {:?}", url))?;
259 let mut request = Request::new(Method::Get, url);
260 request.middleware(surf::middleware::Redirect::default());
261
262 let mut response = http
263 .send(request)
264 .await
265 .map_err(|e| anyhow!("failed to send user avatar request: {}", e))?;
266 if !response.status().is_success() {
267 return Err(anyhow!("avatar request failed {:?}", response.status()));
268 }
269 let bytes = response
270 .body_bytes()
271 .await
272 .map_err(|e| anyhow!("failed to read user avatar response body: {}", e))?;
273 let format = image::guess_format(&bytes)?;
274 let image = image::load_from_memory_with_format(&bytes, format)?.into_bgra8();
275 Ok(ImageData::new(image))
276}