nexus_common/models/user/
stream.rs

1use super::{Influencers, Muted, UserCounts, UserSearch, UserView};
2use crate::models::follow::{Followers, Following, Friends, UserFollows};
3use crate::types::{DynError, StreamReach, Timeframe};
4use std::collections::HashSet;
5
6use crate::db::kv::SortOrder;
7use crate::db::{get_neo4j_graph, queries, RedisOps};
8use crate::models::post::{PostStream, POST_REPLIES_PER_POST_KEY_PARTS};
9use serde::{Deserialize, Serialize};
10use tokio::task::spawn;
11use utoipa::ToSchema;
12
13pub const USER_MOSTFOLLOWED_KEY_PARTS: [&str; 2] = ["Users", "MostFollowed"];
14pub const USER_INFLUENCERS_KEY_PARTS: [&str; 2] = ["Users", "Influencers"];
15pub const CACHE_USER_RECOMMENDED_KEY_PARTS: [&str; 3] = ["Cache", "Users", "Recommended"];
16// TTL, 12HR
17pub const CACHE_USER_RECOMMENDED_TTL: i64 = 12 * 60 * 60;
18
19#[derive(Deserialize, ToSchema, Debug, Clone, PartialEq)]
20#[serde(rename_all = "snake_case")]
21pub enum UserStreamSource {
22    Followers,
23    Following,
24    Friends,
25    Muted,
26    MostFollowed,
27    Influencers,
28    Recommended,
29    PostReplies,
30}
31
32pub struct UserStreamInput {
33    pub user_id: Option<String>,
34    pub skip: Option<usize>,
35    pub limit: Option<usize>,
36    pub source: UserStreamSource,
37    pub reach: Option<StreamReach>,
38    pub timeframe: Option<Timeframe>,
39    pub preview: Option<bool>,
40    pub author_id: Option<String>,
41    pub post_id: Option<String>,
42}
43
44#[derive(Serialize, Deserialize, ToSchema, Default, Debug)]
45pub struct UserStream(pub Vec<UserView>);
46
47impl RedisOps for UserStream {}
48
49impl UserStream {
50    pub fn extend(&mut self, user_stream: UserStream) {
51        self.0.extend(user_stream.0);
52    }
53
54    pub async fn get_by_id(
55        input: UserStreamInput,
56        viewer_id: Option<String>,
57        depth: Option<u8>,
58    ) -> Result<Option<Self>, DynError> {
59        let user_ids = Self::get_user_list_from_source(input).await?;
60        match user_ids {
61            Some(users) => Self::from_listed_user_ids(&users, viewer_id.as_deref(), depth).await,
62            None => Ok(None),
63        }
64    }
65
66    pub async fn get_from_username_search(
67        username: &str,
68        viewer_id: Option<&str>,
69        skip: Option<usize>,
70        limit: Option<usize>,
71    ) -> Result<Option<Self>, DynError> {
72        let user_ids = UserSearch::get_by_name(username, skip, limit)
73            .await?
74            .map(|result| result.0);
75
76        match user_ids {
77            Some(users) => Self::from_listed_user_ids(&users, viewer_id, None).await,
78            None => Ok(None),
79        }
80    }
81
82    pub async fn from_listed_user_ids(
83        user_ids: &[String],
84        viewer_id: Option<&str>,
85        depth: Option<u8>,
86    ) -> Result<Option<Self>, DynError> {
87        // TODO: potentially we could use a new redis_com.mget() with a single call to retrieve all
88        // user details at once and build the user profiles on the fly.
89        // But still, using tokio to create them concurrently has VERY high performance.
90        let viewer_id = viewer_id.map(|id| id.to_string());
91        let mut handles = Vec::with_capacity(user_ids.len());
92
93        for user_id in user_ids {
94            let user_id = user_id.clone();
95            let viewer_id = viewer_id.clone();
96            let handle =
97                spawn(
98                    async move { UserView::get_by_id(&user_id, viewer_id.as_deref(), depth).await },
99                );
100            handles.push(handle);
101        }
102
103        let mut user_views = Vec::with_capacity(user_ids.len());
104
105        for handle in handles {
106            if let Some(user_view) = handle.await?? {
107                user_views.push(user_view);
108            }
109        }
110
111        match user_views.is_empty() {
112            true => Ok(None),
113            false => Ok(Some(Self(user_views))),
114        }
115    }
116
117    /// Adds the post to a Redis sorted set using the follower counts as score.
118    pub async fn add_to_most_followed_sorted_set(
119        user_id: &str,
120        counts: &UserCounts,
121    ) -> Result<(), DynError> {
122        Self::put_index_sorted_set(
123            &USER_MOSTFOLLOWED_KEY_PARTS,
124            &[(counts.followers as f64, user_id)],
125            None,
126            None,
127        )
128        .await
129    }
130
131    /// Adds the post to a Redis sorted set using the follower counts as score.
132    pub async fn add_to_influencers_sorted_set(
133        user_id: &str,
134        counts: &UserCounts,
135    ) -> Result<(), DynError> {
136        let score = (counts.tagged + counts.posts) as f64 * (counts.followers as f64).sqrt();
137        Self::put_index_sorted_set(&USER_INFLUENCERS_KEY_PARTS, &[(score, user_id)], None, None)
138            .await
139    }
140    /// Retrieves recommended user IDs based on the specified criteria.
141    pub async fn get_recommended_ids(
142        user_id: &str,
143        limit: Option<usize>,
144    ) -> Result<Option<Vec<String>>, DynError> {
145        let count = limit.unwrap_or(5) as isize;
146
147        // Attempt to get cached data from Redis
148        if let Some(cached_data) = Self::try_get_cached_recommended(user_id, count).await? {
149            return Ok(Some(cached_data));
150        }
151
152        // Cache miss; proceed to query Neo4j
153        let mut result;
154        {
155            let graph = get_neo4j_graph()?;
156            // Query Neo4j for 30 user IDs
157            let query = queries::get::recommend_users(user_id, 30);
158
159            let graph = graph.lock().await;
160            result = graph.execute(query).await?;
161        }
162
163        let mut user_ids = Vec::new();
164
165        while let Some(row) = result.next().await? {
166            if let Some(user_id) = row.get::<Option<String>>("recommended_user_id")? {
167                user_ids.push(user_id);
168            }
169        }
170
171        if user_ids.is_empty() {
172            Ok(None)
173        } else {
174            Self::cache_recommended_users(user_id, &user_ids).await?;
175            if let Some(limit) = limit {
176                user_ids.truncate(limit);
177            };
178            Ok(Some(user_ids))
179        }
180    }
181
182    async fn try_get_cached_recommended(
183        user_id: &str,
184        count: isize,
185    ) -> Result<Option<Vec<String>>, DynError> {
186        let key_parts = &["Cache", "Recommended", user_id];
187        Self::try_get_random_from_index_set(
188            key_parts,
189            count,
190            Some(CACHE_USER_RECOMMENDED_KEY_PARTS.join(":")),
191        )
192        .await
193    }
194
195    /// Helper method to cache recommended users in Redis with a TTL.
196    async fn cache_recommended_users(user_id: &str, user_ids: &[String]) -> Result<(), DynError> {
197        let values: Vec<&str> = user_ids.iter().map(|s| s.as_str()).collect();
198        // Cache the result in Redis with a TTL of 12 hours
199        Self::put_index_set(
200            &[user_id],
201            &values,
202            Some(CACHE_USER_RECOMMENDED_TTL),
203            Some(CACHE_USER_RECOMMENDED_KEY_PARTS.join(":")),
204        )
205        .await
206    }
207
208    async fn get_post_replies_ids(
209        post_id: Option<String>,
210        author_id: Option<String>,
211    ) -> Result<Option<Vec<String>>, DynError> {
212        let post_id = post_id
213            .ok_or("Post ID should be provided for user streams with source 'post_replies'")?;
214        let author_id = author_id
215            .ok_or("Author ID should be provided for user streams with source 'post_replies'")?;
216        let key_parts = [
217            &POST_REPLIES_PER_POST_KEY_PARTS[..],
218            &[author_id.as_str(), post_id.as_str()],
219        ]
220        .concat();
221        let replies = PostStream::try_from_index_sorted_set(
222            &key_parts,
223            None,
224            None,
225            None,
226            None,
227            SortOrder::Descending,
228            None,
229        )
230        .await?;
231
232        // If there are replies, extract unique user IDs using a HashSet.
233        let unique_user_ids: HashSet<String> = if let Some(replies) = replies {
234            replies
235                .into_iter()
236                .filter_map(|reply| reply.0.split(':').next().map(|s| s.to_string()))
237                .collect()
238        } else {
239            // If no replies are found, return None.
240            return Ok(None);
241        };
242
243        // Convert the HashSet to a Vec. (Note: the ordering will be arbitrary.)
244        Ok(Some(unique_user_ids.into_iter().collect()))
245    }
246
247    // Get list of users based on the specified reach type
248    pub async fn get_user_list_from_source(
249        input: UserStreamInput,
250    ) -> Result<Option<Vec<String>>, DynError> {
251        let UserStreamInput {
252            user_id,
253            skip,
254            limit,
255            source,
256            reach,
257            timeframe,
258            preview,
259            author_id,
260            post_id,
261        } = input;
262        let user_ids =
263            match source {
264                UserStreamSource::Followers => Followers::get_by_id(
265                    user_id
266                        .ok_or(
267                            "User ID should be provided for user streams with source 'followers'"
268                                .to_string(),
269                        )?
270                        .as_str(),
271                    skip,
272                    limit,
273                )
274                .await?
275                .map(|u| u.0),
276                UserStreamSource::Following => Following::get_by_id(
277                    user_id
278                        .ok_or(
279                            "User ID should be provided for user streams with source 'following'"
280                                .to_string(),
281                        )?
282                        .as_str(),
283                    skip,
284                    limit,
285                )
286                .await?
287                .map(|u| u.0),
288                UserStreamSource::Friends => Friends::get_by_id(
289                    user_id
290                        .ok_or(
291                            "User ID should be provided for user streams with source 'friends'"
292                                .to_string(),
293                        )?
294                        .as_str(),
295                    skip,
296                    limit,
297                )
298                .await?
299                .map(|u| u.0),
300                UserStreamSource::Muted => Muted::get_by_id(
301                    user_id
302                        .ok_or(
303                            "User ID should be provided for user streams with source 'muted'"
304                                .to_string(),
305                        )?
306                        .as_str(),
307                    skip,
308                    limit,
309                )
310                .await?
311                .map(|u| u.0),
312                UserStreamSource::MostFollowed => Self::try_from_index_sorted_set(
313                    &USER_MOSTFOLLOWED_KEY_PARTS,
314                    None,
315                    None,
316                    skip,
317                    limit,
318                    SortOrder::Descending,
319                    None,
320                )
321                .await?
322                .map(|set| set.into_iter().map(|(user_id, _score)| user_id).collect()),
323                UserStreamSource::Influencers => Influencers::get_influencers(
324                    user_id.as_deref(),
325                    Some(reach.unwrap_or(StreamReach::Wot(3))),
326                    skip.unwrap_or(0),
327                    limit.unwrap_or(10).min(100),
328                    timeframe.unwrap_or(Timeframe::AllTime),
329                    preview.unwrap_or(false),
330                )
331                .await?
332                .map(|result| {
333                    result
334                        .iter()
335                        .map(|(influencer_id, _)| influencer_id.clone())
336                        .collect()
337                }),
338                UserStreamSource::Recommended => UserStream::get_recommended_ids(
339                    user_id
340                        .ok_or(
341                            "User ID should be provided for user streams with source 'recommended'"
342                                .to_string(),
343                        )?
344                        .as_str(),
345                    limit,
346                )
347                .await?,
348                UserStreamSource::PostReplies => {
349                    UserStream::get_post_replies_ids(post_id, author_id).await?
350                }
351            };
352        Ok(user_ids)
353    }
354}
nexus_common/models/user/stream.rs

nexus_common/models/user/
stream.rs
