Integrate Redis for caching and address management

server/Cargo.toml

@@ -42,3 +42,6 @@ utoipa-axum = "0.2.0"
 utoipa-scalar = { version = "0.3.0", features = ["axum"] }
 sha2 = "0.10.9"
 uuid = { version = "1.18.1", features = ["v4"] }
+bb8-redis = "0.24"
+redis-macros = "0.5.6"
+redis = "0.32.7"

server/docker-compose.yml

@@ -16,10 +16,12 @@ services:
       JWT_SECRET: "${JWT_SECRET}"
       JWT_EXPIRED_IN: "${JWT_EXPIRED_IN}"
       JWT_MAXAGE: "${JWT_MAXAGE}"
+      REDIS_URL: "${REDIS_URL}"
     ports:
       - "8080:8080"
     depends_on:
       - db
+      - redis
 
   db:
     image: postgres:16
@@ -31,5 +33,11 @@ services:
     volumes:
       - pgdata:/var/lib/postgresql/data
 
+  redis:
+    image: redis:7-alpine
+    restart: always
+    ports:
+      - "6379:6379"
+
 volumes:
   pgdata:

server/src/lib.rs

@@ -1,11 +1,11 @@
 pub mod libs {
     pub mod auth;
-    pub mod cache;
     pub mod config;
     pub mod db;
     pub mod error;
     pub mod logging;
     pub mod middleware;
+    pub mod redis;
     pub mod router;
     pub mod utopia;
 }
@@ -28,13 +28,74 @@ pub mod util {
     pub mod validate_address;
 }
 
-use crate::libs::{cache::Cache, config::Env};
+use crate::libs::{config::Env, error::ApiError, redis::RedisPool};
 
+use redis::AsyncCommands;
 use sqlx::PgPool;
 
 #[derive(Clone)]
 pub struct AppState {
     pub db: PgPool,
-    pub cache: Cache,
+    pub redis: RedisPool,
     pub env: Env,
 }
+
+impl AppState {
+    async fn add_group_address(&self, addr: &str) -> Result<(), ApiError> {
+        let mut conn = self.redis.get().await.expect("Error connecting to redis");
+        conn.sadd("group_funding_address", addr).await.map_err(|e| {
+            tracing::error!("Error adding group address to redis: {:?}", e);
+            ApiError::Internal("Error adding group address to redis")
+        })
+    }
+    async fn has_group_address(&self, addr: &str) -> Result<bool, ApiError> {
+        let mut conn = self.redis.get().await.expect("Error connecting to redis");
+        conn.sismember("group_funding_address", addr)
+            .await
+            .map_err(|e| {
+                tracing::error!("Error checking group address in redis: {:?}", e);
+                ApiError::Internal("Error checking group address in redis")
+            })
+    }
+
+    async fn get_all_group_addresses(&self) -> Result<Vec<String>, ApiError> {
+        let mut conn = self.redis.get().await.expect("Error connecting to redis");
+        let addresses: Vec<String> =
+            conn.smembers("group_funding_addresses")
+                .await
+                .map_err(|e| {
+                    tracing::error!("Error getting group addresses from redis: {:?}", e);
+                    ApiError::Internal("Error getting group addresses from redis")
+                })?;
+        Ok(addresses)
+    }
+
+    async fn add_crowd_funding_address(&self, addr: &str) -> Result<(), ApiError> {
+        let mut conn = self.redis.get().await.expect("Error connecting to redis");
+        conn.sadd("crowd_funding_address", addr).await.map_err(|e| {
+            tracing::error!("Error adding crowd funding address to redis: {:?}", e);
+            ApiError::Internal("Error adding crowd funding address to redis")
+        })
+    }
+    async fn _has_crowd_funding_address(&self, addr: &str) -> Result<bool, ApiError> {
+        let mut conn = self.redis.get().await.expect("Error connecting to redis");
+        conn.sismember("crowd_funding_address", addr)
+            .await
+            .map_err(|e| {
+                tracing::error!("Error checking crowd funding address in redis: {:?}", e);
+                ApiError::Internal("Error checking crowd funding address in redis")
+            })
+    }
+
+    async fn get_all_crowd_funding_addresses(&self) -> Result<Vec<String>, ApiError> {
+        let mut conn = self.redis.get().await.expect("Error connecting to redis");
+        let addresses: Vec<String> =
+            conn.smembers("crowd_funding_addresses")
+                .await
+                .map_err(|e| {
+                    tracing::error!("Error getting crowd funding addresses from redis: {:?}", e);
+                    ApiError::Internal("Error getting crowd funding addresses from redis")
+                })?;
+        Ok(addresses)
+    }
+}

server/src/libs/redis.rs

@@ -0,0 +1,78 @@
+use std::env;
+
+use bb8_redis::{RedisConnectionManager, bb8};
+use redis::AsyncCommands;
+use sqlx::PgPool;
+
+pub type RedisPool = bb8::Pool<RedisConnectionManager>;
+
+pub async fn init_redis(pool: &PgPool) -> RedisPool {
+    let redis_url = env::var("REDIS_URL").unwrap_or_else(|_| "redis://localhost:6379".to_string());
+    let manager = RedisConnectionManager::new(redis_url).expect("init redis failed");
+
+    let redis_pool = bb8::Pool::builder()
+        .max_size(100)
+        .build(manager)
+        .await
+        .expect("Failed to build Redis pool");
+
+    {
+        let crowd_funding_addresses: Vec<String> =
+            sqlx::query_scalar("SELECT pool_address FROM crowd_funding")
+                .fetch_all(pool)
+                .await
+                .expect("Failed to get crowd funding addresses");
+
+        let group_funding_addresses: Vec<String> =
+            sqlx::query_scalar("SELECT group_address FROM groups")
+                .fetch_all(pool)
+                .await
+                .expect("Failed to get group addresses");
+
+        let mut conn = redis_pool
+            .get()
+            .await
+            .expect("Failed to get Redis connection");
+
+        let _: () = conn
+            .sadd("crowd_funding_addresses", &crowd_funding_addresses)
+            .await
+            .expect("Failed to store crowd funding addresses");
+
+        let _: () = conn
+            .sadd("group_funding_addresses", &group_funding_addresses)
+            .await
+            .expect("Failed to store group addresses");
+    }
+
+    redis_pool
+}
+
+pub async fn refresh_cache(pool: &PgPool, redis_pool: &RedisPool) {
+    let crowd_funding_addresses: Vec<String> =
+        sqlx::query_scalar("SELECT pool_address FROM crowd_funding")
+            .fetch_all(pool)
+            .await
+            .expect("Failed to get crowd funding addresses");
+
+    let group_funding_addresses: Vec<String> =
+        sqlx::query_scalar("SELECT group_address FROM groups")
+            .fetch_all(pool)
+            .await
+            .expect("Failed to get group addresses");
+
+    let mut conn = redis_pool
+        .get()
+        .await
+        .expect("Failed to get Redis connection");
+
+    let _: () = conn
+        .sadd("crowd_funding_addresses", &crowd_funding_addresses)
+        .await
+        .expect("Failed to store crowd funding addresses");
+
+    let _: () = conn
+        .sadd("group_funding_addresses", &group_funding_addresses)
+        .await
+        .expect("Failed to store group addresses");
+}

server/src/main.rs

@@ -1,6 +1,6 @@
 use server::{
     AppState,
-    libs::{cache::init_cache, config::Env, db::Db, logging::init_tracing, router::router},
+    libs::{config::Env, db::Db, logging::init_tracing, redis, router::router},
 };
 use tokio::net::TcpListener;
 
@@ -16,27 +16,27 @@ async fn main() {
     tracing::debug!("Initializing db");
     let db = Db::new().await.expect("Failed to initialize DB");
 
-    tracing::debug!("Initializing cache");
-    let cache = init_cache(&db.pool.clone()).await;
+    tracing::debug!("Initializing redis");
+    let redis_pool = redis::init_redis(&db.pool.clone()).await;
 
     let config = AppState {
         db: db.pool.clone(),
-        cache,
+        redis: redis_pool,
         env,
     };
 
     {
-        let cache = config.cache.clone();
+        let redis_pool = config.redis.clone();
         let db = config.db.clone();
         tokio::spawn(async move {
             let mut interval = tokio::time::interval(std::time::Duration::from_secs(300));
             loop {
-                let new_cache = init_cache(&db).await;
-                *cache.write().await = new_cache.read().await.clone();
+                tracing::info!("Refreshing redis cache");
+                redis::refresh_cache(&db, &redis_pool).await;
                 interval.tick().await;
             }
         });
-        tracing::info!("Cache Refreshed");
+        tracing::info!("Started cache refresher task");
     }
 
     tracing::debug!("Running Migrations");