replace the HashMap structure with a JSON-based implementation

benches/watcher.rs

@@ -5,7 +5,7 @@ use pubky_app_specs::{PubkyAppUser, PubkyAppUserLink};
 use pubky_common::crypto::Keypair;
 use pubky_homeserver::Homeserver;
 use pubky_nexus::{
-    events::manager::{RetryManager, SenderChannel},
+    events::retry::manager::{RetryManager, SenderChannel},
     types::PubkyId,
     EventProcessor,
 };

src/db/kv/index/mod.rs

@@ -1,4 +1,3 @@
-pub mod hash_map;
 /// Module for redis Indexing operations split into modules by Redis types
 pub mod json;
 pub mod lists;

src/db/kv/traits.rs

@@ -681,42 +681,4 @@ pub trait RedisOps: Serialize + DeserializeOwned + Send + Sync {
         let key = key_parts.join(":");
         sorted_sets::get_lex_range("Sorted", &key, min, max, skip, limit).await
     }
-
-    // ############################################################
-    // ############ HASH MAP related functions ####################
-    // ############################################################
-
-    /// Inserts a value into a Redis hash map at the specified key and field.
-    /// # Arguments
-    /// * `prefix` - An optional string slice representing the prefix for the key
-    /// * `key_parts` - A slice of string slices representing parts of the key
-    /// * `field` - A string slice representing the field in the hash map where the value will be stored.
-    /// * `value` - A `String` containing the value to be stored in the hash map
-    async fn put_index_hash_map(
-        prefix: Option<&str>,
-        key_parts: &[&str],
-        field: &str,
-        value: String,
-    ) -> Result<(), DynError> {
-        let prefix = prefix.unwrap_or(SORTED_PREFIX);
-        let key = key_parts.join(":");
-        // Store the elements in the Redis sorted set
-        hash_map::put(prefix, &key, field, value).await
-    }
-
-    /// Retrieves a value from a Redis hash map at the specified key and field.
-    /// # Arguments
-    /// * `prefix` - An optional string slice representing the prefix for the key
-    /// * `key_parts` - A slice of string slices representing parts of the key
-    /// * `field` - A string slice representing the field in the hash map from which the value will be retrieved
-    async fn get_index_hash_map(
-        prefix: Option<&str>,
-        key_parts: &[&str],
-        field: &str,
-    ) -> Result<Option<String>, DynError> {
-        let prefix = prefix.unwrap_or(SORTED_PREFIX);
-        let key = key_parts.join(":");
-        // Store the elements in the Redis sorted set
-        hash_map::get(prefix, &key, field).await
-    }
 }

src/events/processor.rs

@@ -164,7 +164,7 @@ impl EventProcessor {
                     // This block is unlikely to be reached, as it would typically fail during the validation process.
                     return Ok(());
                 };
-                let index_key = format!("{} {}", event.event_type, index);
+                let index_key = format!("{}:{}", event.event_type, index);
                 let sender = self.sender.lock().await;
                 match sender
                     .send(SenderMessage::Add(index_key, retry_event))

src/events/retry/event.rs

@@ -1,3 +1,4 @@
+use async_trait::async_trait;
 use chrono::Utc;
 use serde::{Deserialize, Serialize};
 
@@ -6,6 +7,7 @@ use crate::{events::error::EventProcessorError, types::DynError, RedisOps};
 pub const RETRY_MAMAGER_PREFIX: &str = "RetryManager";
 pub const RETRY_MANAGER_EVENTS_INDEX: [&str; 1] = ["events"];
 pub const RETRY_MANAGER_STATE_INDEX: [&str; 1] = ["state"];
+pub const HOMESERVER_PROTOCOL: &str = "pubky:";
 pub const HOMESERVER_PUBLIC_REPOSITORY: &str = "pub";
 pub const HOMESERVER_APP_REPOSITORY: &str = "pubky.app";
 
@@ -17,7 +19,12 @@ pub struct RetryEvent {
     pub error_type: EventProcessorError,
 }
 
-impl RedisOps for RetryEvent {}
+#[async_trait]
+impl RedisOps for RetryEvent {
+    async fn prefix() -> String {
+        String::from(RETRY_MAMAGER_PREFIX)
+    }
+}
 
 impl RetryEvent {
     pub fn new(error_type: EventProcessorError) -> Self {
@@ -28,22 +35,27 @@ impl RetryEvent {
     }
 
     /// It processes a homeserver URI and extracts specific components to form a index key
-    /// in the format `"{pubkyId}:{repository_model}/{event_id}"`
+    /// in the format `"{pubkyId}:{repository_model}:{event_id}"`
     /// # Parameters
     /// - `event_uri`: A string slice representing the event URI to be processed
     pub fn generate_index_key(event_uri: &str) -> Option<String> {
         let parts: Vec<&str> = event_uri.split('/').collect();
         if parts.len() >= 7
-            && parts[0] == "pubky:"
+            && parts[0] == HOMESERVER_PROTOCOL
             && parts[3] == HOMESERVER_PUBLIC_REPOSITORY
             && parts[4] == HOMESERVER_APP_REPOSITORY
         {
-            Some(format!("{}:{}/{}", parts[2], parts[5], parts[6]))
+            Some(format!("{}:{}:{}", parts[2], parts[5], parts[6]))
         } else {
             None
         }
     }
 
+    /// Stores an event in both a sorted set and a JSON index in Redis.
+    /// It adds an event line to a Redis sorted set with a timestamp-based score
+    /// and also stores the event details in a separate JSON index for retrieval.
+    /// # Arguments
+    /// * `event_line` - A `String` representing the event line to be indexed.
     pub async fn put_to_index(&self, event_line: String) -> Result<(), DynError> {
         Self::put_index_sorted_set(
             &RETRY_MANAGER_EVENTS_INDEX,
@@ -54,23 +66,20 @@ impl RetryEvent {
         )
         .await?;
 
-        let event_serialized = serde_json::to_string(self)?;
+        let index = &[RETRY_MANAGER_STATE_INDEX, [&event_line]].concat();
+        self.put_index_json(index, None).await?;
 
-        Self::put_index_hash_map(
-            Some(RETRY_MAMAGER_PREFIX),
-            &RETRY_MANAGER_STATE_INDEX,
-            &event_line,
-            event_serialized,
-        )
-        .await?;
         Ok(())
     }
 
-    pub async fn check_uri(event_line: &str) -> Result<Option<isize>, DynError> {
+    /// Checks if a specific event exists in the Redis sorted set
+    /// # Arguments
+    /// * `event_index` - A `&str` representing the event index to check
+    pub async fn check_uri(event_index: &str) -> Result<Option<isize>, DynError> {
         if let Some(post_details) = Self::check_sorted_set_member(
             Some(RETRY_MAMAGER_PREFIX),
             &RETRY_MANAGER_EVENTS_INDEX,
-            &[event_line],
+            &[event_index],
         )
         .await?
         {
@@ -79,17 +88,14 @@ impl RetryEvent {
         Ok(None)
     }
 
-    pub async fn get_from_index(pubky_uri: &str) -> Result<Option<Self>, DynError> {
+    /// Retrieves an event from the JSON index in Redis based on its index
+    /// # Arguments
+    /// * `event_index` - A `&str` representing the event index to retrieve
+    pub async fn get_from_index(event_index: &str) -> Result<Option<Self>, DynError> {
         let mut found_event = None;
-        if let Some(event_state) = Self::get_index_hash_map(
-            Some(RETRY_MAMAGER_PREFIX),
-            &RETRY_MANAGER_STATE_INDEX,
-            pubky_uri,
-        )
-        .await?
-        {
-            let event = serde_json::from_str::<Self>(&event_state)?;
-            found_event = Some(event);
+        let index = &[RETRY_MANAGER_STATE_INDEX, [event_index]].concat();
+        if let Some(fail_event) = Self::try_from_index_json(index).await? {
+            found_event = Some(fail_event);
         }
         Ok(found_event)
     }

tests/watcher/tags/retry_post_tag.rs

@@ -62,7 +62,7 @@ async fn test_homeserver_post_tag_event_to_queue() -> Result<()> {
     tokio::time::sleep(Duration::from_millis(500)).await;
 
     let index_key = format!(
-        "{} {}",
+        "{}:{}",
         EventType::Put,
         RetryEvent::generate_index_key(&tag_url).unwrap()
     );

tests/watcher/tags/retry_user_tag.rs

@@ -51,7 +51,7 @@ async fn test_homeserver_user_tag_event_to_queue() -> Result<()> {
     tokio::time::sleep(Duration::from_millis(500)).await;
 
     let index_key = format!(
-        "{} {}",
+        "{}:{}",
         EventType::Put,
         RetryEvent::generate_index_key(&tag_url).unwrap()
     );