Bug: [FIX] add a fallback to key without prefix for redis



To keep the tenant data isolated from each other, We are using a prefix for Redis keys i.e A key which would be saved as card123 without multitenancy would be saved as public:card123 with Multitenancy.

If the user makes a payment on the older pod where MultiTenancy is disabled and  client_secret is saved without a prefix, Now if the confirm calls goes to the pod where Multitenancy is enabled the Payment would fail with Unauthorized. This is just one of the instances where the payments would fail.
diff --git a/crates/drainer/src/health_check.rs b/crates/drainer/src/health_check.rs
index 2ca2c1cc79c..aaf455663f5 100644
--- a/crates/drainer/src/health_check.rs
+++ b/crates/drainer/src/health_check.rs
@@ -165,21 +165,21 @@ impl HealthCheckInterface for Store {
         let redis_conn = self.redis_conn.clone();
 
         redis_conn
-            .serialize_and_set_key_with_expiry("test_key", "test_value", 30)
+            .serialize_and_set_key_with_expiry(&"test_key".into(), "test_value", 30)
             .await
             .change_context(HealthCheckRedisError::SetFailed)?;
 
         logger::debug!("Redis set_key was successful");
 
         redis_conn
-            .get_key::<()>("test_key")
+            .get_key::<()>(&"test_key".into())
             .await
             .change_context(HealthCheckRedisError::GetFailed)?;
 
         logger::debug!("Redis get_key was successful");
 
         redis_conn
-            .delete_key("test_key")
+            .delete_key(&"test_key".into())
             .await
             .change_context(HealthCheckRedisError::DeleteFailed)?;
 
@@ -187,7 +187,7 @@ impl HealthCheckInterface for Store {
 
         redis_conn
             .stream_append_entry(
-                TEST_STREAM_NAME,
+                &TEST_STREAM_NAME.into(),
                 &redis_interface::RedisEntryId::AutoGeneratedID,
                 TEST_STREAM_DATA.to_vec(),
             )
@@ -216,7 +216,7 @@ impl HealthCheckInterface for Store {
 
         redis_conn
             .stream_trim_entries(
-                TEST_STREAM_NAME,
+                &TEST_STREAM_NAME.into(),
                 (
                     redis_interface::StreamCapKind::MinID,
                     redis_interface::StreamCapTrim::Exact,
diff --git a/crates/drainer/src/stream.rs b/crates/drainer/src/stream.rs
index f5b41c53672..9e092b038e8 100644
--- a/crates/drainer/src/stream.rs
+++ b/crates/drainer/src/stream.rs
@@ -31,7 +31,7 @@ impl Store {
 
         match self
             .redis_conn
-            .set_key_if_not_exists_with_expiry(stream_key_flag.as_str(), true, None)
+            .set_key_if_not_exists_with_expiry(&stream_key_flag.as_str().into(), true, None)
             .await
         {
             Ok(resp) => resp == redis::types::SetnxReply::KeySet,
@@ -43,7 +43,7 @@ impl Store {
     }
 
     pub async fn make_stream_available(&self, stream_name_flag: &str) -> errors::DrainerResult<()> {
-        match self.redis_conn.delete_key(stream_name_flag).await {
+        match self.redis_conn.delete_key(&stream_name_flag.into()).await {
             Ok(redis::DelReply::KeyDeleted) => Ok(()),
             Ok(redis::DelReply::KeyNotDeleted) => {
                 logger::error!("Tried to unlock a stream which is already unlocked");
@@ -87,14 +87,14 @@ impl Store {
             common_utils::date_time::time_it::<errors::DrainerResult<_>, _, _>(|| async {
                 let trim_result = self
                     .redis_conn
-                    .stream_trim_entries(stream_name, (trim_kind, trim_type, trim_id))
+                    .stream_trim_entries(&stream_name.into(), (trim_kind, trim_type, trim_id))
                     .await
                     .map_err(errors::DrainerError::from)?;
 
                 // Since xtrim deletes entries below given id excluding the given id.
                 // Hence, deleting the minimum entry id
                 self.redis_conn
-                    .stream_delete_entries(stream_name, minimum_entry_id)
+                    .stream_delete_entries(&stream_name.into(), minimum_entry_id)
                     .await
                     .map_err(errors::DrainerError::from)?;
 
diff --git a/crates/redis_interface/Cargo.toml b/crates/redis_interface/Cargo.toml
index f9159564830..87f5721365c 100644
--- a/crates/redis_interface/Cargo.toml
+++ b/crates/redis_interface/Cargo.toml
@@ -7,6 +7,9 @@ rust-version.workspace = true
 readme = "README.md"
 license.workspace = true
 
+[features]
+multitenancy_fallback = []
+
 [dependencies]
 error-stack = "0.4.1"
 fred = { version = "7.1.2", features = ["metrics", "partial-tracing", "subscriber-client", "check-unresponsive"] }
diff --git a/crates/redis_interface/src/commands.rs b/crates/redis_interface/src/commands.rs
index 56e9ab98104..affd2b0faa9 100644
--- a/crates/redis_interface/src/commands.rs
+++ b/crates/redis_interface/src/commands.rs
@@ -17,8 +17,7 @@ use fred::{
     prelude::{LuaInterface, RedisErrorKind},
     types::{
         Expiration, FromRedis, MultipleIDs, MultipleKeys, MultipleOrderedPairs, MultipleStrings,
-        MultipleValues, RedisKey, RedisMap, RedisValue, ScanType, Scanner, SetOptions, XCap,
-        XReadResponse,
+        MultipleValues, RedisMap, RedisValue, ScanType, Scanner, SetOptions, XCap, XReadResponse,
     },
 };
 use futures::StreamExt;
@@ -26,7 +25,7 @@ use tracing::instrument;
 
 use crate::{
     errors,
-    types::{DelReply, HsetnxReply, MsetnxReply, RedisEntryId, SaddReply, SetnxReply},
+    types::{DelReply, HsetnxReply, MsetnxReply, RedisEntryId, RedisKey, SaddReply, SetnxReply},
 };
 
 impl super::RedisConnectionPool {
@@ -37,15 +36,16 @@ impl super::RedisConnectionPool {
             format!("{}:{}", self.key_prefix, key)
         }
     }
+
     #[instrument(level = "DEBUG", skip(self))]
-    pub async fn set_key<V>(&self, key: &str, value: V) -> CustomResult<(), errors::RedisError>
+    pub async fn set_key<V>(&self, key: &RedisKey, value: V) -> CustomResult<(), errors::RedisError>
     where
         V: TryInto<RedisValue> + Debug + Send + Sync,
         V::Error: Into<fred::error::RedisError> + Send + Sync,
     {
         self.pool
             .set(
-                self.add_prefix(key),
+                key.tenant_aware_key(self),
                 value,
                 Some(Expiration::EX(self.config.default_ttl.into())),
                 None,
@@ -57,7 +57,7 @@ impl super::RedisConnectionPool {
 
     pub async fn set_key_without_modifying_ttl<V>(
         &self,
-        key: &str,
+        key: &RedisKey,
         value: V,
     ) -> CustomResult<(), errors::RedisError>
     where
@@ -65,7 +65,13 @@ impl super::RedisConnectionPool {
         V::Error: Into<fred::error::RedisError> + Send + Sync,
     {
         self.pool
-            .set(key, value, Some(Expiration::KEEPTTL), None, false)
+            .set(
+                key.tenant_aware_key(self),
+                value,
+                Some(Expiration::KEEPTTL),
+                None,
+                false,
+            )
             .await
             .change_context(errors::RedisError::SetFailed)
     }
@@ -87,7 +93,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn serialize_and_set_key_if_not_exist<V>(
         &self,
-        key: &str,
+        key: &RedisKey,
         value: V,
         ttl: Option<i64>,
     ) -> CustomResult<SetnxReply, errors::RedisError>
@@ -104,7 +110,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn serialize_and_set_key<V>(
         &self,
-        key: &str,
+        key: &RedisKey,
         value: V,
     ) -> CustomResult<(), errors::RedisError>
     where
@@ -120,7 +126,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn serialize_and_set_key_without_modifying_ttl<V>(
         &self,
-        key: &str,
+        key: &RedisKey,
         value: V,
     ) -> CustomResult<(), errors::RedisError>
     where
@@ -137,7 +143,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn serialize_and_set_key_with_expiry<V>(
         &self,
-        key: &str,
+        key: &RedisKey,
         value: V,
         seconds: i64,
     ) -> CustomResult<(), errors::RedisError>
@@ -150,7 +156,7 @@ impl super::RedisConnectionPool {
 
         self.pool
             .set(
-                self.add_prefix(key),
+                key.tenant_aware_key(self),
                 serialized.as_slice(),
                 Some(Expiration::EX(seconds)),
                 None,
@@ -161,31 +167,67 @@ impl super::RedisConnectionPool {
     }
 
     #[instrument(level = "DEBUG", skip(self))]
-    pub async fn get_key<V>(&self, key: &str) -> CustomResult<V, errors::RedisError>
+    pub async fn get_key<V>(&self, key: &RedisKey) -> CustomResult<V, errors::RedisError>
     where
         V: FromRedis + Unpin + Send + 'static,
     {
-        self.pool
-            .get(self.add_prefix(key))
+        match self
+            .pool
+            .get(key.tenant_aware_key(self))
             .await
             .change_context(errors::RedisError::GetFailed)
+        {
+            Ok(v) => Ok(v),
+            Err(_err) => {
+                #[cfg(not(feature = "multitenancy_fallback"))]
+                {
+                    Err(_err)
+                }
+
+                #[cfg(feature = "multitenancy_fallback")]
+                {
+                    self.pool
+                        .get(key.tenant_unaware_key(self))
+                        .await
+                        .change_context(errors::RedisError::GetFailed)
+                }
+            }
+        }
     }
 
     #[instrument(level = "DEBUG", skip(self))]
-    pub async fn exists<V>(&self, key: &str) -> CustomResult<bool, errors::RedisError>
+    pub async fn exists<V>(&self, key: &RedisKey) -> CustomResult<bool, errors::RedisError>
     where
         V: Into<MultipleKeys> + Unpin + Send + 'static,
     {
-        self.pool
-            .exists(self.add_prefix(key))
+        match self
+            .pool
+            .exists(key.tenant_aware_key(self))
             .await
             .change_context(errors::RedisError::GetFailed)
+        {
+            Ok(v) => Ok(v),
+            Err(_err) => {
+                #[cfg(not(feature = "multitenancy_fallback"))]
+                {
+                    Err(_err)
+                }
+
+                #[cfg(feature = "multitenancy_fallback")]
+                {
+                    self.pool
+                        .exists(key.tenant_unaware_key(self))
+                        .await
+                        .change_context(errors::RedisError::GetFailed)
+                }
+            }
+        }
     }
 
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn get_and_deserialize_key<T>(
         &self,
-        key: &str,
+        key: &RedisKey,
         type_name: &'static str,
     ) -> CustomResult<T, errors::RedisError>
     where
@@ -201,19 +243,37 @@ impl super::RedisConnectionPool {
     }
 
     #[instrument(level = "DEBUG", skip(self))]
-    pub async fn delete_key(&self, key: &str) -> CustomResult<DelReply, errors::RedisError> {
-        self.pool
-            .del(self.add_prefix(key))
+    pub async fn delete_key(&self, key: &RedisKey) -> CustomResult<DelReply, errors::RedisError> {
+        match self
+            .pool
+            .del(key.tenant_aware_key(self))
             .await
             .change_context(errors::RedisError::DeleteFailed)
+        {
+            Ok(v) => Ok(v),
+            Err(_err) => {
+                #[cfg(not(feature = "multitenancy_fallback"))]
+                {
+                    Err(_err)
+                }
+
+                #[cfg(feature = "multitenancy_fallback")]
+                {
+                    self.pool
+                        .del(key.tenant_unaware_key(self))
+                        .await
+                        .change_context(errors::RedisError::DeleteFailed)
+                }
+            }
+        }
     }
 
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn delete_multiple_keys(
         &self,
-        keys: &[String],
+        keys: &[RedisKey],
     ) -> CustomResult<Vec<DelReply>, errors::RedisError> {
-        let futures = keys.iter().map(|key| self.pool.del(self.add_prefix(key)));
+        let futures = keys.iter().map(|key| self.delete_key(key));
 
         let del_result = futures::future::try_join_all(futures)
             .await
@@ -225,7 +285,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn set_key_with_expiry<V>(
         &self,
-        key: &str,
+        key: &RedisKey,
         value: V,
         seconds: i64,
     ) -> CustomResult<(), errors::RedisError>
@@ -235,7 +295,7 @@ impl super::RedisConnectionPool {
     {
         self.pool
             .set(
-                self.add_prefix(key),
+                key.tenant_aware_key(self),
                 value,
                 Some(Expiration::EX(seconds)),
                 None,
@@ -248,7 +308,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn set_key_if_not_exists_with_expiry<V>(
         &self,
-        key: &str,
+        key: &RedisKey,
         value: V,
         seconds: Option<i64>,
     ) -> CustomResult<SetnxReply, errors::RedisError>
@@ -258,7 +318,7 @@ impl super::RedisConnectionPool {
     {
         self.pool
             .set(
-                self.add_prefix(key),
+                key.tenant_aware_key(self),
                 value,
                 Some(Expiration::EX(
                     seconds.unwrap_or(self.config.default_ttl.into()),
@@ -273,11 +333,11 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn set_expiry(
         &self,
-        key: &str,
+        key: &RedisKey,
         seconds: i64,
     ) -> CustomResult<(), errors::RedisError> {
         self.pool
-            .expire(self.add_prefix(key), seconds)
+            .expire(key.tenant_aware_key(self), seconds)
             .await
             .change_context(errors::RedisError::SetExpiryFailed)
     }
@@ -285,11 +345,11 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn set_expire_at(
         &self,
-        key: &str,
+        key: &RedisKey,
         timestamp: i64,
     ) -> CustomResult<(), errors::RedisError> {
         self.pool
-            .expire_at(self.add_prefix(key), timestamp)
+            .expire_at(key.tenant_aware_key(self), timestamp)
             .await
             .change_context(errors::RedisError::SetExpiryFailed)
     }
@@ -297,7 +357,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn set_hash_fields<V>(
         &self,
-        key: &str,
+        key: &RedisKey,
         values: V,
         ttl: Option<i64>,
     ) -> CustomResult<(), errors::RedisError>
@@ -307,7 +367,7 @@ impl super::RedisConnectionPool {
     {
         let output: Result<(), _> = self
             .pool
-            .hset(self.add_prefix(key), values)
+            .hset(key.tenant_aware_key(self), values)
             .await
             .change_context(errors::RedisError::SetHashFailed);
         // setting expiry for the key
@@ -321,7 +381,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn set_hash_field_if_not_exist<V>(
         &self,
-        key: &str,
+        key: &RedisKey,
         field: &str,
         value: V,
         ttl: Option<u32>,
@@ -332,7 +392,7 @@ impl super::RedisConnectionPool {
     {
         let output: Result<HsetnxReply, _> = self
             .pool
-            .hsetnx(self.add_prefix(key), field, value)
+            .hsetnx(key.tenant_aware_key(self), field, value)
             .await
             .change_context(errors::RedisError::SetHashFieldFailed);
 
@@ -348,7 +408,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn serialize_and_set_hash_field_if_not_exist<V>(
         &self,
-        key: &str,
+        key: &RedisKey,
         field: &str,
         value: V,
         ttl: Option<u32>,
@@ -367,7 +427,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn serialize_and_set_multiple_hash_field_if_not_exist<V>(
         &self,
-        kv: &[(&str, V)],
+        kv: &[(&RedisKey, V)],
         field: &str,
         ttl: Option<u32>,
     ) -> CustomResult<Vec<HsetnxReply>, errors::RedisError>
@@ -387,7 +447,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn increment_fields_in_hash<T>(
         &self,
-        key: &str,
+        key: &RedisKey,
         fields_to_increment: &[(T, i64)],
     ) -> CustomResult<Vec<usize>, errors::RedisError>
     where
@@ -397,7 +457,7 @@ impl super::RedisConnectionPool {
         for (field, increment) in fields_to_increment.iter() {
             values_after_increment.push(
                 self.pool
-                    .hincrby(self.add_prefix(key), field.to_string(), *increment)
+                    .hincrby(key.tenant_aware_key(self), field.to_string(), *increment)
                     .await
                     .change_context(errors::RedisError::IncrementHashFieldFailed)?,
             )
@@ -409,14 +469,14 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn hscan(
         &self,
-        key: &str,
+        key: &RedisKey,
         pattern: &str,
         count: Option<u32>,
     ) -> CustomResult<Vec<String>, errors::RedisError> {
         Ok(self
             .pool
             .next()
-            .hscan::<&str, &str>(&self.add_prefix(key), pattern, count)
+            .hscan::<&str, &str>(&key.tenant_aware_key(self), pattern, count)
             .filter_map(|value| async move {
                 match value {
                     Ok(mut v) => {
@@ -440,14 +500,14 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn scan(
         &self,
-        pattern: &str,
+        pattern: &RedisKey,
         count: Option<u32>,
         scan_type: Option<ScanType>,
     ) -> CustomResult<Vec<String>, errors::RedisError> {
         Ok(self
             .pool
             .next()
-            .scan(&self.add_prefix(pattern), count, scan_type)
+            .scan(pattern.tenant_aware_key(self), count, scan_type)
             .filter_map(|value| async move {
                 match value {
                     Ok(mut v) => {
@@ -471,7 +531,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn hscan_and_deserialize<T>(
         &self,
-        key: &str,
+        key: &RedisKey,
         pattern: &str,
         count: Option<u32>,
     ) -> CustomResult<Vec<T>, errors::RedisError>
@@ -491,33 +551,69 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn get_hash_field<V>(
         &self,
-        key: &str,
+        key: &RedisKey,
         field: &str,
     ) -> CustomResult<V, errors::RedisError>
     where
         V: FromRedis + Unpin + Send + 'static,
     {
-        self.pool
-            .hget(self.add_prefix(key), field)
+        match self
+            .pool
+            .hget(key.tenant_aware_key(self), field)
             .await
             .change_context(errors::RedisError::GetHashFieldFailed)
+        {
+            Ok(v) => Ok(v),
+            Err(_err) => {
+                #[cfg(feature = "multitenancy_fallback")]
+                {
+                    self.pool
+                        .hget(key.tenant_unaware_key(self), field)
+                        .await
+                        .change_context(errors::RedisError::GetHashFieldFailed)
+                }
+
+                #[cfg(not(feature = "multitenancy_fallback"))]
+                {
+                    Err(_err)
+                }
+            }
+        }
     }
 
     #[instrument(level = "DEBUG", skip(self))]
-    pub async fn get_hash_fields<V>(&self, key: &str) -> CustomResult<V, errors::RedisError>
+    pub async fn get_hash_fields<V>(&self, key: &RedisKey) -> CustomResult<V, errors::RedisError>
     where
         V: FromRedis + Unpin + Send + 'static,
     {
-        self.pool
-            .hgetall(self.add_prefix(key))
+        match self
+            .pool
+            .hgetall(key.tenant_aware_key(self))
             .await
             .change_context(errors::RedisError::GetHashFieldFailed)
+        {
+            Ok(v) => Ok(v),
+            Err(_err) => {
+                #[cfg(feature = "multitenancy_fallback")]
+                {
+                    self.pool
+                        .hgetall(key.tenant_unaware_key(self))
+                        .await
+                        .change_context(errors::RedisError::GetHashFieldFailed)
+                }
+
+                #[cfg(not(feature = "multitenancy_fallback"))]
+                {
+                    Err(_err)
+                }
+            }
+        }
     }
 
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn get_hash_field_and_deserialize<V>(
         &self,
-        key: &str,
+        key: &RedisKey,
         field: &str,
         type_name: &'static str,
     ) -> CustomResult<V, errors::RedisError>
@@ -538,7 +634,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn sadd<V>(
         &self,
-        key: &str,
+        key: &RedisKey,
         members: V,
     ) -> CustomResult<SaddReply, errors::RedisError>
     where
@@ -546,7 +642,7 @@ impl super::RedisConnectionPool {
         V::Error: Into<fred::error::RedisError> + Send,
     {
         self.pool
-            .sadd(self.add_prefix(key), members)
+            .sadd(key.tenant_aware_key(self), members)
             .await
             .change_context(errors::RedisError::SetAddMembersFailed)
     }
@@ -554,7 +650,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn stream_append_entry<F>(
         &self,
-        stream: &str,
+        stream: &RedisKey,
         entry_id: &RedisEntryId,
         fields: F,
     ) -> CustomResult<(), errors::RedisError>
@@ -563,7 +659,7 @@ impl super::RedisConnectionPool {
         F::Error: Into<fred::error::RedisError> + Send + Sync,
     {
         self.pool
-            .xadd(self.add_prefix(stream), false, None, entry_id, fields)
+            .xadd(stream.tenant_aware_key(self), false, None, entry_id, fields)
             .await
             .change_context(errors::RedisError::StreamAppendFailed)
     }
@@ -571,14 +667,14 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn stream_delete_entries<Ids>(
         &self,
-        stream: &str,
+        stream: &RedisKey,
         ids: Ids,
     ) -> CustomResult<usize, errors::RedisError>
     where
         Ids: Into<MultipleStrings> + Debug + Send + Sync,
     {
         self.pool
-            .xdel(self.add_prefix(stream), ids)
+            .xdel(stream.tenant_aware_key(self), ids)
             .await
             .change_context(errors::RedisError::StreamDeleteFailed)
     }
@@ -586,7 +682,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn stream_trim_entries<C>(
         &self,
-        stream: &str,
+        stream: &RedisKey,
         xcap: C,
     ) -> CustomResult<usize, errors::RedisError>
     where
@@ -594,7 +690,7 @@ impl super::RedisConnectionPool {
         C::Error: Into<fred::error::RedisError> + Send + Sync,
     {
         self.pool
-            .xtrim(self.add_prefix(stream), xcap)
+            .xtrim(stream.tenant_aware_key(self), xcap)
             .await
             .change_context(errors::RedisError::StreamTrimFailed)
     }
@@ -602,7 +698,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn stream_acknowledge_entries<Ids>(
         &self,
-        stream: &str,
+        stream: &RedisKey,
         group: &str,
         ids: Ids,
     ) -> CustomResult<usize, errors::RedisError>
@@ -610,15 +706,18 @@ impl super::RedisConnectionPool {
         Ids: Into<MultipleIDs> + Debug + Send + Sync,
     {
         self.pool
-            .xack(self.add_prefix(stream), group, ids)
+            .xack(stream.tenant_aware_key(self), group, ids)
             .await
             .change_context(errors::RedisError::StreamAcknowledgeFailed)
     }
 
     #[instrument(level = "DEBUG", skip(self))]
-    pub async fn stream_get_length(&self, stream: &str) -> CustomResult<usize, errors::RedisError> {
+    pub async fn stream_get_length(
+        &self,
+        stream: &RedisKey,
+    ) -> CustomResult<usize, errors::RedisError> {
         self.pool
-            .xlen(self.add_prefix(stream))
+            .xlen(stream.tenant_aware_key(self))
             .await
             .change_context(errors::RedisError::GetLengthFailed)
     }
@@ -631,10 +730,10 @@ impl super::RedisConnectionPool {
         let res = multiple_keys
             .inner()
             .iter()
-            .filter_map(|key| key.as_str())
-            .map(|k| self.add_prefix(k))
-            .map(RedisKey::from)
+            .filter_map(|key| key.as_str().map(RedisKey::from))
+            .map(|k: RedisKey| k.tenant_aware_key(self))
             .collect::<Vec<_>>();
+
         MultipleKeys::from(res)
     }
 
@@ -710,7 +809,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn append_elements_to_list<V>(
         &self,
-        key: &str,
+        key: &RedisKey,
         elements: V,
     ) -> CustomResult<(), errors::RedisError>
     where
@@ -718,7 +817,7 @@ impl super::RedisConnectionPool {
         V::Error: Into<fred::error::RedisError> + Send,
     {
         self.pool
-            .rpush(self.add_prefix(key), elements)
+            .rpush(key.tenant_aware_key(self), elements)
             .await
             .change_context(errors::RedisError::AppendElementsToListFailed)
     }
@@ -726,20 +825,20 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn get_list_elements(
         &self,
-        key: &str,
+        key: &RedisKey,
         start: i64,
         stop: i64,
     ) -> CustomResult<Vec<String>, errors::RedisError> {
         self.pool
-            .lrange(self.add_prefix(key), start, stop)
+            .lrange(key.tenant_aware_key(self), start, stop)
             .await
             .change_context(errors::RedisError::GetListElementsFailed)
     }
 
     #[instrument(level = "DEBUG", skip(self))]
-    pub async fn get_list_length(&self, key: &str) -> CustomResult<usize, errors::RedisError> {
+    pub async fn get_list_length(&self, key: &RedisKey) -> CustomResult<usize, errors::RedisError> {
         self.pool
-            .llen(self.add_prefix(key))
+            .llen(key.tenant_aware_key(self))
             .await
             .change_context(errors::RedisError::GetListLengthFailed)
     }
@@ -747,11 +846,11 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn lpop_list_elements(
         &self,
-        key: &str,
+        key: &RedisKey,
         count: Option<usize>,
     ) -> CustomResult<Vec<String>, errors::RedisError> {
         self.pool
-            .lpop(self.add_prefix(key), count)
+            .lpop(key.tenant_aware_key(self), count)
             .await
             .change_context(errors::RedisError::PopListElementsFailed)
     }
@@ -761,7 +860,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn consumer_group_create(
         &self,
-        stream: &str,
+        stream: &RedisKey,
         group: &str,
         id: &RedisEntryId,
     ) -> CustomResult<(), errors::RedisError> {
@@ -774,7 +873,7 @@ impl super::RedisConnectionPool {
         }
 
         self.pool
-            .xgroup_create(self.add_prefix(stream), group, id, true)
+            .xgroup_create(stream.tenant_aware_key(self), group, id, true)
             .await
             .change_context(errors::RedisError::ConsumerGroupCreateFailed)
     }
@@ -782,11 +881,11 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn consumer_group_destroy(
         &self,
-        stream: &str,
+        stream: &RedisKey,
         group: &str,
     ) -> CustomResult<usize, errors::RedisError> {
         self.pool
-            .xgroup_destroy(self.add_prefix(stream), group)
+            .xgroup_destroy(stream.tenant_aware_key(self), group)
             .await
             .change_context(errors::RedisError::ConsumerGroupDestroyFailed)
     }
@@ -795,12 +894,12 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn consumer_group_delete_consumer(
         &self,
-        stream: &str,
+        stream: &RedisKey,
         group: &str,
         consumer: &str,
     ) -> CustomResult<usize, errors::RedisError> {
         self.pool
-            .xgroup_delconsumer(self.add_prefix(stream), group, consumer)
+            .xgroup_delconsumer(stream.tenant_aware_key(self), group, consumer)
             .await
             .change_context(errors::RedisError::ConsumerGroupRemoveConsumerFailed)
     }
@@ -808,12 +907,12 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn consumer_group_set_last_id(
         &self,
-        stream: &str,
+        stream: &RedisKey,
         group: &str,
         id: &RedisEntryId,
     ) -> CustomResult<String, errors::RedisError> {
         self.pool
-            .xgroup_setid(self.add_prefix(stream), group, id)
+            .xgroup_setid(stream.tenant_aware_key(self), group, id)
             .await
             .change_context(errors::RedisError::ConsumerGroupSetIdFailed)
     }
@@ -821,7 +920,7 @@ impl super::RedisConnectionPool {
     #[instrument(level = "DEBUG", skip(self))]
     pub async fn consumer_group_set_message_owner<Ids, R>(
         &self,
-        stream: &str,
+        stream: &RedisKey,
         group: &str,
         consumer: &str,
         min_idle_time: u64,
@@ -833,7 +932,7 @@ impl super::RedisConnectionPool {
     {
         self.pool
             .xclaim(
-                self.add_prefix(stream),
+                stream.tenant_aware_key(self),
                 group,
                 consumer,
                 min_idle_time,
@@ -888,11 +987,15 @@ mod tests {
 
                 // Act
                 let result1 = redis_conn
-                    .consumer_group_create("TEST1", "GTEST", &RedisEntryId::AutoGeneratedID)
+                    .consumer_group_create(&"TEST1".into(), "GTEST", &RedisEntryId::AutoGeneratedID)
                     .await;
 
                 let result2 = redis_conn
-                    .consumer_group_create("TEST3", "GTEST", &RedisEntryId::UndeliveredEntryID)
+                    .consumer_group_create(
+                        &"TEST3".into(),
+                        "GTEST",
+                        &RedisEntryId::UndeliveredEntryID,
+                    )
                     .await;
 
                 // Assert Setup
@@ -914,10 +1017,10 @@ mod tests {
                 let pool = RedisConnectionPool::new(&RedisSettings::default())
                     .await
                     .expect("failed to create redis connection pool");
-                let _ = pool.set_key("key", "value".to_string()).await;
+                let _ = pool.set_key(&"key".into(), "value".to_string()).await;
 
                 // Act
-                let result = pool.delete_key("key").await;
+                let result = pool.delete_key(&"key".into()).await;
 
                 // Assert setup
                 result.is_ok()
@@ -938,7 +1041,7 @@ mod tests {
                     .expect("failed to create redis connection pool");
 
                 // Act
-                let result = pool.delete_key("key not exists").await;
+                let result = pool.delete_key(&"key not exists".into()).await;
 
                 // Assert Setup
                 result.is_ok()
diff --git a/crates/redis_interface/src/types.rs b/crates/redis_interface/src/types.rs
index 92429f617d1..848996deb4a 100644
--- a/crates/redis_interface/src/types.rs
+++ b/crates/redis_interface/src/types.rs
@@ -4,7 +4,7 @@
 use common_utils::errors::CustomResult;
 use fred::types::RedisValue as FredRedisValue;
 
-use crate::errors;
+use crate::{errors, RedisConnectionPool};
 
 pub struct RedisValue {
     inner: FredRedisValue,
@@ -293,3 +293,24 @@ impl fred::types::FromRedis for SaddReply {
         }
     }
 }
+
+#[derive(Debug)]
+pub struct RedisKey(String);
+
+impl RedisKey {
+    pub fn tenant_aware_key(&self, pool: &RedisConnectionPool) -> String {
+        pool.add_prefix(&self.0)
+    }
+
+    pub fn tenant_unaware_key(&self, _pool: &RedisConnectionPool) -> String {
+        self.0.clone()
+    }
+}
+
+impl<T: AsRef<str>> From<T> for RedisKey {
+    fn from(value: T) -> Self {
+        let value = value.as_ref();
+
+        Self(value.to_string())
+    }
+}
diff --git a/crates/router/src/core/api_locking.rs b/crates/router/src/core/api_locking.rs
index 9b45fecee4e..7043a090b2b 100644
--- a/crates/router/src/core/api_locking.rs
+++ b/crates/router/src/core/api_locking.rs
@@ -79,7 +79,7 @@ impl LockAction {
                 for _retry in 0..lock_retries {
                     let redis_lock_result = redis_conn
                         .set_key_if_not_exists_with_expiry(
-                            redis_locking_key.as_str(),
+                            &redis_locking_key.as_str().into(),
                             state.get_request_id(),
                             Some(i64::from(redis_lock_expiry_seconds)),
                         )
@@ -134,12 +134,15 @@ impl LockAction {
                 let redis_locking_key = input.get_redis_locking_key(merchant_id);
 
                 match redis_conn
-                    .get_key::<Option<String>>(&redis_locking_key)
+                    .get_key::<Option<String>>(&redis_locking_key.as_str().into())
                     .await
                 {
                     Ok(val) => {
                         if val == state.get_request_id() {
-                            match redis_conn.delete_key(redis_locking_key.as_str()).await {
+                            match redis_conn
+                                .delete_key(&redis_locking_key.as_str().into())
+                                .await
+                            {
                                 Ok(redis::types::DelReply::KeyDeleted) => {
                                     logger::info!("Lock freed for locking input {:?}", input);
                                     tracing::Span::current()
diff --git a/crates/router/src/core/health_check.rs b/crates/router/src/core/health_check.rs
index 31e8cc75f5b..2e5c2956013 100644
--- a/crates/router/src/core/health_check.rs
+++ b/crates/router/src/core/health_check.rs
@@ -52,21 +52,21 @@ impl HealthCheckInterface for app::SessionState {
             .change_context(errors::HealthCheckRedisError::RedisConnectionError)?;
 
         redis_conn
-            .serialize_and_set_key_with_expiry("test_key", "test_value", 30)
+            .serialize_and_set_key_with_expiry(&"test_key".into(), "test_value", 30)
             .await
             .change_context(errors::HealthCheckRedisError::SetFailed)?;
 
         logger::debug!("Redis set_key was successful");
 
         redis_conn
-            .get_key::<()>("test_key")
+            .get_key::<()>(&"test_key".into())
             .await
             .change_context(errors::HealthCheckRedisError::GetFailed)?;
 
         logger::debug!("Redis get_key was successful");
 
         redis_conn
-            .delete_key("test_key")
+            .delete_key(&"test_key".into())
             .await
             .change_context(errors::HealthCheckRedisError::DeleteFailed)?;
 
diff --git a/crates/router/src/core/payment_methods/cards.rs b/crates/router/src/core/payment_methods/cards.rs
index 5bd2c07073e..7be645a81c7 100644
--- a/crates/router/src/core/payment_methods/cards.rs
+++ b/crates/router/src/core/payment_methods/cards.rs
@@ -3716,7 +3716,7 @@ pub async fn list_payment_methods(
         let redis_expiry = state.conf.payment_method_auth.get_inner().redis_expiry;
 
         if let Some(rc) = redis_conn {
-            rc.serialize_and_set_key_with_expiry(pm_auth_key.as_str(), val, redis_expiry)
+            rc.serialize_and_set_key_with_expiry(&pm_auth_key.as_str().into(), val, redis_expiry)
                 .await
                 .attach_printable("Failed to store pm auth data in redis")
                 .unwrap_or_else(|error| {
@@ -5030,7 +5030,7 @@ pub async fn list_customer_payment_method(
                 );
 
                 redis_conn
-                    .set_key_with_expiry(&key, pm_metadata.1, intent_fulfillment_time)
+                    .set_key_with_expiry(&key.into(), pm_metadata.1, intent_fulfillment_time)
                     .await
                     .change_context(errors::StorageError::KVError)
                     .change_context(errors::ApiErrorResponse::InternalServerError)
diff --git a/crates/router/src/core/payment_methods/vault.rs b/crates/router/src/core/payment_methods/vault.rs
index 65ef16cb449..300f1e3054f 100644
--- a/crates/router/src/core/payment_methods/vault.rs
+++ b/crates/router/src/core/payment_methods/vault.rs
@@ -1043,7 +1043,7 @@ pub async fn create_tokenize(
 
         redis_conn
             .set_key_if_not_exists_with_expiry(
-                redis_key.as_str(),
+                &redis_key.as_str().into(),
                 bytes::Bytes::from(encrypted_payload),
                 Some(i64::from(consts::LOCKER_REDIS_EXPIRY_SECONDS)),
             )
@@ -1089,7 +1089,9 @@ pub async fn get_tokenized_data(
             .change_context(errors::ApiErrorResponse::InternalServerError)
             .attach_printable("Failed to get redis connection")?;
 
-        let response = redis_conn.get_key::<bytes::Bytes>(redis_key.as_str()).await;
+        let response = redis_conn
+            .get_key::<bytes::Bytes>(&redis_key.as_str().into())
+            .await;
 
         match response {
             Ok(resp) => {
@@ -1150,7 +1152,7 @@ pub async fn delete_tokenized_data(
             .change_context(errors::ApiErrorResponse::InternalServerError)
             .attach_printable("Failed to get redis connection")?;
 
-        let response = redis_conn.delete_key(redis_key.as_str()).await;
+        let response = redis_conn.delete_key(&redis_key.as_str().into()).await;
 
         match response {
             Ok(redis_interface::DelReply::KeyDeleted) => Ok(()),
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index ceeb384ed21..4766dcda6a1 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -2363,7 +2363,7 @@ impl PaymentRedirectFlow for PaymentAuthenticateCompleteAuthorize {
                 .attach_printable("Failed to get redis connection")?;
             redis_conn
                 .set_key_with_expiry(
-                    &poll_id,
+                    &poll_id.into(),
                     api_models::poll::PollStatus::Pending.to_string(),
                     crate::consts::POLL_ID_TTL,
                 )
@@ -4118,7 +4118,7 @@ async fn decide_payment_method_tokenize_action(
                 );
 
                 let connector_token_option = redis_conn
-                    .get_key::<Option<String>>(&key)
+                    .get_key::<Option<String>>(&key.into())
                     .await
                     .change_context(errors::ApiErrorResponse::InternalServerError)
                     .attach_printable("Failed to fetch the token from redis")?;
@@ -6705,7 +6705,7 @@ pub async fn get_extended_card_info(
 
     let key = helpers::get_redis_key_for_extended_card_info(&merchant_id, &payment_id);
     let payload = redis_conn
-        .get_key::<String>(&key)
+        .get_key::<String>(&key.into())
         .await
         .change_context(errors::ApiErrorResponse::ExtendedCardInfoNotFound)?;
 
diff --git a/crates/router/src/core/payments/helpers.rs b/crates/router/src/core/payments/helpers.rs
index 93769c71e6c..797aadbbc05 100644
--- a/crates/router/src/core/payments/helpers.rs
+++ b/crates/router/src/core/payments/helpers.rs
@@ -2289,7 +2289,7 @@ pub async fn retrieve_payment_token_data(
     );
 
     let token_data_string = redis_conn
-        .get_key::<Option<String>>(&key)
+        .get_key::<Option<String>>(&key.into())
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Failed to fetch the token from redis")?
@@ -3781,7 +3781,7 @@ pub async fn insert_merchant_connector_creds_to_config(
 
         redis
             .serialize_and_set_key_with_expiry(
-                key.as_str(),
+                &key.as_str().into(),
                 &encoded_data.peek(),
                 consts::CONNECTOR_CREDS_TOKEN_TTL,
             )
@@ -3897,7 +3897,7 @@ pub async fn get_merchant_connector_account(
                     .attach_printable("Failed to get redis connection")
                     .async_and_then(|redis| async move {
                         redis
-                            .get_and_deserialize_key(key.clone().as_str(), "String")
+                            .get_and_deserialize_key(&key.as_str().into(), "String")
                             .await
                             .change_context(
                                 errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
@@ -5843,7 +5843,7 @@ pub async fn get_payment_method_details_from_payment_token(
                 .get_required_value("payment_method")?,
         );
         let token_data_string = redis_conn
-            .get_key::<Option<String>>(&key)
+            .get_key::<Option<String>>(&key.into())
             .await
             .change_context(errors::ApiErrorResponse::InternalServerError)
             .attach_printable("Failed to fetch the token from redis")?
diff --git a/crates/router/src/core/payments/operations/payment_confirm.rs b/crates/router/src/core/payments/operations/payment_confirm.rs
index a87fab9b17a..da68ee27fd9 100644
--- a/crates/router/src/core/payments/operations/payment_confirm.rs
+++ b/crates/router/src/core/payments/operations/payment_confirm.rs
@@ -1244,7 +1244,7 @@ impl<F: Clone + Send + Sync> Domain<F, api::PaymentsRequest, PaymentData<F>> for
 
             redis_conn
                 .set_key_with_expiry(
-                    &key,
+                    &key.into(),
                     encrypted_payload.clone(),
                     (*merchant_config.ttl_in_secs).into(),
                 )
diff --git a/crates/router/src/core/payments/types.rs b/crates/router/src/core/payments/types.rs
index 80813ac29b5..13f4de956ec 100644
--- a/crates/router/src/core/payments/types.rs
+++ b/crates/router/src/core/payments/types.rs
@@ -325,7 +325,11 @@ impl SurchargeMetadata {
                 .get_order_fulfillment_time()
                 .unwrap_or(router_consts::DEFAULT_FULFILLMENT_TIME);
             redis_conn
-                .set_hash_fields(&redis_key, value_list, Some(intent_fulfillment_time))
+                .set_hash_fields(
+                    &redis_key.as_str().into(),
+                    value_list,
+                    Some(intent_fulfillment_time),
+                )
                 .await
                 .change_context(errors::ApiErrorResponse::InternalServerError)
                 .attach_printable("Failed to write to redis")?;
@@ -347,7 +351,11 @@ impl SurchargeMetadata {
         let redis_key = Self::get_surcharge_metadata_redis_key(payment_attempt_id);
         let value_key = Self::get_surcharge_details_redis_hashset_key(&surcharge_key);
         let result = redis_conn
-            .get_hash_field_and_deserialize(&redis_key, &value_key, "SurchargeDetails")
+            .get_hash_field_and_deserialize(
+                &redis_key.as_str().into(),
+                &value_key,
+                "SurchargeDetails",
+            )
             .await;
         logger::debug!(
             "Surcharge result fetched from redis with key = {} and {}",
diff --git a/crates/router/src/core/payouts/helpers.rs b/crates/router/src/core/payouts/helpers.rs
index 060fc64f891..5ebdea0de67 100644
--- a/crates/router/src/core/payouts/helpers.rs
+++ b/crates/router/src/core/payouts/helpers.rs
@@ -79,7 +79,7 @@ pub async fn make_payout_method_data(
                 .attach_printable("Failed to get redis connection")?;
 
             let hyperswitch_token = redis_conn
-                .get_key::<Option<String>>(&key)
+                .get_key::<Option<String>>(&key.into())
                 .await
                 .change_context(errors::ApiErrorResponse::InternalServerError)
                 .attach_printable("Failed to fetch the token from redis")?
diff --git a/crates/router/src/core/pm_auth.rs b/crates/router/src/core/pm_auth.rs
index afba73e839c..0577b532cf0 100644
--- a/crates/router/src/core/pm_auth.rs
+++ b/crates/router/src/core/pm_auth.rs
@@ -63,7 +63,7 @@ pub async fn create_link_token(
     let pm_auth_key = payload.payment_id.get_pm_auth_key();
 
     redis_conn
-        .exists::<Vec<u8>>(&pm_auth_key)
+        .exists::<Vec<u8>>(&pm_auth_key.as_str().into())
         .await
         .change_context(ApiErrorResponse::InvalidRequestData {
             message: "Incorrect payment_id provided in request".to_string(),
@@ -77,7 +77,7 @@ pub async fn create_link_token(
 
     let pm_auth_configs = redis_conn
         .get_and_deserialize_key::<Vec<api_models::pm_auth::PaymentMethodAuthConnectorChoice>>(
-            pm_auth_key.as_str(),
+            &pm_auth_key.as_str().into(),
             "Vec<PaymentMethodAuthConnectorChoice>",
         )
         .await
@@ -772,7 +772,7 @@ async fn get_selected_config_from_redis(
     let pm_auth_key = payload.payment_id.get_pm_auth_key();
 
     redis_conn
-        .exists::<Vec<u8>>(&pm_auth_key)
+        .exists::<Vec<u8>>(&pm_auth_key.as_str().into())
         .await
         .change_context(ApiErrorResponse::InvalidRequestData {
             message: "Incorrect payment_id provided in request".to_string(),
@@ -786,7 +786,7 @@ async fn get_selected_config_from_redis(
 
     let pm_auth_configs = redis_conn
         .get_and_deserialize_key::<Vec<api_models::pm_auth::PaymentMethodAuthConnectorChoice>>(
-            pm_auth_key.as_str(),
+            &pm_auth_key.as_str().into(),
             "Vec<PaymentMethodAuthConnectorChoice>",
         )
         .await
diff --git a/crates/router/src/core/poll.rs b/crates/router/src/core/poll.rs
index f0a464be287..dd468a26a70 100644
--- a/crates/router/src/core/poll.rs
+++ b/crates/router/src/core/poll.rs
@@ -23,7 +23,7 @@ pub async fn retrieve_poll_status(
     // prepend 'poll_{merchant_id}_' to restrict access to only fetching Poll IDs, as this is a freely passed string in the request
     let poll_id = super::utils::get_poll_id(merchant_account.get_id(), request_poll_id.clone());
     let redis_value = redis_conn
-        .get_key::<Option<String>>(poll_id.as_str())
+        .get_key::<Option<String>>(&poll_id.as_str().into())
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable_lazy(|| {
diff --git a/crates/router/src/core/webhooks/incoming.rs b/crates/router/src/core/webhooks/incoming.rs
index 4c1a5aeb780..71130f247ad 100644
--- a/crates/router/src/core/webhooks/incoming.rs
+++ b/crates/router/src/core/webhooks/incoming.rs
@@ -1357,7 +1357,7 @@ async fn external_authentication_incoming_webhook_flow(
                             .attach_printable("Failed to get redis connection")?;
                         redis_conn
                             .set_key_without_modifying_ttl(
-                                &poll_id,
+                                &poll_id.into(),
                                 api_models::poll::PollStatus::Completed.to_string(),
                             )
                             .await
diff --git a/crates/router/src/db/ephemeral_key.rs b/crates/router/src/db/ephemeral_key.rs
index a77995e4e58..b3e33cd777f 100644
--- a/crates/router/src/db/ephemeral_key.rs
+++ b/crates/router/src/db/ephemeral_key.rs
@@ -103,7 +103,10 @@ mod storage {
                 .get_redis_conn()
                 .map_err(Into::<errors::StorageError>::into)?
                 .serialize_and_set_multiple_hash_field_if_not_exist(
-                    &[(&secret_key, &created_ek), (&id_key, &created_ek)],
+                    &[
+                        (&secret_key.as_str().into(), &created_ek),
+                        (&id_key.as_str().into(), &created_ek),
+                    ],
                     "ephkey",
                     None,
                 )
@@ -120,12 +123,12 @@ mod storage {
                     let expire_at = expires.assume_utc().unix_timestamp();
                     self.get_redis_conn()
                         .map_err(Into::<errors::StorageError>::into)?
-                        .set_expire_at(&secret_key, expire_at)
+                        .set_expire_at(&secret_key.into(), expire_at)
                         .await
                         .change_context(errors::StorageError::KVError)?;
                     self.get_redis_conn()
                         .map_err(Into::<errors::StorageError>::into)?
-                        .set_expire_at(&id_key, expire_at)
+                        .set_expire_at(&id_key.into(), expire_at)
                         .await
                         .change_context(errors::StorageError::KVError)?;
                     Ok(created_ek)
@@ -161,8 +164,8 @@ mod storage {
                 .map_err(Into::<errors::StorageError>::into)?
                 .serialize_and_set_multiple_hash_field_if_not_exist(
                     &[
-                        (&secret_key, &created_ephemeral_key),
-                        (&id_key, &created_ephemeral_key),
+                        (&secret_key.as_str().into(), &created_ephemeral_key),
+                        (&id_key.as_str().into(), &created_ephemeral_key),
                     ],
                     "ephkey",
                     None,
@@ -180,12 +183,12 @@ mod storage {
                     let expire_at = expires.assume_utc().unix_timestamp();
                     self.get_redis_conn()
                         .map_err(Into::<errors::StorageError>::into)?
-                        .set_expire_at(&secret_key, expire_at)
+                        .set_expire_at(&secret_key.into(), expire_at)
                         .await
                         .change_context(errors::StorageError::KVError)?;
                     self.get_redis_conn()
                         .map_err(Into::<errors::StorageError>::into)?
-                        .set_expire_at(&id_key, expire_at)
+                        .set_expire_at(&id_key.into(), expire_at)
                         .await
                         .change_context(errors::StorageError::KVError)?;
                     Ok(created_ephemeral_key)
@@ -203,7 +206,7 @@ mod storage {
             let key = format!("epkey_{key}");
             self.get_redis_conn()
                 .map_err(Into::<errors::StorageError>::into)?
-                .get_hash_field_and_deserialize(&key, "ephkey", "EphemeralKey")
+                .get_hash_field_and_deserialize(&key.into(), "ephkey", "EphemeralKey")
                 .await
                 .change_context(errors::StorageError::KVError)
         }
@@ -217,7 +220,7 @@ mod storage {
             let key = format!("epkey_{key}");
             self.get_redis_conn()
                 .map_err(Into::<errors::StorageError>::into)?
-                .get_hash_field_and_deserialize(&key, "ephkey", "EphemeralKeyType")
+                .get_hash_field_and_deserialize(&key.into(), "ephkey", "EphemeralKeyType")
                 .await
                 .change_context(errors::StorageError::KVError)
         }
@@ -231,13 +234,13 @@ mod storage {
 
             self.get_redis_conn()
                 .map_err(Into::<errors::StorageError>::into)?
-                .delete_key(&format!("epkey_{}", &ek.id))
+                .delete_key(&format!("epkey_{}", &ek.id).into())
                 .await
                 .change_context(errors::StorageError::KVError)?;
 
             self.get_redis_conn()
                 .map_err(Into::<errors::StorageError>::into)?
-                .delete_key(&format!("epkey_{}", &ek.secret))
+                .delete_key(&format!("epkey_{}", &ek.secret).into())
                 .await
                 .change_context(errors::StorageError::KVError)?;
             Ok(ek)
@@ -254,7 +257,7 @@ mod storage {
 
             self.get_redis_conn()
                 .map_err(Into::<errors::StorageError>::into)?
-                .delete_key(&redis_id_key)
+                .delete_key(&redis_id_key.as_str().into())
                 .await
                 .map_err(|err| match err.current_context() {
                     RedisError::NotFound => {
@@ -265,7 +268,7 @@ mod storage {
 
             self.get_redis_conn()
                 .map_err(Into::<errors::StorageError>::into)?
-                .delete_key(&secret_key)
+                .delete_key(&secret_key.as_str().into())
                 .await
                 .map_err(|err| match err.current_context() {
                     RedisError::NotFound => {
diff --git a/crates/router/src/db/merchant_connector_account.rs b/crates/router/src/db/merchant_connector_account.rs
index ba654b4fce3..34b23a833ac 100644
--- a/crates/router/src/db/merchant_connector_account.rs
+++ b/crates/router/src/db/merchant_connector_account.rs
@@ -59,7 +59,7 @@ impl ConnectorAccessToken for Store {
         let maybe_token = self
             .get_redis_conn()
             .map_err(Into::<errors::StorageError>::into)?
-            .get_key::<Option<Vec<u8>>>(&key)
+            .get_key::<Option<Vec<u8>>>(&key.into())
             .await
             .change_context(errors::StorageError::KVError)
             .attach_printable("DB error when getting access token")?;
@@ -88,7 +88,7 @@ impl ConnectorAccessToken for Store {
             .change_context(errors::StorageError::SerializationFailed)?;
         self.get_redis_conn()
             .map_err(Into::<errors::StorageError>::into)?
-            .set_key_with_expiry(&key, serialized_access_token, access_token.expires)
+            .set_key_with_expiry(&key.into(), serialized_access_token, access_token.expires)
             .await
             .change_context(errors::StorageError::KVError)
     }
diff --git a/crates/router/src/routes/dummy_connector/core.rs b/crates/router/src/routes/dummy_connector/core.rs
index aadd7dcaa72..cafb6b58bca 100644
--- a/crates/router/src/routes/dummy_connector/core.rs
+++ b/crates/router/src/routes/dummy_connector/core.rs
@@ -109,7 +109,7 @@ pub async fn payment_complete(
         .change_context(errors::DummyConnectorErrors::InternalServerError)
         .attach_printable("Failed to get redis connection")?;
 
-    let _ = redis_conn.delete_key(req.attempt_id.as_str()).await;
+    let _ = redis_conn.delete_key(&req.attempt_id.as_str().into()).await;
 
     if let Ok(payment_data) = payment_data {
         let updated_payment_data = types::DummyConnectorPaymentData {
@@ -220,7 +220,7 @@ pub async fn refund_data(
         .attach_printable("Failed to get redis connection")?;
     let refund_data = redis_conn
         .get_and_deserialize_key::<types::DummyConnectorRefundResponse>(
-            refund_id.as_str(),
+            &refund_id.as_str().into(),
             "DummyConnectorRefundResponse",
         )
         .await
diff --git a/crates/router/src/routes/dummy_connector/utils.rs b/crates/router/src/routes/dummy_connector/utils.rs
index 6211c3abf80..edbe6d9dddc 100644
--- a/crates/router/src/routes/dummy_connector/utils.rs
+++ b/crates/router/src/routes/dummy_connector/utils.rs
@@ -38,7 +38,7 @@ pub async fn store_data_in_redis(
         .attach_printable("Failed to get redis connection")?;
 
     redis_conn
-        .serialize_and_set_key_with_expiry(&key, data, ttl)
+        .serialize_and_set_key_with_expiry(&key.into(), data, ttl)
         .await
         .change_context(errors::DummyConnectorErrors::PaymentStoringError)
         .attach_printable("Failed to add data in redis")?;
@@ -57,7 +57,7 @@ pub async fn get_payment_data_from_payment_id(
 
     redis_conn
         .get_and_deserialize_key::<types::DummyConnectorPaymentData>(
-            payment_id.as_str(),
+            &payment_id.as_str().into(),
             "types DummyConnectorPaymentData",
         )
         .await
@@ -75,12 +75,12 @@ pub async fn get_payment_data_by_attempt_id(
         .attach_printable("Failed to get redis connection")?;
 
     redis_conn
-        .get_and_deserialize_key::<String>(attempt_id.as_str(), "String")
+        .get_and_deserialize_key::<String>(&attempt_id.as_str().into(), "String")
         .await
         .async_and_then(|payment_id| async move {
             redis_conn
                 .get_and_deserialize_key::<types::DummyConnectorPaymentData>(
-                    payment_id.as_str(),
+                    &payment_id.as_str().into(),
                     "DummyConnectorPaymentData",
                 )
                 .await
diff --git a/crates/router/src/routes/payment_methods.rs b/crates/router/src/routes/payment_methods.rs
index c8e6303562c..5f9fe729eeb 100644
--- a/crates/router/src/routes/payment_methods.rs
+++ b/crates/router/src/routes/payment_methods.rs
@@ -980,7 +980,11 @@ impl ParentPaymentMethodToken {
             .change_context(errors::ApiErrorResponse::InternalServerError)
             .attach_printable("Failed to get redis connection")?;
         redis_conn
-            .serialize_and_set_key_with_expiry(&self.key_for_token, token, fulfillment_time)
+            .serialize_and_set_key_with_expiry(
+                &self.key_for_token.as_str().into(),
+                token,
+                fulfillment_time,
+            )
             .await
             .change_context(errors::StorageError::KVError)
             .change_context(errors::ApiErrorResponse::InternalServerError)
@@ -1004,7 +1008,10 @@ impl ParentPaymentMethodToken {
             .get_redis_conn()
             .change_context(errors::ApiErrorResponse::InternalServerError)
             .attach_printable("Failed to get redis connection")?;
-        match redis_conn.delete_key(&self.key_for_token).await {
+        match redis_conn
+            .delete_key(&self.key_for_token.as_str().into())
+            .await
+        {
             Ok(_) => Ok(()),
             Err(err) => {
                 {
diff --git a/crates/router/src/services/authentication/blacklist.rs b/crates/router/src/services/authentication/blacklist.rs
index da5f87a450e..6be6cc467bf 100644
--- a/crates/router/src/services/authentication/blacklist.rs
+++ b/crates/router/src/services/authentication/blacklist.rs
@@ -30,7 +30,7 @@ pub async fn insert_user_in_blacklist(state: &SessionState, user_id: &str) -> Us
     let redis_conn = get_redis_connection(state).change_context(UserErrors::InternalServerError)?;
     redis_conn
         .set_key_with_expiry(
-            user_blacklist_key.as_str(),
+            &user_blacklist_key.as_str().into(),
             date_time::now_unix_timestamp(),
             expiry,
         )
@@ -46,7 +46,7 @@ pub async fn insert_role_in_blacklist(state: &SessionState, role_id: &str) -> Us
     let redis_conn = get_redis_connection(state).change_context(UserErrors::InternalServerError)?;
     redis_conn
         .set_key_with_expiry(
-            role_blacklist_key.as_str(),
+            &role_blacklist_key.as_str().into(),
             date_time::now_unix_timestamp(),
             expiry,
         )
@@ -61,7 +61,7 @@ pub async fn insert_role_in_blacklist(state: &SessionState, role_id: &str) -> Us
 async fn invalidate_role_cache(state: &SessionState, role_id: &str) -> RouterResult<()> {
     let redis_conn = get_redis_connection(state)?;
     redis_conn
-        .delete_key(authz::get_cache_key_from_role_id(role_id).as_str())
+        .delete_key(&authz::get_cache_key_from_role_id(role_id).as_str().into())
         .await
         .map(|_| ())
         .change_context(ApiErrorResponse::InternalServerError)
@@ -76,7 +76,7 @@ pub async fn check_user_in_blacklist<A: SessionStateInfo>(
     let token_issued_at = expiry_to_i64(token_expiry - JWT_TOKEN_TIME_IN_SECS)?;
     let redis_conn = get_redis_connection(state)?;
     redis_conn
-        .get_key::<Option<i64>>(token.as_str())
+        .get_key::<Option<i64>>(&token.as_str().into())
         .await
         .change_context(ApiErrorResponse::InternalServerError)
         .map(|timestamp| timestamp > Some(token_issued_at))
@@ -91,7 +91,7 @@ pub async fn check_role_in_blacklist<A: SessionStateInfo>(
     let token_issued_at = expiry_to_i64(token_expiry - JWT_TOKEN_TIME_IN_SECS)?;
     let redis_conn = get_redis_connection(state)?;
     redis_conn
-        .get_key::<Option<i64>>(token.as_str())
+        .get_key::<Option<i64>>(&token.as_str().into())
         .await
         .change_context(ApiErrorResponse::InternalServerError)
         .map(|timestamp| timestamp > Some(token_issued_at))
@@ -104,7 +104,7 @@ pub async fn insert_email_token_in_blacklist(state: &SessionState, token: &str)
     let expiry =
         expiry_to_i64(EMAIL_TOKEN_TIME_IN_SECS).change_context(UserErrors::InternalServerError)?;
     redis_conn
-        .set_key_with_expiry(blacklist_key.as_str(), true, expiry)
+        .set_key_with_expiry(&blacklist_key.as_str().into(), true, expiry)
         .await
         .change_context(UserErrors::InternalServerError)
 }
@@ -114,7 +114,7 @@ pub async fn check_email_token_in_blacklist(state: &SessionState, token: &str) -
     let redis_conn = get_redis_connection(state).change_context(UserErrors::InternalServerError)?;
     let blacklist_key = format!("{}{token}", EMAIL_TOKEN_BLACKLIST_PREFIX);
     let key_exists = redis_conn
-        .exists::<()>(blacklist_key.as_str())
+        .exists::<()>(&blacklist_key.as_str().into())
         .await
         .change_context(UserErrors::InternalServerError)?;
 
diff --git a/crates/router/src/services/authorization.rs b/crates/router/src/services/authorization.rs
index db3483f8164..0e66654b7b9 100644
--- a/crates/router/src/services/authorization.rs
+++ b/crates/router/src/services/authorization.rs
@@ -64,7 +64,7 @@ where
     let redis_conn = get_redis_connection(state)?;
 
     redis_conn
-        .get_and_deserialize_key(&get_cache_key_from_role_id(role_id), "RoleInfo")
+        .get_and_deserialize_key(&get_cache_key_from_role_id(role_id).into(), "RoleInfo")
         .await
         .change_context(ApiErrorResponse::InternalServerError)
 }
@@ -103,7 +103,11 @@ where
     let redis_conn = get_redis_connection(state)?;
 
     redis_conn
-        .serialize_and_set_key_with_expiry(&get_cache_key_from_role_id(role_id), role_info, expiry)
+        .serialize_and_set_key_with_expiry(
+            &get_cache_key_from_role_id(role_id).into(),
+            role_info,
+            expiry,
+        )
         .await
         .change_context(ApiErrorResponse::InternalServerError)
 }
diff --git a/crates/router/src/services/openidconnect.rs b/crates/router/src/services/openidconnect.rs
index 0c0f86431a1..44f95b42607 100644
--- a/crates/router/src/services/openidconnect.rs
+++ b/crates/router/src/services/openidconnect.rs
@@ -35,7 +35,7 @@ pub async fn get_authorization_url(
     // Save csrf & nonce as key value respectively
     let key = get_oidc_redis_key(csrf_token.secret());
     get_redis_connection(&state)?
-        .set_key_with_expiry(&key, nonce.secret(), consts::user::REDIS_SSO_TTL)
+        .set_key_with_expiry(&key.into(), nonce.secret(), consts::user::REDIS_SSO_TTL)
         .await
         .change_context(UserErrors::InternalServerError)
         .attach_printable("Failed to save csrf-nonce in redis")?;
@@ -142,7 +142,7 @@ async fn get_nonce_from_redis(
     let redirect_state = redirect_state.clone().expose();
     let key = get_oidc_redis_key(&redirect_state);
     redis_connection
-        .get_key::<Option<String>>(&key)
+        .get_key::<Option<String>>(&key.into())
         .await
         .change_context(UserErrors::InternalServerError)
         .attach_printable("Error Fetching CSRF from redis")?
diff --git a/crates/router/src/utils/currency.rs b/crates/router/src/utils/currency.rs
index 9ab2780da73..08077442e73 100644
--- a/crates/router/src/utils/currency.rs
+++ b/crates/router/src/utils/currency.rs
@@ -465,7 +465,7 @@ async fn release_redis_lock(
         .store
         .get_redis_conn()
         .change_context(ForexCacheError::RedisConnectionError)?
-        .delete_key(REDIX_FOREX_CACHE_KEY)
+        .delete_key(&REDIX_FOREX_CACHE_KEY.into())
         .await
         .change_context(ForexCacheError::RedisLockReleaseFailed)
         .attach_printable("Unable to release redis lock")
@@ -478,7 +478,7 @@ async fn acquire_redis_lock(state: &SessionState) -> CustomResult<bool, ForexCac
         .get_redis_conn()
         .change_context(ForexCacheError::RedisConnectionError)?
         .set_key_if_not_exists_with_expiry(
-            REDIX_FOREX_CACHE_KEY,
+            &REDIX_FOREX_CACHE_KEY.into(),
             "",
             Some(
                 i64::try_from(
@@ -502,7 +502,7 @@ async fn save_forex_to_redis(
         .store
         .get_redis_conn()
         .change_context(ForexCacheError::RedisConnectionError)?
-        .serialize_and_set_key(REDIX_FOREX_CACHE_DATA, forex_exchange_cache_entry)
+        .serialize_and_set_key(&REDIX_FOREX_CACHE_DATA.into(), forex_exchange_cache_entry)
         .await
         .change_context(ForexCacheError::RedisWriteError)
         .attach_printable("Unable to save forex data to redis")
@@ -515,7 +515,7 @@ async fn retrieve_forex_from_redis(
         .store
         .get_redis_conn()
         .change_context(ForexCacheError::RedisConnectionError)?
-        .get_and_deserialize_key(REDIX_FOREX_CACHE_DATA, "FxExchangeRatesCache")
+        .get_and_deserialize_key(&REDIX_FOREX_CACHE_DATA.into(), "FxExchangeRatesCache")
         .await
         .change_context(ForexCacheError::EntryNotFound)
         .attach_printable("Forex entry not found in redis")
diff --git a/crates/router/src/utils/user.rs b/crates/router/src/utils/user.rs
index ef06531b421..721b0052bda 100644
--- a/crates/router/src/utils/user.rs
+++ b/crates/router/src/utils/user.rs
@@ -246,7 +246,7 @@ pub async fn set_sso_id_in_redis(
     let connection = get_redis_connection(state)?;
     let key = get_oidc_key(&oidc_state.expose());
     connection
-        .set_key_with_expiry(&key, sso_id, REDIS_SSO_TTL)
+        .set_key_with_expiry(&key.into(), sso_id, REDIS_SSO_TTL)
         .await
         .change_context(UserErrors::InternalServerError)
         .attach_printable("Failed to set sso id in redis")
@@ -259,7 +259,7 @@ pub async fn get_sso_id_from_redis(
     let connection = get_redis_connection(state)?;
     let key = get_oidc_key(&oidc_state.expose());
     connection
-        .get_key::<Option<String>>(&key)
+        .get_key::<Option<String>>(&key.into())
         .await
         .change_context(UserErrors::InternalServerError)
         .attach_printable("Failed to get sso id from redis")?
diff --git a/crates/router/src/utils/user/two_factor_auth.rs b/crates/router/src/utils/user/two_factor_auth.rs
index 73d692a00e0..b4ced70d735 100644
--- a/crates/router/src/utils/user/two_factor_auth.rs
+++ b/crates/router/src/utils/user/two_factor_auth.rs
@@ -36,7 +36,7 @@ pub async fn check_totp_in_redis(state: &SessionState, user_id: &str) -> UserRes
     let redis_conn = super::get_redis_connection(state)?;
     let key = format!("{}{}", consts::user::REDIS_TOTP_PREFIX, user_id);
     redis_conn
-        .exists::<()>(&key)
+        .exists::<()>(&key.into())
         .await
         .change_context(UserErrors::InternalServerError)
 }
@@ -45,7 +45,7 @@ pub async fn check_recovery_code_in_redis(state: &SessionState, user_id: &str) -
     let redis_conn = super::get_redis_connection(state)?;
     let key = format!("{}{}", consts::user::REDIS_RECOVERY_CODE_PREFIX, user_id);
     redis_conn
-        .exists::<()>(&key)
+        .exists::<()>(&key.into())
         .await
         .change_context(UserErrors::InternalServerError)
 }
@@ -55,7 +55,7 @@ pub async fn insert_totp_in_redis(state: &SessionState, user_id: &str) -> UserRe
     let key = format!("{}{}", consts::user::REDIS_TOTP_PREFIX, user_id);
     redis_conn
         .set_key_with_expiry(
-            key.as_str(),
+            &key.as_str().into(),
             common_utils::date_time::now_unix_timestamp(),
             state.conf.user.two_factor_auth_expiry_in_secs,
         )
@@ -71,7 +71,7 @@ pub async fn insert_totp_secret_in_redis(
     let redis_conn = super::get_redis_connection(state)?;
     redis_conn
         .set_key_with_expiry(
-            &get_totp_secret_key(user_id),
+            &get_totp_secret_key(user_id).into(),
             secret.peek(),
             consts::user::REDIS_TOTP_SECRET_TTL_IN_SECS,
         )
@@ -85,7 +85,7 @@ pub async fn get_totp_secret_from_redis(
 ) -> UserResult<Option<masking::Secret<String>>> {
     let redis_conn = super::get_redis_connection(state)?;
     redis_conn
-        .get_key::<Option<String>>(&get_totp_secret_key(user_id))
+        .get_key::<Option<String>>(&get_totp_secret_key(user_id).into())
         .await
         .change_context(UserErrors::InternalServerError)
         .map(|secret| secret.map(Into::into))
@@ -94,7 +94,7 @@ pub async fn get_totp_secret_from_redis(
 pub async fn delete_totp_secret_from_redis(state: &SessionState, user_id: &str) -> UserResult<()> {
     let redis_conn = super::get_redis_connection(state)?;
     redis_conn
-        .delete_key(&get_totp_secret_key(user_id))
+        .delete_key(&get_totp_secret_key(user_id).into())
         .await
         .change_context(UserErrors::InternalServerError)
         .map(|_| ())
@@ -109,7 +109,7 @@ pub async fn insert_recovery_code_in_redis(state: &SessionState, user_id: &str)
     let key = format!("{}{}", consts::user::REDIS_RECOVERY_CODE_PREFIX, user_id);
     redis_conn
         .set_key_with_expiry(
-            key.as_str(),
+            &key.as_str().into(),
             common_utils::date_time::now_unix_timestamp(),
             state.conf.user.two_factor_auth_expiry_in_secs,
         )
@@ -121,7 +121,7 @@ pub async fn delete_totp_from_redis(state: &SessionState, user_id: &str) -> User
     let redis_conn = super::get_redis_connection(state)?;
     let key = format!("{}{}", consts::user::REDIS_TOTP_PREFIX, user_id);
     redis_conn
-        .delete_key(&key)
+        .delete_key(&key.into())
         .await
         .change_context(UserErrors::InternalServerError)
         .map(|_| ())
@@ -134,7 +134,7 @@ pub async fn delete_recovery_code_from_redis(
     let redis_conn = super::get_redis_connection(state)?;
     let key = format!("{}{}", consts::user::REDIS_RECOVERY_CODE_PREFIX, user_id);
     redis_conn
-        .delete_key(&key)
+        .delete_key(&key.into())
         .await
         .change_context(UserErrors::InternalServerError)
         .map(|_| ())
@@ -159,7 +159,7 @@ pub async fn insert_totp_attempts_in_redis(
     let redis_conn = super::get_redis_connection(state)?;
     redis_conn
         .set_key_with_expiry(
-            &get_totp_attempts_key(user_id),
+            &get_totp_attempts_key(user_id).into(),
             user_totp_attempts,
             consts::user::REDIS_TOTP_ATTEMPTS_TTL_IN_SECS,
         )
@@ -169,7 +169,7 @@ pub async fn insert_totp_attempts_in_redis(
 pub async fn get_totp_attempts_from_redis(state: &SessionState, user_id: &str) -> UserResult<u8> {
     let redis_conn = super::get_redis_connection(state)?;
     redis_conn
-        .get_key::<Option<u8>>(&get_totp_attempts_key(user_id))
+        .get_key::<Option<u8>>(&get_totp_attempts_key(user_id).into())
         .await
         .change_context(UserErrors::InternalServerError)
         .map(|v| v.unwrap_or(0))
@@ -183,7 +183,7 @@ pub async fn insert_recovery_code_attempts_in_redis(
     let redis_conn = super::get_redis_connection(state)?;
     redis_conn
         .set_key_with_expiry(
-            &get_recovery_code_attempts_key(user_id),
+            &get_recovery_code_attempts_key(user_id).into(),
             user_recovery_code_attempts,
             consts::user::REDIS_RECOVERY_CODE_ATTEMPTS_TTL_IN_SECS,
         )
@@ -197,7 +197,7 @@ pub async fn get_recovery_code_attempts_from_redis(
 ) -> UserResult<u8> {
     let redis_conn = super::get_redis_connection(state)?;
     redis_conn
-        .get_key::<Option<u8>>(&get_recovery_code_attempts_key(user_id))
+        .get_key::<Option<u8>>(&get_recovery_code_attempts_key(user_id).into())
         .await
         .change_context(UserErrors::InternalServerError)
         .map(|v| v.unwrap_or(0))
@@ -209,7 +209,7 @@ pub async fn delete_totp_attempts_from_redis(
 ) -> UserResult<()> {
     let redis_conn = super::get_redis_connection(state)?;
     redis_conn
-        .delete_key(&get_totp_attempts_key(user_id))
+        .delete_key(&get_totp_attempts_key(user_id).into())
         .await
         .change_context(UserErrors::InternalServerError)
         .map(|_| ())
@@ -221,7 +221,7 @@ pub async fn delete_recovery_code_attempts_from_redis(
 ) -> UserResult<()> {
     let redis_conn = super::get_redis_connection(state)?;
     redis_conn
-        .delete_key(&get_recovery_code_attempts_key(user_id))
+        .delete_key(&get_recovery_code_attempts_key(user_id).into())
         .await
         .change_context(UserErrors::InternalServerError)
         .map(|_| ())
diff --git a/crates/router/tests/cache.rs b/crates/router/tests/cache.rs
index a1f85534b6b..7e1fab4bc5a 100644
--- a/crates/router/tests/cache.rs
+++ b/crates/router/tests/cache.rs
@@ -30,7 +30,7 @@ async fn invalidate_existing_cache_success() {
         .store
         .get_redis_conn()
         .unwrap()
-        .set_key(&cache_key.clone(), cache_key_value.clone())
+        .set_key(&cache_key.clone().into(), cache_key_value.clone())
         .await;
 
     let api_key = ("api-key", "test_admin");
diff --git a/crates/scheduler/src/db/queue.rs b/crates/scheduler/src/db/queue.rs
index 748e1f489c2..80f2b8b4c92 100644
--- a/crates/scheduler/src/db/queue.rs
+++ b/crates/scheduler/src/db/queue.rs
@@ -70,7 +70,7 @@ impl QueueInterface for Store {
         id: &RedisEntryId,
     ) -> CustomResult<(), RedisError> {
         self.get_redis_conn()?
-            .consumer_group_create(stream, group, id)
+            .consumer_group_create(&stream.into(), group, id)
             .await
     }
 
@@ -83,16 +83,16 @@ impl QueueInterface for Store {
     ) -> CustomResult<bool, RedisError> {
         let conn = self.get_redis_conn()?.clone();
         let is_lock_acquired = conn
-            .set_key_if_not_exists_with_expiry(lock_key, lock_val, None)
+            .set_key_if_not_exists_with_expiry(&lock_key.into(), lock_val, None)
             .await;
         Ok(match is_lock_acquired {
-            Ok(SetnxReply::KeySet) => match conn.set_expiry(lock_key, ttl).await {
+            Ok(SetnxReply::KeySet) => match conn.set_expiry(&lock_key.into(), ttl).await {
                 Ok(()) => true,
 
                 #[allow(unused_must_use)]
                 Err(error) => {
                     logger::error!(?error);
-                    conn.delete_key(lock_key).await;
+                    conn.delete_key(&lock_key.into()).await;
                     false
                 }
             },
@@ -108,7 +108,7 @@ impl QueueInterface for Store {
     }
 
     async fn release_pt_lock(&self, tag: &str, lock_key: &str) -> CustomResult<bool, RedisError> {
-        let is_lock_released = self.get_redis_conn()?.delete_key(lock_key).await;
+        let is_lock_released = self.get_redis_conn()?.delete_key(&lock_key.into()).await;
         Ok(match is_lock_released {
             Ok(_del_reply) => true,
             Err(error) => {
@@ -125,12 +125,12 @@ impl QueueInterface for Store {
         fields: Vec<(&str, String)>,
     ) -> CustomResult<(), RedisError> {
         self.get_redis_conn()?
-            .stream_append_entry(stream, entry_id, fields)
+            .stream_append_entry(&stream.into(), entry_id, fields)
             .await
     }
 
     async fn get_key(&self, key: &str) -> CustomResult<Vec<u8>, RedisError> {
-        self.get_redis_conn()?.get_key::<Vec<u8>>(key).await
+        self.get_redis_conn()?.get_key::<Vec<u8>>(&key.into()).await
     }
 }
 
diff --git a/crates/scheduler/src/utils.rs b/crates/scheduler/src/utils.rs
index 89328479537..51938e6c14d 100644
--- a/crates/scheduler/src/utils.rs
+++ b/crates/scheduler/src/utils.rs
@@ -231,13 +231,13 @@ pub async fn get_batches(
     let batches = batches.into_iter().flatten().collect::<Vec<_>>();
     let entry_ids = entry_ids.into_iter().flatten().collect::<Vec<_>>();
 
-    conn.stream_acknowledge_entries(stream_name, group_name, entry_ids.clone())
+    conn.stream_acknowledge_entries(&stream_name.into(), group_name, entry_ids.clone())
         .await
         .map_err(|error| {
             logger::error!(?error, "Error acknowledging batch in stream");
             error.change_context(errors::ProcessTrackerError::BatchUpdateFailed)
         })?;
-    conn.stream_delete_entries(stream_name, entry_ids.clone())
+    conn.stream_delete_entries(&stream_name.into(), entry_ids.clone())
         .await
         .map_err(|error| {
             logger::error!(?error, "Error deleting batch from stream");
diff --git a/crates/storage_impl/src/lib.rs b/crates/storage_impl/src/lib.rs
index e0722ef52ea..8a329010447 100644
--- a/crates/storage_impl/src/lib.rs
+++ b/crates/storage_impl/src/lib.rs
@@ -256,7 +256,7 @@ impl<T: DatabaseStore> KVRouterStore<T> {
             .cache_store
             .redis_conn
             .stream_append_entry(
-                &stream_name,
+                &stream_name.into(),
                 &redis_interface::RedisEntryId::AutoGeneratedID,
                 redis_entry
                     .to_field_value_pairs(request_id, global_id)
@@ -309,7 +309,7 @@ pub trait UniqueConstraints {
         let constraints = self.unique_constraints();
         let sadd_result = redis_conn
             .sadd(
-                &format!("unique_constraint:{}", self.table_name()),
+                &format!("unique_constraint:{}", self.table_name()).into(),
                 constraints,
             )
             .await?;
diff --git a/crates/storage_impl/src/redis/cache.rs b/crates/storage_impl/src/redis/cache.rs
index 323d3d6df25..8302b5bf933 100644
--- a/crates/storage_impl/src/redis/cache.rs
+++ b/crates/storage_impl/src/redis/cache.rs
@@ -299,11 +299,13 @@ where
 {
     let type_name = std::any::type_name::<T>();
     let key = key.as_ref();
-    let redis_val = redis.get_and_deserialize_key::<T>(key, type_name).await;
+    let redis_val = redis
+        .get_and_deserialize_key::<T>(&key.into(), type_name)
+        .await;
     let get_data_set_redis = || async {
         let data = fun().await?;
         redis
-            .serialize_and_set_key(key, &data)
+            .serialize_and_set_key(&key.into(), &data)
             .await
             .change_context(StorageError::KVError)?;
         Ok::<_, Report<StorageError>>(data)
@@ -380,7 +382,7 @@ pub async fn redact_from_redis_and_publish<
     let redis_keys_to_be_deleted = keys
         .clone()
         .into_iter()
-        .map(|val| val.get_key_without_prefix().to_owned())
+        .map(|val| val.get_key_without_prefix().to_owned().into())
         .collect::<Vec<_>>();
 
     let del_replies = redis_conn
diff --git a/crates/storage_impl/src/redis/kv_store.rs b/crates/storage_impl/src/redis/kv_store.rs
index 17974a3b9bc..6e1340abca5 100644
--- a/crates/storage_impl/src/redis/kv_store.rs
+++ b/crates/storage_impl/src/redis/kv_store.rs
@@ -181,7 +181,7 @@ where
                 logger::debug!(kv_operation= %operation, value = ?value);
 
                 redis_conn
-                    .set_hash_fields(&key, value, Some(ttl.into()))
+                    .set_hash_fields(&key.into(), value, Some(ttl.into()))
                     .await?;
 
                 store
@@ -193,14 +193,14 @@ where
 
             KvOperation::HGet(field) => {
                 let result = redis_conn
-                    .get_hash_field_and_deserialize(&key, field, type_name)
+                    .get_hash_field_and_deserialize(&key.into(), field, type_name)
                     .await?;
                 Ok(KvResult::HGet(result))
             }
 
             KvOperation::Scan(pattern) => {
                 let result: Vec<T> = redis_conn
-                    .hscan_and_deserialize(&key, pattern, None)
+                    .hscan_and_deserialize(&key.into(), pattern, None)
                     .await
                     .and_then(|result| {
                         if result.is_empty() {
@@ -218,7 +218,7 @@ where
                 value.check_for_constraints(&redis_conn).await?;
 
                 let result = redis_conn
-                    .serialize_and_set_hash_field_if_not_exist(&key, field, value, Some(ttl))
+                    .serialize_and_set_hash_field_if_not_exist(&key.into(), field, value, Some(ttl))
                     .await?;
 
                 if matches!(result, redis_interface::HsetnxReply::KeySet) {
@@ -235,7 +235,7 @@ where
                 logger::debug!(kv_operation= %operation, value = ?value);
 
                 let result = redis_conn
-                    .serialize_and_set_key_if_not_exist(&key, value, Some(ttl.into()))
+                    .serialize_and_set_key_if_not_exist(&key.into(), value, Some(ttl.into()))
                     .await?;
 
                 value.check_for_constraints(&redis_conn).await?;
@@ -251,7 +251,9 @@ where
             }
 
             KvOperation::Get => {
-                let result = redis_conn.get_and_deserialize_key(&key, type_name).await?;
+                let result = redis_conn
+                    .get_and_deserialize_key(&key.into(), type_name)
+                    .await?;
                 Ok(KvResult::Get(result))
             }
         }

## Type of Change
<!-- Put an `x` in the boxes that apply -->
- [x] Enhancement

## Description
<!-- Describe your changes in detail -->
- Add a fallback for Redis gets with prefix
- Add a Domain type for Redis for the correctness of adding a prefix

## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->
To keep the tenant data isolated from each other, We are using a prefix for Redis keys i.e A key which would be saved as card123 without multitenancy would be saved as public:card123 with Multitenancy.
If the user makes a payment on the older pod where MultiTenancy is disabled and  client_secret is saved without a prefix, Now if the confirm calls goes to the pod where Multitenancy is enabled the Payment would fail with Unauthorized. This is just one of the instances where the payments would fail.

## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->
** THIS CANNOT BE TESTED ON HIGHER ENVIRONMENTS **

## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code