Bug: [FIX] fix cache invalidation for multitenancy




diff --git a/crates/redis_interface/src/types.rs b/crates/redis_interface/src/types.rs
index 6469df50778..f40b81af68e 100644
--- a/crates/redis_interface/src/types.rs
+++ b/crates/redis_interface/src/types.rs
@@ -27,6 +27,12 @@ impl RedisValue {
     pub fn into_inner(self) -> FredRedisValue {
         self.inner
     }
+
+    pub fn from_bytes(val: Vec<u8>) -> Self {
+        Self {
+            inner: FredRedisValue::Bytes(val.into()),
+        }
+    }
     pub fn from_string(value: String) -> Self {
         Self {
             inner: FredRedisValue::String(value.into()),
@@ -34,6 +40,12 @@ impl RedisValue {
     }
 }
 
+impl From<RedisValue> for FredRedisValue {
+    fn from(v: RedisValue) -> Self {
+        v.inner
+    }
+}
+
 #[derive(Debug, serde::Deserialize, Clone)]
 #[serde(default)]
 pub struct RedisSettings {
diff --git a/crates/storage_impl/src/redis/cache.rs b/crates/storage_impl/src/redis/cache.rs
index 6cfadbcb256..ba4a8200fa7 100644
--- a/crates/storage_impl/src/redis/cache.rs
+++ b/crates/storage_impl/src/redis/cache.rs
@@ -2,7 +2,7 @@ use std::{any::Any, borrow::Cow, fmt::Debug, sync::Arc};
 
 use common_utils::{
     errors::{self, CustomResult},
-    ext_traits::AsyncExt,
+    ext_traits::{AsyncExt, ByteSliceExt},
 };
 use dyn_clone::DynClone;
 use error_stack::{Report, ResultExt};
@@ -23,30 +23,6 @@ use crate::{
 /// Redis channel name used for publishing invalidation messages
 pub const IMC_INVALIDATION_CHANNEL: &str = "hyperswitch_invalidate";
 
-/// Prefix for config cache key
-const CONFIG_CACHE_PREFIX: &str = "config";
-
-/// Prefix for accounts cache key
-const ACCOUNTS_CACHE_PREFIX: &str = "accounts";
-
-/// Prefix for routing cache key
-const ROUTING_CACHE_PREFIX: &str = "routing";
-
-/// Prefix for three ds decision manager cache key
-const DECISION_MANAGER_CACHE_PREFIX: &str = "decision_manager";
-
-/// Prefix for surcharge cache key
-const SURCHARGE_CACHE_PREFIX: &str = "surcharge";
-
-/// Prefix for cgraph cache key
-const CGRAPH_CACHE_PREFIX: &str = "cgraph";
-
-/// Prefix for all kinds of cache key
-const ALL_CACHE_PREFIX: &str = "all_cache_kind";
-
-/// Prefix for PM Filter cgraph cache key
-const PM_FILTERS_CGRAPH_CACHE_PREFIX: &str = "pm_filters_cgraph";
-
 /// Time to live 30 mins
 const CACHE_TTL: u64 = 30 * 60;
 
@@ -101,6 +77,13 @@ pub trait Cacheable: Any + Send + Sync + DynClone {
     fn as_any(&self) -> &dyn Any;
 }
 
+#[derive(serde::Serialize, serde::Deserialize)]
+pub struct CacheRedact<'a> {
+    pub tenant: String,
+    pub kind: CacheKind<'a>,
+}
+
+#[derive(serde::Serialize, serde::Deserialize)]
 pub enum CacheKind<'a> {
     Config(Cow<'a, str>),
     Accounts(Cow<'a, str>),
@@ -112,45 +95,30 @@ pub enum CacheKind<'a> {
     All(Cow<'a, str>),
 }
 
-impl<'a> From<CacheKind<'a>> for RedisValue {
-    fn from(kind: CacheKind<'a>) -> Self {
-        let value = match kind {
-            CacheKind::Config(s) => format!("{CONFIG_CACHE_PREFIX},{s}"),
-            CacheKind::Accounts(s) => format!("{ACCOUNTS_CACHE_PREFIX},{s}"),
-            CacheKind::Routing(s) => format!("{ROUTING_CACHE_PREFIX},{s}"),
-            CacheKind::DecisionManager(s) => format!("{DECISION_MANAGER_CACHE_PREFIX},{s}"),
-            CacheKind::Surcharge(s) => format!("{SURCHARGE_CACHE_PREFIX},{s}"),
-            CacheKind::CGraph(s) => format!("{CGRAPH_CACHE_PREFIX},{s}"),
-            CacheKind::PmFiltersCGraph(s) => format!("{PM_FILTERS_CGRAPH_CACHE_PREFIX},{s}"),
-            CacheKind::All(s) => format!("{ALL_CACHE_PREFIX},{s}"),
-        };
-        Self::from_string(value)
+impl<'a> TryFrom<CacheRedact<'a>> for RedisValue {
+    type Error = Report<errors::ValidationError>;
+    fn try_from(v: CacheRedact<'a>) -> Result<Self, Self::Error> {
+        Ok(Self::from_bytes(serde_json::to_vec(&v).change_context(
+            errors::ValidationError::InvalidValue {
+                message: "Invalid publish key provided in pubsub".into(),
+            },
+        )?))
     }
 }
 
-impl<'a> TryFrom<RedisValue> for CacheKind<'a> {
+impl<'a> TryFrom<RedisValue> for CacheRedact<'a> {
     type Error = Report<errors::ValidationError>;
-    fn try_from(kind: RedisValue) -> Result<Self, Self::Error> {
-        let validation_err = errors::ValidationError::InvalidValue {
-            message: "Invalid publish key provided in pubsub".into(),
-        };
-        let kind = kind.as_string().ok_or(validation_err.clone())?;
-        let split = kind.split_once(',').ok_or(validation_err.clone())?;
-        match split.0 {
-            ACCOUNTS_CACHE_PREFIX => Ok(Self::Accounts(Cow::Owned(split.1.to_string()))),
-            CONFIG_CACHE_PREFIX => Ok(Self::Config(Cow::Owned(split.1.to_string()))),
-            ROUTING_CACHE_PREFIX => Ok(Self::Routing(Cow::Owned(split.1.to_string()))),
-            DECISION_MANAGER_CACHE_PREFIX => {
-                Ok(Self::DecisionManager(Cow::Owned(split.1.to_string())))
-            }
-            SURCHARGE_CACHE_PREFIX => Ok(Self::Surcharge(Cow::Owned(split.1.to_string()))),
-            CGRAPH_CACHE_PREFIX => Ok(Self::CGraph(Cow::Owned(split.1.to_string()))),
-            PM_FILTERS_CGRAPH_CACHE_PREFIX => {
-                Ok(Self::PmFiltersCGraph(Cow::Owned(split.1.to_string())))
-            }
-            ALL_CACHE_PREFIX => Ok(Self::All(Cow::Owned(split.1.to_string()))),
-            _ => Err(validation_err.into()),
-        }
+
+    fn try_from(v: RedisValue) -> Result<Self, Self::Error> {
+        let bytes = v.as_bytes().ok_or(errors::ValidationError::InvalidValue {
+            message: "InvalidValue received in pubsub".to_string(),
+        })?;
+
+        bytes
+            .parse_struct("CacheRedact")
+            .change_context(errors::ValidationError::InvalidValue {
+                message: "Unable to deserialize the value from pubsub".to_string(),
+            })
     }
 }
 
diff --git a/crates/storage_impl/src/redis/pub_sub.rs b/crates/storage_impl/src/redis/pub_sub.rs
index 89877f7e8dd..8f24b62fcaf 100644
--- a/crates/storage_impl/src/redis/pub_sub.rs
+++ b/crates/storage_impl/src/redis/pub_sub.rs
@@ -5,8 +5,8 @@ use redis_interface::{errors as redis_errors, PubsubInterface, RedisValue};
 use router_env::{logger, tracing::Instrument};
 
 use crate::redis::cache::{
-    CacheKey, CacheKind, ACCOUNTS_CACHE, CGRAPH_CACHE, CONFIG_CACHE, DECISION_MANAGER_CACHE,
-    PM_FILTERS_CGRAPH_CACHE, ROUTING_CACHE, SURCHARGE_CACHE,
+    CacheKey, CacheKind, CacheRedact, ACCOUNTS_CACHE, CGRAPH_CACHE, CONFIG_CACHE,
+    DECISION_MANAGER_CACHE, PM_FILTERS_CGRAPH_CACHE, ROUTING_CACHE, SURCHARGE_CACHE,
 };
 
 #[async_trait::async_trait]
@@ -66,15 +66,23 @@ impl PubSubInterface for std::sync::Arc<redis_interface::RedisConnectionPool> {
         channel: &str,
         key: CacheKind<'a>,
     ) -> error_stack::Result<usize, redis_errors::RedisError> {
+        let key = CacheRedact {
+            kind: key,
+            tenant: self.key_prefix.clone(),
+        };
+
         self.publisher
-            .publish(channel, RedisValue::from(key).into_inner())
+            .publish(
+                channel,
+                RedisValue::try_from(key).change_context(redis_errors::RedisError::PublishError)?,
+            )
             .await
             .change_context(redis_errors::RedisError::SubscribeError)
     }
 
     #[inline]
     async fn on_message(&self) -> error_stack::Result<(), redis_errors::RedisError> {
-        logger::debug!("Started on message: {:?}", self.key_prefix);
+        logger::debug!("Started on message");
         let mut rx = self.subscriber.on_message();
         while let Ok(message) = rx.recv().await {
             let channel_name = message.channel.to_string();
@@ -82,7 +90,7 @@ impl PubSubInterface for std::sync::Arc<redis_interface::RedisConnectionPool> {
 
             match channel_name.as_str() {
                 super::cache::IMC_INVALIDATION_CHANNEL => {
-                    let key = match CacheKind::try_from(RedisValue::new(message.value))
+                    let message = match CacheRedact::try_from(RedisValue::new(message.value))
                         .change_context(redis_errors::RedisError::OnMessageError)
                     {
                         Ok(value) => value,
@@ -92,12 +100,12 @@ impl PubSubInterface for std::sync::Arc<redis_interface::RedisConnectionPool> {
                         }
                     };
 
-                    let key = match key {
+                    let key = match message.kind {
                         CacheKind::Config(key) => {
                             CONFIG_CACHE
                                 .remove(CacheKey {
                                     key: key.to_string(),
-                                    prefix: self.key_prefix.clone(),
+                                    prefix: message.tenant.clone(),
                                 })
                                 .await;
                             key
@@ -106,7 +114,7 @@ impl PubSubInterface for std::sync::Arc<redis_interface::RedisConnectionPool> {
                             ACCOUNTS_CACHE
                                 .remove(CacheKey {
                                     key: key.to_string(),
-                                    prefix: self.key_prefix.clone(),
+                                    prefix: message.tenant.clone(),
                                 })
                                 .await;
                             key
@@ -115,7 +123,7 @@ impl PubSubInterface for std::sync::Arc<redis_interface::RedisConnectionPool> {
                             CGRAPH_CACHE
                                 .remove(CacheKey {
                                     key: key.to_string(),
-                                    prefix: self.key_prefix.clone(),
+                                    prefix: message.tenant.clone(),
                                 })
                                 .await;
                             key
@@ -124,7 +132,7 @@ impl PubSubInterface for std::sync::Arc<redis_interface::RedisConnectionPool> {
                             PM_FILTERS_CGRAPH_CACHE
                                 .remove(CacheKey {
                                     key: key.to_string(),
-                                    prefix: self.key_prefix.clone(),
+                                    prefix: message.tenant.clone(),
                                 })
                                 .await;
                             key
@@ -133,7 +141,7 @@ impl PubSubInterface for std::sync::Arc<redis_interface::RedisConnectionPool> {
                             ROUTING_CACHE
                                 .remove(CacheKey {
                                     key: key.to_string(),
-                                    prefix: self.key_prefix.clone(),
+                                    prefix: message.tenant.clone(),
                                 })
                                 .await;
                             key
@@ -142,7 +150,7 @@ impl PubSubInterface for std::sync::Arc<redis_interface::RedisConnectionPool> {
                             DECISION_MANAGER_CACHE
                                 .remove(CacheKey {
                                     key: key.to_string(),
-                                    prefix: self.key_prefix.clone(),
+                                    prefix: message.tenant.clone(),
                                 })
                                 .await;
                             key
@@ -151,7 +159,7 @@ impl PubSubInterface for std::sync::Arc<redis_interface::RedisConnectionPool> {
                             SURCHARGE_CACHE
                                 .remove(CacheKey {
                                     key: key.to_string(),
-                                    prefix: self.key_prefix.clone(),
+                                    prefix: message.tenant.clone(),
                                 })
                                 .await;
                             key
@@ -160,43 +168,43 @@ impl PubSubInterface for std::sync::Arc<redis_interface::RedisConnectionPool> {
                             CONFIG_CACHE
                                 .remove(CacheKey {
                                     key: key.to_string(),
-                                    prefix: self.key_prefix.clone(),
+                                    prefix: message.tenant.clone(),
                                 })
                                 .await;
                             ACCOUNTS_CACHE
                                 .remove(CacheKey {
                                     key: key.to_string(),
-                                    prefix: self.key_prefix.clone(),
+                                    prefix: message.tenant.clone(),
                                 })
                                 .await;
                             CGRAPH_CACHE
                                 .remove(CacheKey {
                                     key: key.to_string(),
-                                    prefix: self.key_prefix.clone(),
+                                    prefix: message.tenant.clone(),
                                 })
                                 .await;
                             PM_FILTERS_CGRAPH_CACHE
                                 .remove(CacheKey {
                                     key: key.to_string(),
-                                    prefix: self.key_prefix.clone(),
+                                    prefix: message.tenant.clone(),
                                 })
                                 .await;
                             ROUTING_CACHE
                                 .remove(CacheKey {
                                     key: key.to_string(),
-                                    prefix: self.key_prefix.clone(),
+                                    prefix: message.tenant.clone(),
                                 })
                                 .await;
                             DECISION_MANAGER_CACHE
                                 .remove(CacheKey {
                                     key: key.to_string(),
-                                    prefix: self.key_prefix.clone(),
+                                    prefix: message.tenant.clone(),
                                 })
                                 .await;
                             SURCHARGE_CACHE
                                 .remove(CacheKey {
                                     key: key.to_string(),
-                                    prefix: self.key_prefix.clone(),
+                                    prefix: message.tenant.clone(),
                                 })
                                 .await;
 
@@ -210,7 +218,9 @@ impl PubSubInterface for std::sync::Arc<redis_interface::RedisConnectionPool> {
                         .ok();
 
                     logger::debug!(
-                        "Handled message on channel {channel_name} - Done invalidating {key}"
+                        key_prefix=?message.tenant.clone(),
+                        channel_name=?channel_name,
+                        "Done invalidating {key}"
                     );
                 }
                 _ => {

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

## Description
<!-- Describe your changes in detail -->
Added a tenant field in the pubsub redact message. To give the information for the subscriber to delete the message for the appropriate tenant

## 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).
-->
In the current implementation,  the pub-sub message does not have any context of the tenant key for the cache it's supposed to invalidate cache for. Rather it was designed to spawn n number of tasks for cache invalidation for n tenants (This was fixed few weeks ago). Now the problem is there's only one thread which is spawned for n tenants, so whenever the cache is invalidated it will only be invalidated for the first tenant in the list.

This PR fixes it by passing the tenant information in the pub-sub message, from which we can take the tenant id and invalidate cache for the particular tenant.

## 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 sandbox without enabling multi tenancy or adding `redis_key_prefix` for the tenant**

1. Enable multitenancy by adding another tenant
```bash
[multitenancy]
enabled = true
global_tenant = { schema = "public", redis_key_prefix = "", clickhouse_database = "default"}

[multitenancy.tenants]
public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = "hyperswitch", clickhouse_database = "default"}
foo = { name = "bar", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = "bar", clickhouse_database = "default"}
```

2. Create merchant account with X-tenant-id as foo
```bash
curl --location 'http://localhost:8080/accounts' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'x-feature: integ-custom' \
--header 'x-tenant-id: foo' \
--header 'api-key: test_admin' \
--data-raw '{
  "merchant_id": "merchant_1723049018",
  "locker_id": "m0010",
  "merchant_name": "NewAge Retailer",
  "merchant_details": {
    "primary_contact_person": "John Test",
    "primary_email": "JohnTest@test.com",
    "primary_phone": "sunt laborum",
    "secondary_contact_person": "John Test2",
    "secondary_email": "JohnTest2@test.com",
    "secondary_phone": "cillum do dolor id",
    "website": "www.example.com",
    "about_business": "Online Retail with a wide selection of organic products for North America",
    "address": {
      "line1": "1467",
      "line2": "Harrison Street",
      "line3": "Harrison Street",
      "city": "San Fransico",
      "state": "California",
      "zip": "94122",
      "country": "US"
    }
  },
  "return_url": "https://google.com/success",
  "webhook_details": {
    "webhook_version": "1.0.1",
    "webhook_username": "ekart_retail",
    "webhook_password": "password_ekart@123",
    "payment_created_enabled": true,
    "payment_succeeded_enabled": true,
    "payment_failed_enabled": true
  },
  "sub_merchants_enabled": false,
  "metadata": {
    "city": "NY",
    "unit": "245"
  },
  "primary_business_details": [
    {
      "country": "US",
      "business": "default"
    }
  ]
}'
```
Check that key is invalidated for the subsequent `redis_key_prefix` of the tenant

![Screenshot 2024-08-07 at 10 14 36 PM](https://github.com/user-attachments/assets/9f7a84cb-efdb-4238-872e-731d3651e634)

3. Create merchant account with x-tenant-id as public
```bash
curl --location 'http://localhost:8080/accounts' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'x-feature: integ-custom' \
--header 'x-tenant-id: public' \
--header 'api-key: test_admin' \
--data-raw '{
  "merchant_id": "merchant_1723049152",
  "locker_id": "m0010",
  "merchant_name": "NewAge Retailer",
  "merchant_details": {
    "primary_contact_person": "John Test",
    "primary_email": "JohnTest@test.com",
    "primary_phone": "sunt laborum",
    "secondary_contact_person": "John Test2",
    "secondary_email": "JohnTest2@test.com",
    "secondary_phone": "cillum do dolor id",
    "website": "www.example.com",
    "about_business": "Online Retail with a wide selection of organic products for North America",
    "address": {
      "line1": "1467",
      "line2": "Harrison Street",
      "line3": "Harrison Street",
      "city": "San Fransico",
      "state": "California",
      "zip": "94122",
      "country": "US"
    }
  },
  "return_url": "https://google.com/success",
  "webhook_details": {
    "webhook_version": "1.0.1",
    "webhook_username": "ekart_retail",
    "webhook_password": "password_ekart@123",
    "payment_created_enabled": true,
    "payment_succeeded_enabled": true,
    "payment_failed_enabled": true
  },
  "sub_merchants_enabled": false,
  "metadata": {
    "city": "NY",
    "unit": "245"
  },
  "primary_business_details": [
    {
      "country": "US",
      "business": "default"
    }
  ]
}'
```
Check that key is invalidated for the subsequent `redis_key_prefix` of the tenant

![Screenshot 2024-08-07 at 10 15 40 PM](https://github.com/user-attachments/assets/94ce138b-b38e-4c63-ad3f-01d18e36891f)


## 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
