nexus_common/models/user/
influencers.rs

1use crate::db::kv::SortOrder;
2use crate::types::DynError;
3use crate::types::StreamReach;
4use crate::types::Timeframe;
5use chrono::Utc;
6use serde::{Deserialize, Serialize};
7use std::ops::Deref;
8use tracing::debug;
9use utoipa::ToSchema;
10
11use crate::db::{queries, retrieve_from_graph, RedisOps};
12
13const GLOBAL_INFLUENCERS_PREFIX: &str = "Cache:Influencers";
14
15#[derive(Serialize, Deserialize, Debug, ToSchema, Default, Clone)]
16pub struct Influencers(pub Vec<(String, f64)>); // (user_id, score)
17
18impl RedisOps for Influencers {}
19
20// Create a Influencers instance directly from an iterator of Influencer items
21// Need it in collect()
22impl FromIterator<(String, f64)> for Influencers {
23    fn from_iter<I: IntoIterator<Item = (String, f64)>>(iter: I) -> Self {
24        Influencers(iter.into_iter().collect())
25    }
26}
27
28// Implement Deref so Influencers can be used like Vec<String>
29impl Deref for Influencers {
30    type Target = Vec<(String, f64)>;
31
32    fn deref(&self) -> &Self::Target {
33        &self.0
34    }
35}
36
37impl Influencers {
38    /// Retrieves a list of influencers based on the provided context.
39    ///
40    /// If a `user_id` is provided, the function returns influencers relevant to the user
41    /// using the specified `reach` level (e.g., friends, followers). Otherwise, it returns
42    /// global influencers. When `preview` mode is enabled, it overrides the `skip` and `limit`
43    /// values with pseudo-random values to support randomized previews
44    ///
45    /// # Arguments
46    ///
47    /// * `user_id` - Optional user ID to fetch influencers relative to the user
48    /// * `reach` - Optional reach filter (e.g., Friends, Followers) for user-scoped queries
49    /// * `skip` - Number of results to skip (ignored in preview mode)
50    /// * `limit` - Maximum number of results to return (ignored in preview mode)
51    /// * `timeframe` - Time range to filter influencer activity
52    /// * `preview` - If true, uses pseudo-random pagination to return a small randomized subset
53    ///
54    pub async fn get_influencers(
55        user_id: Option<&str>,
56        reach: Option<StreamReach>,
57        skip: usize,
58        limit: usize,
59        timeframe: Timeframe,
60        preview: bool,
61    ) -> Result<Option<Influencers>, DynError> {
62        let (skip, limit) = if preview {
63            // Generate a pseudo-random number between 0 and 97
64            // We cache 100 influencers, and pick 3 starting from this number
65            // Using modulo 98 ensures we always have room for 3 without going out of bounds
66            let skip = Utc::now().timestamp_subsec_micros() % 98;
67            debug!("Influencer preview active: skip number {}", skip);
68            (skip as usize, 3)
69        } else {
70            (skip, limit)
71        };
72        if let Some(user) = user_id.filter(|_| timeframe != Timeframe::AllTime) {
73            return Influencers::get_influencers_by_reach(
74                user,
75                reach.unwrap_or(StreamReach::Friends),
76                skip,
77                limit,
78                &timeframe,
79            )
80            .await;
81        }
82        Influencers::get_global_influencers(skip, limit, &timeframe).await
83    }
84
85    /// It first attempts to fetch a subset of global influencers from cache
86    /// based on the provided `skip` and `limit`. If the cache is empty or unavailable,
87    /// it queries the graph database for up to 100 global influencers, stores the result
88    /// in cache, and then retrieves the requested subset again from cache.
89    ///
90    /// # Arguments
91    ///
92    /// * `skip` - Number of entries to skip (for pagination)
93    /// * `limit` - Maximum number of influencers to return
94    /// * `timeframe` - The time range to filter influencer activity
95    ///
96    async fn get_global_influencers(
97        skip: usize,
98        limit: usize,
99        timeframe: &Timeframe,
100    ) -> Result<Option<Influencers>, DynError> {
101        let cached_influencers = Influencers::get_from_global_cache(skip, limit, timeframe).await?;
102        if cached_influencers.is_some() {
103            return Ok(cached_influencers);
104        }
105
106        let query = queries::get::get_global_influencers(0, 100, timeframe);
107        let result = retrieve_from_graph::<Influencers>(query, "influencers").await?;
108
109        let influencers = match result {
110            Some(influencers) => influencers,
111            None => return Ok(None),
112        };
113
114        if !influencers.is_empty() {
115            Influencers::put_to_global_cache(influencers.clone(), timeframe).await?;
116        }
117
118        Influencers::get_from_global_cache(skip, limit, timeframe).await
119    }
120
121    /// Retrieves a paginated list of global influencers from the cache for the given timeframe
122    ///
123    /// # Arguments
124    ///
125    /// * `skip` - Number of entries to skip in the sorted set
126    /// * `limit` - Maximum number of influencers to return
127    /// * `timeframe` - The time window to filter influencer rankings, used to generate the cache key
128    async fn get_from_global_cache(
129        skip: usize,
130        limit: usize,
131        timeframe: &Timeframe,
132    ) -> Result<Option<Influencers>, DynError> {
133        let ranking = match timeframe {
134            // When timeframe is AllTime, we get the influencer list directly from Sorted::Users::Influencers,
135            // which is dynamically updated with each user action and therefore needs no TTL.
136            // Had we used the cache with TTL, it would have meant a random user gets hit with
137            // a full graph lookup, if they query this right after the TTL expires.
138            Timeframe::AllTime => {
139                Influencers::try_from_index_sorted_set(
140                    super::USER_INFLUENCERS_KEY_PARTS.as_slice(),
141                    None,
142                    None,
143                    Some(skip),
144                    Some(limit),
145                    SortOrder::Descending,
146                    None,
147                )
148                .await?
149            }
150
151            // For all other timeframes, we fallback to the cache with TTL (Cache::Influencers::Timeframe)
152            _ => {
153                let key_parts = Influencers::get_cache_key_parts(timeframe);
154                let key_parts_vector: Vec<&str> = key_parts.iter().map(|s| s.as_str()).collect();
155
156                Influencers::try_from_index_sorted_set(
157                    key_parts_vector.as_slice(),
158                    None,
159                    None,
160                    Some(skip),
161                    Some(limit),
162                    SortOrder::Descending,
163                    Some(GLOBAL_INFLUENCERS_PREFIX),
164                )
165                .await?
166            }
167        };
168
169        Ok(ranking.map(Influencers))
170    }
171
172    /// Stores a list of global influencers in the cache as a sorted set for the given timeframe
173    ///
174    /// # Arguments
175    /// * `result` - The list of influencers with their scores to cache
176    /// * `timeframe` - The timeframe used to generate the cache key and expiry
177    async fn put_to_global_cache(
178        result: Influencers,
179        timeframe: &Timeframe,
180    ) -> Result<(), DynError> {
181        let key_parts = Influencers::get_cache_key_parts(timeframe);
182        let key_parts_vector: Vec<&str> =
183            key_parts.iter().map(|s| s.as_str()).collect::<Vec<&str>>();
184
185        // store the ranking as sorted set in cache
186        Influencers::put_index_sorted_set(
187            key_parts_vector.as_slice(),
188            result
189                .iter()
190                .map(|influencer| (influencer.1, influencer.0.as_str()))
191                .collect::<Vec<(f64, &str)>>()
192                .as_slice(),
193            Some(GLOBAL_INFLUENCERS_PREFIX),
194            Some(timeframe.to_cache_period()),
195        )
196        .await?;
197        Ok(())
198    }
199
200    /// Retrieves influencers for a user based on the given `reach` and `timeframe` from the graph
201    ///
202    /// # Arguments
203    /// * `user_id` - The ID of the user to scope the influencer query
204    /// * `reach` - The reach filter (e.g., Friends, Followers)
205    /// * `skip` - Number of results to skip (for pagination)
206    /// * `limit` - Maximum number of influencers to return
207    /// * `timeframe` - Time window to filter influencer activity
208    async fn get_influencers_by_reach(
209        user_id: &str,
210        reach: StreamReach,
211        skip: usize,
212        limit: usize,
213        timeframe: &Timeframe,
214    ) -> Result<Option<Influencers>, DynError> {
215        let query = queries::get::get_influencers_by_reach(user_id, reach, skip, limit, timeframe);
216        retrieve_from_graph::<Influencers>(query, "influencers").await
217    }
218
219    fn get_cache_key_parts(timeframe: &Timeframe) -> Vec<String> {
220        vec![timeframe.to_string()]
221    }
222
223    /// Rebuilds the global influencer cache for `AllTime` and `ThisMonth` timeframes
224    ///
225    pub async fn reindex() -> Result<(), DynError> {
226        Influencers::get_global_influencers(0, 100, &Timeframe::AllTime).await?;
227        Influencers::get_global_influencers(0, 100, &Timeframe::ThisMonth).await?;
228        Ok(())
229    }
230}
nexus_common/models/user/influencers.rs

nexus_common/models/user/
influencers.rs
