1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
use crate::db::kv::JsonAction;
use crate::db::{get_neo4j_graph, queries, RedisOps};
use crate::models::tag::user::USER_TAGS_KEY_PARTS;
use crate::types::DynError;
use serde::{Deserialize, Serialize};
use utoipa::ToSchema;
use super::UserStream;
/// Represents total counts of relationships of a user.
#[derive(Serialize, Deserialize, ToSchema, Debug, Default)]
pub struct UserCounts {
// The number of tags assigned to other entities by the user (e.g. user, posts)
pub tagged: u32,
// User received tags counts
pub tags: u32,
// Distinct tags where the user was referenced
pub unique_tags: u32,
pub posts: u32,
pub replies: u32,
pub following: u32,
pub followers: u32,
pub friends: u32,
pub bookmarks: u32,
}
impl RedisOps for UserCounts {}
impl UserCounts {
/// Retrieves counts by user ID, first trying to get from Redis, then from Neo4j if not found.
pub async fn get_by_id(user_id: &str) -> Result<Option<UserCounts>, DynError> {
match Self::get_from_index(user_id).await? {
Some(counts) => Ok(Some(counts)),
None => {
let graph_response = Self::get_from_graph(user_id).await?;
if let Some(user_counts) = graph_response {
user_counts.put_to_index(user_id).await?;
return Ok(Some(user_counts));
}
Ok(None)
}
}
}
/// Retrieves the counts from Neo4j.
pub async fn get_from_graph(user_id: &str) -> Result<Option<UserCounts>, DynError> {
let mut result;
{
let graph = get_neo4j_graph()?;
let query = queries::get::user_counts(user_id);
let graph = graph.lock().await;
result = graph.execute(query).await?;
}
if let Some(row) = result.next().await? {
let user_exists: bool = row.get("exists").unwrap_or(false);
if user_exists {
match row.get("counts") {
Ok(user_counts) => return Ok(Some(user_counts)),
// Like this we give a chance, in the next request to populate index
// If we populate the cache with default value, from that point we will have
// inconsistent state
Err(_e) => return Ok(None),
}
}
}
Ok(None)
}
pub async fn get_from_index(user_id: &str) -> Result<Option<UserCounts>, DynError> {
if let Some(user_counts) = Self::try_from_index_json(&[user_id], None).await? {
return Ok(Some(user_counts));
}
Ok(None)
}
pub async fn put_to_index(&self, user_id: &str) -> Result<(), DynError> {
self.put_index_json(&[user_id], None, None).await?;
UserStream::add_to_most_followed_sorted_set(user_id, self).await?;
UserStream::add_to_influencers_sorted_set(user_id, self).await?;
Ok(())
}
pub async fn update_index_field(
author_id: &str,
field: &str,
action: JsonAction,
) -> Result<(), DynError> {
Self::modify_json_field(&[author_id], field, action).await?;
Ok(())
}
/// Updates a user's counts index field and conditionally updates ranking sets
/// based on follower, tag, or post counts.
///
/// # Arguments
///
/// * `user_id` - The unique identifier of the user whose index field is being updated.
/// * `field` - The name of the user-related field to update (e.g., `"followers"`, `"tags"`, `"posts"`).
/// * `action` - The action to perform on the field (increment or decrement).
/// * `tag_label` - An optional tag label used to check membership in the user's tag-related sorted set. Important if we want to update the unique_tags field
///
/// # Behavior
///
/// - Conditional Update Based on `tag_label`
/// - Update User Counts Index
/// - Update Ranking Sets for Specific Fields
pub async fn update(
user_id: &str,
field: &str,
action: JsonAction,
tag_label: Option<&str>,
) -> Result<(), DynError> {
// This condition applies only when updating `unique_tags`
if let Some(label) = tag_label {
let index_parts = [&USER_TAGS_KEY_PARTS[..], &[user_id]].concat();
let score = Self::check_sorted_set_member(None, &index_parts, &[label]).await?;
match (score, &action) {
// If tag value is less than 1, `unique_tags` can be incremented or decremented
(Some(tag_value), _) if tag_value < 1 => (),
// Incrementing `unique_tags` is also allowed when the tag value doesn't exist yet in the sorted set
(None, JsonAction::Increment(_)) => (),
// Do not update the index
_ => return Ok(()),
}
}
// Update user counts index
Self::update_index_field(user_id, field, action).await?;
// Just update influencer and most followed indexes, when that fields are updated
if field == "followers" || field == "tags" || field == "posts" {
let exist_count = Self::get_by_id(user_id).await?;
if let Some(user_counts) = exist_count {
UserStream::add_to_influencers_sorted_set(user_id, &user_counts).await?;
// Increment followers
if field == "followers" {
UserStream::add_to_most_followed_sorted_set(user_id, &user_counts).await?
}
}
}
Ok(())
}
pub async fn reindex(author_id: &str) -> Result<(), DynError> {
match Self::get_from_graph(author_id).await? {
Some(counts) => counts.put_to_index(author_id).await?,
None => tracing::error!("{}: Could not found user counts in the graph", author_id),
}
Ok(())
}
pub async fn delete(user_id: &str) -> Result<(), DynError> {
// Delete user_details on Redis
Self::remove_from_index_multiple_json(&[&[user_id]]).await?;
Ok(())
}
}