nexus_common/models/post/
counts.rs

1use crate::db::kv::JsonAction;
2use crate::db::{get_neo4j_graph, queries, RedisOps};
3use crate::models::tag::post::POST_TAGS_KEY_PARTS;
4use crate::types::DynError;
5use serde::{Deserialize, Serialize};
6use utoipa::ToSchema;
7
8use super::PostStream;
9
10/// Represents total counts of relationships of a user.
11#[derive(Serialize, Deserialize, ToSchema, Default, Debug)]
12pub struct PostCounts {
13    // how many times was pointed the post with a tag
14    pub tags: u32,
15    // Distinct tags where the post was referenced
16    pub unique_tags: u32,
17    pub replies: u32,
18    pub reposts: u32,
19}
20
21impl RedisOps for PostCounts {}
22
23impl PostCounts {
24    /// Retrieves counts by user ID, first trying to get from Redis, then from Neo4j if not found.
25    pub async fn get_by_id(author_id: &str, post_id: &str) -> Result<Option<PostCounts>, DynError> {
26        match Self::get_from_index(author_id, post_id).await? {
27            Some(counts) => Ok(Some(counts)),
28            None => {
29                let graph_response = Self::get_from_graph(author_id, post_id).await?;
30                if let Some((post_counts, is_reply)) = graph_response {
31                    post_counts
32                        .put_to_index(author_id, post_id, !is_reply)
33                        .await?;
34                    return Ok(Some(post_counts));
35                }
36                Ok(None)
37            }
38        }
39    }
40
41    pub async fn get_from_index(
42        author_id: &str,
43        post_id: &str,
44    ) -> Result<Option<PostCounts>, DynError> {
45        if let Some(post_counts) = Self::try_from_index_json(&[author_id, post_id], None).await? {
46            return Ok(Some(post_counts));
47        }
48        Ok(None)
49    }
50
51    /// Retrieves the counts from Neo4j.
52    pub async fn get_from_graph(
53        author_id: &str,
54        post_id: &str,
55    ) -> Result<Option<(PostCounts, bool)>, DynError> {
56        let mut result;
57        {
58            let graph = get_neo4j_graph()?;
59            let query = queries::get::post_counts(author_id, post_id);
60
61            let graph = graph.lock().await;
62            result = graph.execute(query).await?;
63        }
64
65        if let Some(row) = result.next().await? {
66            let post_exists: bool = row.get("exists").unwrap_or(false);
67            if post_exists {
68                let counts: PostCounts = row.get("counts")?;
69                let is_reply: bool = row.get("is_reply").unwrap_or(false);
70
71                return Ok(Some((counts, is_reply)));
72            }
73        }
74        Ok(None)
75    }
76
77    pub async fn put_to_index(
78        &self,
79        author_id: &str,
80        post_id: &str,
81        is_reply: bool,
82    ) -> Result<(), DynError> {
83        self.put_index_json(&[author_id, post_id], None, None)
84            .await?;
85
86        // avoid indexing replies into global feeds
87        if !is_reply {
88            PostStream::add_to_engagement_sorted_set(self, author_id, post_id).await?;
89        }
90        Ok(())
91    }
92
93    /// Updates a specified JSON field in the index
94    ///
95    /// # Arguments
96    ///
97    /// * `index_key` - A slice of string references representing the index key parts.
98    /// * `field` - The name of the JSON field to be updated.
99    /// * `action` - The action to perform on the JSON field (increment or decrement).
100    /// * `tag_label` - An optional tag label used to check membership in a sorted set. Important field to update the unique_tags field
101    pub async fn update_index_field(
102        index_key: &[&str],
103        field: &str,
104        action: JsonAction,
105        tag_label: Option<&str>,
106    ) -> Result<(), DynError> {
107        // This condition applies only when updating `unique_tags`
108        if let Some(label) = tag_label {
109            let index_parts = [&POST_TAGS_KEY_PARTS[..], index_key].concat();
110            let score = Self::check_sorted_set_member(None, &index_parts, &[label]).await?;
111            match (score, &action) {
112                // If tag value is less than 1, `unique_tags` can be incremented or decremented
113                (Some(tag_value), _) if tag_value < 1 => (),
114
115                // Incrementing `unique_tags` is also allowed when the tag value doesn't exist yet in the sorted set
116                (None, JsonAction::Increment(_)) => (),
117
118                // Do not update the index
119                _ => return Ok(()),
120            }
121        }
122
123        Self::modify_json_field(index_key, field, action).await?;
124        Ok(())
125    }
126
127    pub async fn reindex(author_id: &str, post_id: &str) -> Result<(), DynError> {
128        match Self::get_from_graph(author_id, post_id).await? {
129            Some((counts, is_reply)) => counts.put_to_index(author_id, post_id, is_reply).await?,
130            None => tracing::error!(
131                "{}:{} Could not found post counts in the graph",
132                author_id,
133                post_id
134            ),
135        }
136        Ok(())
137    }
138
139    pub async fn delete(
140        author_id: &str,
141        post_id: &str,
142        remove_from_feeds: bool,
143    ) -> Result<(), DynError> {
144        // Delete user_details on Redis
145        Self::remove_from_index_multiple_json(&[&[author_id, post_id]]).await?;
146        // Delete the posts that does not have any relationship as might be replies and reposts. Just root posts
147        if remove_from_feeds {
148            PostStream::delete_from_engagement_sorted_set(author_id, post_id).await?;
149        }
150        Ok(())
151    }
152}
nexus_common/models/post/counts.rs

nexus_common/models/post/
counts.rs
