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,
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: Arc<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: client.clone(),
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 !user_ids.is_empty() {
160 let response = rpc.request(proto::GetUsers { user_ids }).await?;
161 let new_users = future::join_all(
162 response
163 .users
164 .into_iter()
165 .map(|user| User::new(user, http.as_ref())),
166 )
167 .await;
168
169 if let Some(this) = this.upgrade(&cx) {
170 this.update(&mut cx, |this, _| {
171 for user in new_users {
172 this.users.insert(user.id, Arc::new(user));
173 }
174 });
175 }
176 }
177
178 Ok(())
179 })
180 }
181
182 pub fn fetch_user(
183 &mut self,
184 user_id: u64,
185 cx: &mut ModelContext<Self>,
186 ) -> Task<Result<Arc<User>>> {
187 if let Some(user) = self.users.get(&user_id).cloned() {
188 return cx.spawn_weak(|_, _| async move { Ok(user) });
189 }
190
191 let load_users = self.load_users(vec![user_id], cx);
192 cx.spawn(|this, mut cx| async move {
193 load_users.await?;
194 this.update(&mut cx, |this, _| {
195 this.users
196 .get(&user_id)
197 .cloned()
198 .ok_or_else(|| anyhow!("server responded with no users"))
199 })
200 })
201 }
202
203 pub fn current_user(&self) -> Option<Arc<User>> {
204 self.current_user.borrow().clone()
205 }
206
207 pub fn watch_current_user(&self) -> watch::Receiver<Option<Arc<User>>> {
208 self.current_user.clone()
209 }
210}
211
212impl User {
213 async fn new(message: proto::User, http: &dyn HttpClient) -> Self {
214 User {
215 id: message.id,
216 github_login: message.github_login,
217 avatar: fetch_avatar(http, &message.avatar_url).warn_on_err().await,
218 }
219 }
220}
221
222impl Contact {
223 async fn from_proto(
224 contact: proto::Contact,
225 user_store: &ModelHandle<UserStore>,
226 cx: &mut AsyncAppContext,
227 ) -> Result<Self> {
228 let user = user_store
229 .update(cx, |user_store, cx| {
230 user_store.fetch_user(contact.user_id, cx)
231 })
232 .await?;
233 let mut projects = Vec::new();
234 for project in contact.projects {
235 let mut guests = Vec::new();
236 for participant_id in project.guests {
237 guests.push(
238 user_store
239 .update(cx, |user_store, cx| {
240 user_store.fetch_user(participant_id, cx)
241 })
242 .await?,
243 );
244 }
245 projects.push(ProjectMetadata {
246 id: project.id,
247 worktree_root_names: project.worktree_root_names.clone(),
248 is_shared: project.is_shared,
249 guests,
250 });
251 }
252 Ok(Self { user, projects })
253 }
254}
255
256async fn fetch_avatar(http: &dyn HttpClient, url: &str) -> Result<Arc<ImageData>> {
257 let url = Url::parse(url).with_context(|| format!("failed to parse avatar url {:?}", url))?;
258 let mut request = Request::new(Method::Get, url);
259 request.middleware(surf::middleware::Redirect::default());
260
261 let mut response = http
262 .send(request)
263 .await
264 .map_err(|e| anyhow!("failed to send user avatar request: {}", e))?;
265 if !response.status().is_success() {
266 return Err(anyhow!("avatar request failed {:?}", response.status()));
267 }
268 let bytes = response
269 .body_bytes()
270 .await
271 .map_err(|e| anyhow!("failed to read user avatar response body: {}", e))?;
272 let format = image::guess_format(&bytes)?;
273 let image = image::load_from_memory_with_format(&bytes, format)?.into_bgra8();
274 Ok(ImageData::new(image))
275}