nexus_common/models/tag/
stream.rs

1use crate::db::kv::SortOrder;
2use crate::db::{queries, retrieve_from_graph, RedisOps};
3use crate::types::routes::HotTagsInputDTO;
4use crate::types::{DynError, StreamReach, Timeframe};
5use serde::{Deserialize, Serialize};
6use std::collections::HashMap;
7use std::ops::Deref;
8use utoipa::ToSchema;
9
10use super::global::{HotTagsTaggers, Taggers};
11use super::TaggedType;
12
13pub const HOT_TAGS_CACHE_PREFIX: &str = "Cache";
14pub const POST_HOT_TAGS: [&str; 3] = ["Tags", "Post", "Hot"];
15
16#[derive(Deserialize, Serialize, ToSchema, Debug, Clone)]
17pub struct HotTag {
18    pub label: String,
19    pub taggers_id: Taggers,
20    pub tagged_count: u64,
21    pub taggers_count: usize,
22}
23
24// Define a newtype wrapper
25#[derive(Serialize, Deserialize, Debug, ToSchema, Default, Clone)]
26pub struct HotTags(pub Vec<HotTag>);
27
28impl RedisOps for HotTags {}
29
30// Implement Deref so TagList can be used like Vec<String>
31impl Deref for HotTags {
32    type Target = Vec<HotTag>;
33
34    fn deref(&self) -> &Self::Target {
35        &self.0
36    }
37}
38
39// Create a HotTags instance directly from an iterator of HotTag items
40// Need it in collect()
41impl FromIterator<HotTag> for HotTags {
42    fn from_iter<I: IntoIterator<Item = HotTag>>(iter: I) -> Self {
43        HotTags(iter.into_iter().collect())
44    }
45}
46
47impl HotTags {
48    /// It dynamically determines whether to fetch **global hot tags** or **user-specific hot tags**
49    /// based on the provided `user_id` and `reach` parameters
50    ///
51    /// # Arguments
52    /// * `user_id` - An optional user ID
53    /// * `reach` - An optional `TagStreamReach` value specifying the scope of tag retrieval
54    /// * `hot_tags_input` - The input parameters received from the API endpoint
55    pub async fn get_hot_tags(
56        user_id: Option<String>,
57        reach: Option<StreamReach>,
58        hot_tags_input: &HotTagsInputDTO,
59    ) -> Result<Option<HotTags>, DynError> {
60        match user_id {
61            Some(user_id) => {
62                HotTags::get_hot_tags_by_reach(
63                    user_id,
64                    reach.unwrap_or(StreamReach::Following),
65                    hot_tags_input,
66                )
67                .await
68            }
69            None => HotTags::get_global_hot_tags(hot_tags_input).await,
70        }
71    }
72
73    /// Retrieves hot tags based on the user's reach criteria
74    /// Queries the graph database to fetch hot tags relevant to a given user,
75    /// filtered by their reach and additional criteria defined in `hot_tags_input`
76    ///
77    /// # Arguments
78    /// * `user_id` - The ID of the user whose reach is used for filtering hot tags
79    /// * `reach` - The `TagStreamReach` parameter that defines the scope of tag retrieval
80    /// * `hot_tags_input` - The input parameters received from the API endpoint
81    async fn get_hot_tags_by_reach(
82        user_id: String,
83        reach: StreamReach,
84        hot_tags_input: &HotTagsInputDTO,
85    ) -> Result<Option<HotTags>, DynError> {
86        let query = queries::get::get_hot_tags_by_reach(user_id.as_str(), reach, hot_tags_input);
87        retrieve_from_graph::<HotTags>(query, "hot_tags").await
88    }
89
90    /// Retrieves global hot tags, checking the cache first before querying the database.
91    /// This function first attempts to fetch global hot tags from the cache. If the cached
92    /// data is unavailable, it queries the graph database to retrieve the latest hot tags.
93    /// If new data is found, it updates the cache before returning the results.
94    ///
95    /// # Arguments
96    ///
97    /// * `hot_tags_input` - The input parameters received from the API endpoint
98    async fn get_global_hot_tags(
99        hot_tags_input: &HotTagsInputDTO,
100    ) -> Result<Option<HotTags>, DynError> {
101        let cached_hot_tags = HotTags::get_from_global_cache(hot_tags_input).await?;
102
103        if let Some(hot_tags) = &cached_hot_tags {
104            if hot_tags.0.is_empty() {
105                return Ok(None);
106            }
107            return Ok(cached_hot_tags);
108        }
109
110        let hot_tag_input = HotTagsInputDTO::new(
111            hot_tags_input.timeframe.clone(),
112            100,
113            0,
114            20,
115            hot_tags_input.tagged_type.clone(),
116        );
117        let query = queries::get::get_global_hot_tags(&hot_tag_input);
118        let result = retrieve_from_graph::<HotTags>(query, "hot_tags").await?;
119
120        let hot_tags = match result {
121            Some(hot_tags) => hot_tags,
122            None => return Ok(None),
123        };
124        if !hot_tags.is_empty() {
125            HotTags::set_to_global_cache(hot_tags.clone(), hot_tags_input).await?;
126        }
127
128        HotTags::get_from_global_cache(hot_tags_input).await
129    }
130
131    /// Retrieves hot tags from the global cache
132    ///
133    /// Fetches hot tags and their associated taggers from the cache, reconstructing
134    /// a list of hot tags from a stored JSON mapping and a hot tags SORTED SET. It applies filters
135    /// based on `hot_tags_input`, ensuring that only relevant tags and taggers are returned
136    ///
137    /// # Arguments
138    ///
139    /// * `hot_tags_input` - The input parameters received from the API endpoint
140    async fn get_from_global_cache(
141        hot_tags_input: &HotTagsInputDTO,
142    ) -> Result<Option<HotTags>, DynError> {
143        let timeframe = hot_tags_input.timeframe.to_string();
144        let hot_tag_key_parts = Self::build_hot_tags_key_parts(&timeframe);
145
146        let hot_tag_taggers = Taggers::get_from_index(&timeframe).await?;
147
148        let hot_tags_score = HotTags::try_from_index_sorted_set(
149            &hot_tag_key_parts,
150            None,
151            None,
152            Some(hot_tags_input.skip),
153            Some(hot_tags_input.limit),
154            SortOrder::Descending,
155            Some(HOT_TAGS_CACHE_PREFIX),
156        )
157        .await?;
158
159        let (hot_tags_score, hot_tag_taggers) = match (hot_tags_score, hot_tag_taggers) {
160            (Some(score_list), Some(taggers)) => {
161                // Index exist but applyting the DTO filters, there is not records
162                if score_list.is_empty() {
163                    return Ok(Some(HotTags(Vec::new())));
164                }
165                (score_list, taggers)
166            }
167            _ => return Ok(None),
168        };
169
170        let mut hot_tags = Vec::with_capacity(hot_tags_score.len());
171
172        for (label, score) in hot_tags_score {
173            if let Some(taggers) = hot_tag_taggers.get(&label) {
174                // Reduce taggers list
175                let taggers_id: Vec<String> =
176                    Taggers::get_taggers_by_pagination(taggers, 0, hot_tags_input.taggers_limit);
177                hot_tags.push(HotTag {
178                    label,
179                    taggers_id: Taggers(taggers_id),
180                    tagged_count: score as u64,
181                    taggers_count: taggers.len(),
182                });
183            }
184        }
185        Ok(Some(HotTags(hot_tags)))
186    }
187
188    /// Caches the global hot tags taggers and their scores
189    /// Gets hot tags and stores it in a global cache, both as a JSON
190    /// mapping of taggers and as a sorted set for score. It constructs cache keys dynamically
191    /// based on the provided timeframe
192    ///
193    /// # Arguments
194    ///
195    /// * `hot_tags_list` - A vector of `HotTag` elements
196    /// * `hot_tags_input` - The input parameters received from the API endpoint
197    async fn set_to_global_cache(
198        hot_tags_list: HotTags,
199        hot_tags_input: &HotTagsInputDTO,
200    ) -> Result<(), DynError> {
201        let timeframe = hot_tags_input.timeframe.to_string();
202        let hot_tag_key_parts = Self::build_hot_tags_key_parts(&timeframe);
203
204        let mut hot_tags_score = Vec::with_capacity(hot_tags_list.len());
205
206        let taggers: HashMap<String, Taggers> = hot_tags_list
207            .iter()
208            .map(|tag| {
209                hot_tags_score.push((tag.tagged_count as f64, tag.label.as_str()));
210                (tag.label.clone(), tag.taggers_id.clone())
211            })
212            .collect();
213
214        Taggers::put_to_index(HotTagsTaggers(taggers), &hot_tags_input.timeframe).await?;
215
216        // Store the score as sorted set in cache
217        HotTags::put_index_sorted_set(
218            &hot_tag_key_parts,
219            &hot_tags_score,
220            Some(HOT_TAGS_CACHE_PREFIX),
221            Some(hot_tags_input.timeframe.to_cache_period()),
222        )
223        .await?;
224        Ok(())
225    }
226
227    /// Builds key parts for hot tags based on the given timeframe
228    ///
229    /// # Arguments
230    /// * `timeframe` - A string slice representing the timeframe (e.g., "today", "this_month", "all_time")
231    fn build_hot_tags_key_parts(timeframe: &str) -> Vec<&str> {
232        [&POST_HOT_TAGS[..], &[timeframe]].concat()
233    }
234
235    /// Reindexes global hot tags
236    /// Retrieves and updates global hot tags for different timeframes. It fetches the top 100 hot tags
237    ///  with a taggers limit of 20 for both "all-time" and "this month" timeframes
238    pub async fn reindex() -> Result<(), DynError> {
239        let all_timeframe_input =
240            HotTagsInputDTO::new(Timeframe::AllTime, 100, 0, 20, Some(TaggedType::Post));
241        HotTags::get_global_hot_tags(&all_timeframe_input).await?;
242
243        let month_timeframe_input =
244            HotTagsInputDTO::new(Timeframe::ThisMonth, 100, 0, 20, Some(TaggedType::Post));
245        HotTags::get_global_hot_tags(&month_timeframe_input).await?;
246        Ok(())
247    }
248}
nexus_common/models/tag/stream.rs

nexus_common/models/tag/
stream.rs
