From bd31dc85f8e9cb01c1e1a4e49fd4735d24a6da04 Mon Sep 17 00:00:00 2001
From: rtkay123 <dev@kanjala.com>
Date: Sun, 10 Aug 2025 12:55:43 +0200
Subject: chore: collapse stack-up

---
 lib/warden-stack/src/cache/cluster.rs  | 52 +++++++++++++++++++++++++++
 lib/warden-stack/src/cache/sentinel.rs | 65 ++++++++++++++++++++++++++++++++++
 2 files changed, 117 insertions(+)
 create mode 100644 lib/warden-stack/src/cache/cluster.rs
 create mode 100644 lib/warden-stack/src/cache/sentinel.rs

(limited to 'lib/warden-stack/src/cache')

diff --git a/lib/warden-stack/src/cache/cluster.rs b/lib/warden-stack/src/cache/cluster.rs
new file mode 100644
index 0000000..91e3b24
--- /dev/null
+++ b/lib/warden-stack/src/cache/cluster.rs
@@ -0,0 +1,52 @@
+use redis::{
+    ErrorKind, FromRedisValue, IntoConnectionInfo, RedisError,
+    cluster::{ClusterClient, ClusterClientBuilder},
+    cluster_routing::{MultipleNodeRoutingInfo, ResponsePolicy, RoutingInfo},
+};
+
+/// ConnectionManager that implements `bb8::ManageConnection` and supports
+/// asynchronous clustered connections via `redis_cluster_async::Connection`
+#[derive(Clone)]
+pub struct RedisClusterConnectionManager {
+    client: ClusterClient,
+}
+
+impl RedisClusterConnectionManager {
+    pub fn new<T: IntoConnectionInfo>(
+        info: T,
+    ) -> Result<RedisClusterConnectionManager, RedisError> {
+        Ok(RedisClusterConnectionManager {
+            client: ClusterClientBuilder::new(vec![info]).retries(0).build()?,
+        })
+    }
+}
+
+impl bb8::ManageConnection for RedisClusterConnectionManager {
+    type Connection = redis::cluster_async::ClusterConnection;
+    type Error = RedisError;
+
+    async fn connect(&self) -> Result<Self::Connection, Self::Error> {
+        self.client.get_async_connection().await
+    }
+
+    async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
+        let pong = conn
+            .route_command(
+                &redis::cmd("PING"),
+                RoutingInfo::MultiNode((
+                    MultipleNodeRoutingInfo::AllMasters,
+                    Some(ResponsePolicy::OneSucceeded),
+                )),
+            )
+            .await
+            .and_then(|v| String::from_redis_value(&v))?;
+        match pong.as_str() {
+            "PONG" => Ok(()),
+            _ => Err((ErrorKind::ResponseError, "ping request").into()),
+        }
+    }
+
+    fn has_broken(&self, _: &mut Self::Connection) -> bool {
+        false
+    }
+}
diff --git a/lib/warden-stack/src/cache/sentinel.rs b/lib/warden-stack/src/cache/sentinel.rs
new file mode 100644
index 0000000..c9f787a
--- /dev/null
+++ b/lib/warden-stack/src/cache/sentinel.rs
@@ -0,0 +1,65 @@
+use redis::{
+    ErrorKind, IntoConnectionInfo, RedisError,
+    sentinel::{SentinelClient, SentinelNodeConnectionInfo, SentinelServerType},
+};
+use serde::Deserialize;
+use tokio::sync::Mutex;
+
+struct LockedSentinelClient(pub(crate) Mutex<SentinelClient>);
+
+/// ConnectionManager that implements `bb8::ManageConnection` and supports
+/// asynchronous Sentinel connections via `redis::sentinel::SentinelClient`
+pub struct RedisSentinelConnectionManager {
+    client: LockedSentinelClient,
+}
+
+impl RedisSentinelConnectionManager {
+    pub fn new<T: IntoConnectionInfo>(
+        info: Vec<T>,
+        service_name: String,
+        node_connection_info: Option<SentinelNodeConnectionInfo>,
+    ) -> Result<RedisSentinelConnectionManager, RedisError> {
+        Ok(RedisSentinelConnectionManager {
+            client: LockedSentinelClient(Mutex::new(SentinelClient::build(
+                info,
+                service_name,
+                node_connection_info,
+                SentinelServerType::Master,
+            )?)),
+        })
+    }
+}
+
+impl bb8::ManageConnection for RedisSentinelConnectionManager {
+    type Connection = redis::aio::MultiplexedConnection;
+    type Error = RedisError;
+
+    async fn connect(&self) -> Result<Self::Connection, Self::Error> {
+        self.client.0.lock().await.get_async_connection().await
+    }
+
+    async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
+        let pong: String = redis::cmd("PING").query_async(conn).await?;
+        match pong.as_str() {
+            "PONG" => Ok(()),
+            _ => Err((ErrorKind::ResponseError, "ping request").into()),
+        }
+    }
+
+    fn has_broken(&self, _: &mut Self::Connection) -> bool {
+        false
+    }
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
+pub struct SentinelConfig {
+    #[serde(rename = "sentinel_service_name")]
+    pub service_name: String,
+    #[serde(default)]
+    pub redis_tls_mode_secure: bool,
+    pub redis_db: Option<i64>,
+    pub redis_username: Option<String>,
+    pub redis_password: Option<String>,
+    #[serde(default)]
+    pub redis_use_resp3: bool,
+}
-- 
cgit v1.2.3