nexus_common/models/follow/
traits.rs1use crate::db::{execute_graph_operation, get_neo4j_graph, queries, OperationOutcome, RedisOps};
2use crate::types::DynError;
3use async_trait::async_trait;
4use chrono::Utc;
5use neo4rs::Query;
6
7#[async_trait]
8pub trait UserFollows: Sized + RedisOps + AsRef<[String]> + Default {
9 fn from_vec(vec: Vec<String>) -> Self;
10
11 async fn put_to_graph(
12 follower_id: &str,
13 followee_id: &str,
14 ) -> Result<OperationOutcome, DynError> {
15 let indexed_at = Utc::now().timestamp_millis();
16 let query = queries::put::create_follow(follower_id, followee_id, indexed_at);
17 execute_graph_operation(query).await
18 }
19
20 async fn get_by_id(
21 user_id: &str,
22 skip: Option<usize>,
23 limit: Option<usize>,
24 ) -> Result<Option<Self>, DynError> {
25 match Self::get_from_index(user_id, skip, limit).await? {
26 Some(connections) => Ok(Some(Self::from_vec(connections))),
27 None => {
28 let graph_response = Self::get_from_graph(user_id, skip, limit).await?;
29 if let Some(follows) = graph_response {
30 follows.put_to_index(user_id).await?;
31 return Ok(Some(follows));
32 }
33 Ok(None)
34 }
35 }
36 }
37
38 async fn get_from_graph(
39 user_id: &str,
40 skip: Option<usize>,
41 limit: Option<usize>,
42 ) -> Result<Option<Self>, DynError> {
43 let mut result;
44 {
45 let graph = get_neo4j_graph()?;
46 let query = Self::get_query(user_id, skip, limit);
47
48 let graph = graph.lock().await;
49 result = graph.execute(query).await?;
50 }
51
52 if let Some(row) = result.next().await? {
53 let user_exists: bool = row.get("user_exists").unwrap_or(false);
54 if !user_exists {
55 return Ok(None);
56 }
57
58 match row.get::<Option<Vec<String>>>(Self::get_ids_field_name()) {
59 Ok(response) => {
60 if let Some(connections) = response {
61 return Ok(Some(Self::from_vec(connections)));
62 } else {
63 return Ok(Some(Self::default()));
64 }
65 }
66 Err(_e) => return Ok(None),
67 }
68 } else {
69 Ok(None)
70 }
71 }
72
73 async fn get_from_index(
74 user_id: &str,
75 skip: Option<usize>,
76 limit: Option<usize>,
77 ) -> Result<Option<Vec<String>>, DynError> {
78 Self::try_from_index_set(&[user_id], skip, limit, None).await
79 }
80
81 async fn put_to_index(&self, user_id: &str) -> Result<(), DynError> {
82 let user_list_ref: Vec<&str> = self.as_ref().iter().map(|id| id.as_str()).collect();
83 Self::put_index_set(&[user_id], &user_list_ref, None, None).await
84 }
85
86 async fn reindex(user_id: &str) -> Result<(), DynError> {
87 match Self::get_from_graph(user_id, None, None).await? {
88 Some(follow) => follow.put_to_index(user_id).await?,
89 None => tracing::error!(
90 "{}: Could not found user follow relationship in the graph",
91 user_id
92 ),
93 }
94 Ok(())
95 }
96
97 async fn del_from_graph(
98 follower_id: &str,
99 followee_id: &str,
100 ) -> Result<OperationOutcome, DynError> {
101 let query = queries::del::delete_follow(follower_id, followee_id);
102 execute_graph_operation(query).await
103 }
104
105 async fn del_from_index(&self, user_id: &str) -> Result<(), DynError> {
106 self.remove_from_index_set(&[user_id]).await
107 }
108
109 fn get_query(user_id: &str, skip: Option<usize>, limit: Option<usize>) -> Query;
110
111 fn get_ids_field_name() -> &'static str;
112
113 async fn check(user_a_id: &str, user_b_id: &str) -> Result<bool, DynError> {
115 let user_a_key_parts = &[user_a_id][..];
116 let (_, follow) = Self::check_set_member(user_a_key_parts, user_b_id).await?;
117 Ok(follow)
118 }
119}