Feat: Add redis backend

Cargo.toml

@@ -103,7 +103,7 @@ autoconnect_settings = { path = "./autoconnect/autoconnect-settings" }
 autoconnect_web = { path = "./autoconnect/autoconnect-web" }
 autoconnect_ws = { path = "./autoconnect/autoconnect-ws" }
 autoconnect_ws_clientsm = { path = "./autoconnect/autoconnect-ws/autoconnect-ws-clientsm" }
-autopush_common = { path = "./autopush-common", features = ["bigtable"] }
+autopush_common = { path = "./autopush-common" }
 
 [profile.release]
 debug = 1

Dockerfile

@@ -2,6 +2,7 @@
 # RUST_VER
 FROM rust:1.83-bookworm AS builder
 ARG CRATE
+ARG BUILD_ARGS
 
 ADD . /app
 WORKDIR /app
@@ -16,7 +17,7 @@ RUN \
     cargo --version && \
     rustc --version && \
     mkdir -m 755 bin && \
-    cargo install --path $CRATE --locked --root /app
+    cargo install --path $CRATE $BUILD_ARGS --locked --root /app
 
 
 FROM debian:bookworm-slim

autoconnect/Cargo.toml

@@ -56,4 +56,5 @@ docopt = "1.1"
 default = ["bigtable"]
 bigtable = ["autopush_common/bigtable", "autoconnect_settings/bigtable"]
 emulator = ["bigtable"]
+redis = ["autopush_common/redis", "autoconnect_settings/redis"]
 log_vapid = []

autoconnect/autoconnect-settings/Cargo.toml

@@ -25,3 +25,4 @@ autopush_common.workspace = true
 # specify the default via the calling crate, in order to simplify default chains.
 bigtable = ["autopush_common/bigtable"]
 emulator = ["bigtable"]
+redis = ["autopush_common/redis"]

autoconnect/autoconnect-settings/src/app_state.rs

@@ -2,6 +2,8 @@ use std::{sync::Arc, time::Duration};
 
 #[cfg(feature = "bigtable")]
 use autopush_common::db::bigtable::BigTableClientImpl;
+#[cfg(feature = "redis")]
+use autopush_common::db::redis::RedisClientImpl;
 use cadence::StatsdClient;
 use config::ConfigError;
 use fernet::{Fernet, MultiFernet};
@@ -78,6 +80,11 @@ impl AppState {
                 client.spawn_sweeper(Duration::from_secs(30));
                 Box::new(client)
             }
+            #[cfg(feature = "redis")]
+            StorageType::Redis => Box::new(
+                RedisClientImpl::new(metrics.clone(), &db_settings)
+                    .map_err(|e| ConfigError::Message(e.to_string()))?,
+            ),
             _ => panic!(
                 "Invalid Storage type {:?}. Check {}__DB_DSN.",
                 storage_type,

autoconnect/autoconnect-settings/src/lib.rs

@@ -11,7 +11,6 @@ use config::{Config, ConfigError, Environment, File};
 use fernet::Fernet;
 use lazy_static::lazy_static;
 use serde::{Deserialize, Deserializer};
-use serde_json::json;
 
 use autopush_common::util::deserialize_u32_to_duration;
 
@@ -218,6 +217,7 @@ impl Settings {
         Ok(())
     }
 
+    #[cfg(feature = "bigtable")]
     pub fn test_settings() -> Self {
         let db_dsn = Some("grpc://localhost:8086".to_string());
         // BigTable DB_SETTINGS.
@@ -234,6 +234,17 @@ impl Settings {
             ..Default::default()
         }
     }
+
+    #[cfg(all(feature = "redis", not(feature = "bigtable")))]
+    pub fn test_settings() -> Self {
+        let db_dsn = Some("redis://localhost".to_string());
+        let db_settings = "".to_string();
+        Self {
+            db_dsn,
+            db_settings,
+            ..Default::default()
+        }
+    }
 }
 
 fn deserialize_f64_to_duration<'de, D>(deserializer: D) -> Result<Duration, D::Error>

autoendpoint/Cargo.toml

@@ -75,6 +75,8 @@ bigtable = ["autopush_common/bigtable"]
 # enable emulator to call locally run data store.
 emulator = ["bigtable"]
 
+redis = ["autopush_common/redis"]
+
 # Enable "stub" router for local testing purposes.
 # The "stub" will return specified error strings or success
 # depending on which `app_id` client is called based on the registration

autoendpoint/src/server.rs

@@ -13,6 +13,8 @@ use serde_json::json;
 
 #[cfg(feature = "bigtable")]
 use autopush_common::db::bigtable::BigTableClientImpl;
+#[cfg(feature = "redis")]
+use autopush_common::db::redis::RedisClientImpl;
 use autopush_common::{
     db::{client::DbClient, spawn_pool_periodic_reporter, DbSettings, StorageType},
     middleware::sentry::SentryWrapper,
@@ -77,6 +79,8 @@ impl Server {
                 client.spawn_sweeper(Duration::from_secs(30));
                 Box::new(client)
             }
+            #[cfg(feature = "redis")]
+            StorageType::Redis => Box::new(RedisClientImpl::new(metrics.clone(), &db_settings)?),
             _ => {
                 debug!("No idea what {:?} is", &db_settings.dsn);
                 return Err(ApiErrorKind::General(

autopush-common/Cargo.toml

@@ -59,6 +59,7 @@ grpcio = { version = "=0.13.0", features = ["openssl"], optional = true }
 grpcio-sys = { version = "=0.13.0", optional = true }
 protobuf = { version = "=2.28.0", optional = true } # grpcio does not support protobuf 3+
 form_urlencoded = { version = "1.2", optional = true }
+redis = { version = "0.27.6", features = ["aio", "tokio-comp"]}
 
 [dev-dependencies]
 mockito = "0.31"
@@ -80,3 +81,4 @@ bigtable = [
 emulator = [
     "bigtable",
 ] # used for testing big table, requires an external bigtable emulator running.
+redis = []

autopush-common/build.rs

@@ -1,5 +1,5 @@
 pub fn main() {
-    if !cfg!(feature = "bigtable") {
-        panic!("No database defined! Please compile with `features=bigtable`");
+    if !cfg!(feature = "bigtable") && !cfg!(feature = "redis") {
+        panic!("No database defined! Please compile with `features=bigtable` (or redis)");
     }
 }

autopush-common/src/db/error.rs

@@ -26,6 +26,10 @@ pub enum DbError {
     #[error("BigTable error: {0}")]
     BTError(#[from] BigTableError),
 
+    #[cfg(feature = "redis")]
+    #[error("Redis error {0}")]
+    RedisError(#[from] redis::RedisError),
+
     #[error("Connection failure: {0}")]
     ConnectionError(String),
 

autopush-common/src/db/mod.rs

@@ -24,6 +24,8 @@ pub mod bigtable;
 pub mod client;
 pub mod error;
 pub mod models;
+#[cfg(feature = "redis")]
+pub mod redis;
 pub mod reporter;
 pub mod routing;
 
@@ -45,13 +47,17 @@ pub enum StorageType {
     INVALID,
     #[cfg(feature = "bigtable")]
     BigTable,
+    #[cfg(feature = "redis")]
+    Redis,
 }
 
 impl From<&str> for StorageType {
     fn from(name: &str) -> Self {
         match name.to_lowercase().as_str() {
             #[cfg(feature = "bigtable")]
             "bigtable" => Self::BigTable,
+            #[cfg(feature = "redis")]
+            "redis" => Self::Redis,
             _ => Self::INVALID,
         }
     }
@@ -65,6 +71,8 @@ impl StorageType {
         let mut result: Vec<&str> = Vec::new();
         #[cfg(feature = "bigtable")]
         result.push("Bigtable");
+        #[cfg(feature = "redis")]
+        result.push("Redis");
         result
     }
 
@@ -90,6 +98,11 @@ impl StorageType {
             }
             return Self::BigTable;
         }
+        #[cfg(feature = "redis")]
+        if dsn.starts_with("redis") {
+            trace!("Found redis");
+            return Self::Redis;
+        }
         Self::INVALID
     }
 }

autopush-common/src/db/redis/mod.rs

@@ -0,0 +1,64 @@
+/// This uses redis as a storage and management
+/// system for Autopush Notifications and Routing information.
+///
+/// Keys for the data are
+/// `autopush/user/{uaid}` String to store the user data
+/// `autopush/co/{uaid}` u64 to store the last time the user has interacted with the server
+/// `autopush/channels/{uaid}` List to store the list of the channels of the user
+/// `autopush/msgs/{uaid}` SortedSet to store the list of the pending message ids for the user
+/// `autopush/msgs_exp/{uaid}` SortedSet to store the list of the pending message ids, ordered by expiry date, this is because SortedSet elements can't have independant expiry date
+/// `autopush/msg/{uaid}/{chidmessageid}`, with `{chidmessageid} == {chid}:{version}` String to store
+/// the content of the messages
+///
+mod redis_client;
+
+pub use redis_client::RedisClientImpl;
+
+use serde::Deserialize;
+use std::time::Duration;
+
+use crate::db::error::DbError;
+use crate::util::deserialize_opt_u32_to_duration;
+
+/// The settings for accessing the redis contents.
+#[derive(Clone, Debug, Deserialize)]
+pub struct RedisDbSettings {
+    #[serde(default)]
+    #[serde(deserialize_with = "deserialize_opt_u32_to_duration")]
+    pub timeout: Option<Duration>,
+}
+
+// Used by test, but we don't want available for release.
+#[allow(clippy::derivable_impls)]
+impl Default for RedisDbSettings {
+    fn default() -> Self {
+        Self {
+            timeout: Default::default(),
+        }
+    }
+}
+
+impl TryFrom<&str> for RedisDbSettings {
+    type Error = DbError;
+    fn try_from(setting_string: &str) -> Result<Self, Self::Error> {
+        let me: Self = match serde_json::from_str(setting_string) {
+            Ok(me) => me,
+            Err(e) if e.is_eof() => Self::default(),
+            Err(e) => Err(DbError::General(format!(
+                "Could not parse DdbSettings: {:?}",
+                e
+            )))?,
+        };
+        Ok(me)
+    }
+}
+
+mod tests {
+
+    #[test]
+    fn test_settings_parse() -> Result<(), crate::db::error::DbError> {
+        let settings = super::RedisDbSettings::try_from("{\"timeout\": 123}")?;
+        assert_eq!(settings.timeout, Some(std::time::Duration::from_secs(123)));
+        Ok(())
+    }
+}

autopush-common/src/db/redis/redis_client/error.rs

@@ -0,0 +1,7 @@
+use crate::db::error::DbError;
+
+impl From<serde_json::Error> for DbError {
+    fn from(err: serde_json::Error) -> Self {
+        DbError::Serialization(err.to_string())
+    }
+}