Issue:
Bug: [REFACTOR] spawn one subscriber thread for handling all the published messages to different channel



In the current Redis pub-sub implementation, a subscriber spawns a thread to handle all messages published to a channel. As a result, if a subscriber is invoked multiple times (even for subscribing to the same channel), it creates multiple threads for handling messages, which is not the desired behaviour. Ideally, we want to have only one subscriber thread handling all messages published to different channels. To achieve this, we maintain an atomic boolean variable, initially set to `false`, indicating whether the subscriber handler thread has been spawned. This variable will be set to `true` during the first subscriber invocation, thus preventing the creation of multiple threads during subsequent subscriber invocations. 

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [ ] New feature
- [ ] Enhancement
- [x] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
In the current Redis pub-sub implementation, a subscriber spawns a thread to handle all messages published to a channel. As a result, if a subscriber is invoked multiple times (even for subscribing to the same channel), it creates multiple threads for handling messages, which is not the desired behaviour. Ideally, we want to have only one subscriber thread handling all messages published to different channels. To achieve this, we maintain an atomic boolean variable, initially set to `false`, indicating whether the subscriber handler thread has been spawned. This variable will be set to `true` during the first subscriber invocation, thus preventing the creation of multiple threads during subsequent subscriber invocations. 

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## 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).
-->


## 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)?
-->
In the current implementation, the subscriber has been invoked 2 times thus resulting in 2 threads being spawned. Consequently, cache invalidation attempts occur in both threads, with only one succeeding. With this PR, there'll always be only one subscriber thread responsible for invalidating the cache entry.

* Send a request to the `/configs` endpoint to add a new test config.

```
curl --location 'http://localhost:8080/configs/' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: test_admin' \
--data '{
    "key": "test_key",
    "value": "test_val"
}'
```

* This action triggers cache invalidation.
* Check the request logs. You should see only one log entry stating: "Done invalidating test_key"

## 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
- [ ] I added unit tests for my changes where possible


Testing:

In the current implementation, the subscriber has been invoked 2 times thus resulting in 2 threads being spawned. Consequently, cache invalidation attempts occur in both threads, with only one succeeding. With this PR, there'll always be only one subscriber thread responsible for invalidating the cache entry.

* Send a request to the `/configs` endpoint to add a new test config.

```
curl --location 'http://localhost:8080/configs/' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: test_admin' \
--data '{
    "key": "test_key",
    "value": "test_val"
}'
```

* This action triggers cache invalidation.
* Check the request logs. You should see only one log entry stating: "Done invalidating test_key"

Patch:
diff --git a/crates/redis_interface/src/lib.rs b/crates/redis_interface/src/lib.rs
index dc4fd3bbc9c..fa35039800b 100644
--- a/crates/redis_interface/src/lib.rs
+++ b/crates/redis_interface/src/lib.rs
@@ -68,6 +68,7 @@ impl RedisClient {
 
 pub struct SubscriberClient {
     inner: fred::clients::SubscriberClient,
+    pub is_subscriber_handler_spawned: Arc<atomic::AtomicBool>,
 }
 
 impl SubscriberClient {
@@ -83,7 +84,10 @@ impl SubscriberClient {
             .wait_for_connect()
             .await
             .change_context(errors::RedisError::RedisConnectionError)?;
-        Ok(Self { inner: client })
+        Ok(Self {
+            inner: client,
+            is_subscriber_handler_spawned: Arc::new(atomic::AtomicBool::new(false)),
+        })
     }
 }
 
diff --git a/crates/router/src/db/configs.rs b/crates/router/src/db/configs.rs
index fb399a7084f..575481793ca 100644
--- a/crates/router/src/db/configs.rs
+++ b/crates/router/src/db/configs.rs
@@ -72,7 +72,7 @@ impl ConfigInterface for Store {
         self.get_redis_conn()
             .map_err(Into::<errors::StorageError>::into)?
             .publish(
-                cache::PUB_SUB_CHANNEL,
+                cache::IMC_INVALIDATION_CHANNEL,
                 CacheKind::Config((&inserted.key).into()),
             )
             .await
@@ -179,7 +179,10 @@ impl ConfigInterface for Store {
 
         self.get_redis_conn()
             .map_err(Into::<errors::StorageError>::into)?
-            .publish(cache::PUB_SUB_CHANNEL, CacheKind::Config(key.into()))
+            .publish(
+                cache::IMC_INVALIDATION_CHANNEL,
+                CacheKind::Config(key.into()),
+            )
             .await
             .map_err(Into::<errors::StorageError>::into)?;
 
diff --git a/crates/router/src/services.rs b/crates/router/src/services.rs
index 58b5b56fd21..38f859c48e7 100644
--- a/crates/router/src/services.rs
+++ b/crates/router/src/services.rs
@@ -74,7 +74,7 @@ pub async fn get_store(
             tenant,
             master_enc_key,
             cache_store,
-            storage_impl::redis::cache::PUB_SUB_CHANNEL,
+            storage_impl::redis::cache::IMC_INVALIDATION_CHANNEL,
         )
         .await?
     };
diff --git a/crates/storage_impl/src/redis/cache.rs b/crates/storage_impl/src/redis/cache.rs
index 46aa3f8aa96..036a8902bc3 100644
--- a/crates/storage_impl/src/redis/cache.rs
+++ b/crates/storage_impl/src/redis/cache.rs
@@ -21,7 +21,7 @@ use crate::{
 };
 
 /// Redis channel name used for publishing invalidation messages
-pub const PUB_SUB_CHANNEL: &str = "hyperswitch_invalidate";
+pub const IMC_INVALIDATION_CHANNEL: &str = "hyperswitch_invalidate";
 
 /// Prefix for config cache key
 const CONFIG_CACHE_PREFIX: &str = "config";
@@ -392,7 +392,7 @@ pub async fn publish_into_redact_channel<'a, K: IntoIterator<Item = CacheKind<'a
     let futures = keys.into_iter().map(|key| async {
         redis_conn
             .clone()
-            .publish(PUB_SUB_CHANNEL, key)
+            .publish(IMC_INVALIDATION_CHANNEL, key)
             .await
             .change_context(StorageError::KVError)
     });
diff --git a/crates/storage_impl/src/redis/pub_sub.rs b/crates/storage_impl/src/redis/pub_sub.rs
index a50de0a4edd..670ffbb54e1 100644
--- a/crates/storage_impl/src/redis/pub_sub.rs
+++ b/crates/storage_impl/src/redis/pub_sub.rs
@@ -1,3 +1,5 @@
+use std::sync::atomic;
+
 use error_stack::ResultExt;
 use redis_interface::{errors as redis_errors, PubsubInterface, RedisValue};
 use router_env::{logger, tracing::Instrument};
@@ -32,15 +34,29 @@ impl PubSubInterface for std::sync::Arc<redis_interface::RedisConnectionPool> {
             .await
             .change_context(redis_errors::RedisError::SubscribeError)?;
 
-        let redis_clone = self.clone();
-        let _task_handle = tokio::spawn(
-            async move {
-                if let Err(pubsub_error) = redis_clone.on_message().await {
-                    logger::error!(?pubsub_error);
+        // Spawn only one thread handling all the published messages to different channels
+        if self
+            .subscriber
+            .is_subscriber_handler_spawned
+            .compare_exchange(
+                false,
+                true,
+                atomic::Ordering::SeqCst,
+                atomic::Ordering::SeqCst,
+            )
+            .is_ok()
+        {
+            let redis_clone = self.clone();
+            let _task_handle = tokio::spawn(
+                async move {
+                    if let Err(pubsub_error) = redis_clone.on_message().await {
+                        logger::error!(?pubsub_error);
+                    }
                 }
-            }
-            .in_current_span(),
-        );
+                .in_current_span(),
+            );
+        }
+
         Ok(())
     }
 
@@ -61,120 +77,131 @@ impl PubSubInterface for std::sync::Arc<redis_interface::RedisConnectionPool> {
         logger::debug!("Started on message: {:?}", self.key_prefix);
         let mut rx = self.subscriber.on_message();
         while let Ok(message) = rx.recv().await {
-            logger::debug!("Invalidating {message:?}");
-            let key = match CacheKind::try_from(RedisValue::new(message.value))
-                .change_context(redis_errors::RedisError::OnMessageError)
-            {
-                Ok(value) => value,
-                Err(err) => {
-                    logger::error!(value_conversion_err=?err);
-                    continue;
-                }
-            };
-
-            let key = match key {
-                CacheKind::Config(key) => {
-                    CONFIG_CACHE
-                        .remove(CacheKey {
-                            key: key.to_string(),
-                            prefix: self.key_prefix.clone(),
-                        })
-                        .await;
-                    key
-                }
-                CacheKind::Accounts(key) => {
-                    ACCOUNTS_CACHE
-                        .remove(CacheKey {
-                            key: key.to_string(),
-                            prefix: self.key_prefix.clone(),
-                        })
-                        .await;
-                    key
-                }
-                CacheKind::CGraph(key) => {
-                    CGRAPH_CACHE
-                        .remove(CacheKey {
-                            key: key.to_string(),
-                            prefix: self.key_prefix.clone(),
-                        })
-                        .await;
-                    key
-                }
-                CacheKind::Routing(key) => {
-                    ROUTING_CACHE
-                        .remove(CacheKey {
-                            key: key.to_string(),
-                            prefix: self.key_prefix.clone(),
-                        })
-                        .await;
-                    key
-                }
-                CacheKind::DecisionManager(key) => {
-                    DECISION_MANAGER_CACHE
-                        .remove(CacheKey {
-                            key: key.to_string(),
-                            prefix: self.key_prefix.clone(),
-                        })
-                        .await;
-                    key
-                }
-                CacheKind::Surcharge(key) => {
-                    SURCHARGE_CACHE
-                        .remove(CacheKey {
-                            key: key.to_string(),
-                            prefix: self.key_prefix.clone(),
-                        })
-                        .await;
-                    key
-                }
-                CacheKind::All(key) => {
-                    CONFIG_CACHE
-                        .remove(CacheKey {
-                            key: key.to_string(),
-                            prefix: self.key_prefix.clone(),
-                        })
-                        .await;
-                    ACCOUNTS_CACHE
-                        .remove(CacheKey {
-                            key: key.to_string(),
-                            prefix: self.key_prefix.clone(),
-                        })
-                        .await;
-                    CGRAPH_CACHE
-                        .remove(CacheKey {
-                            key: key.to_string(),
-                            prefix: self.key_prefix.clone(),
-                        })
-                        .await;
-                    ROUTING_CACHE
-                        .remove(CacheKey {
-                            key: key.to_string(),
-                            prefix: self.key_prefix.clone(),
-                        })
-                        .await;
-                    DECISION_MANAGER_CACHE
-                        .remove(CacheKey {
-                            key: key.to_string(),
-                            prefix: self.key_prefix.clone(),
-                        })
-                        .await;
-                    SURCHARGE_CACHE
-                        .remove(CacheKey {
-                            key: key.to_string(),
-                            prefix: self.key_prefix.clone(),
-                        })
-                        .await;
-
-                    key
-                }
-            };
+            let channel_name = message.channel.to_string();
+            logger::debug!("Received message on channel: {channel_name}");
+
+            match channel_name.as_str() {
+                super::cache::IMC_INVALIDATION_CHANNEL => {
+                    let key = match CacheKind::try_from(RedisValue::new(message.value))
+                        .change_context(redis_errors::RedisError::OnMessageError)
+                    {
+                        Ok(value) => value,
+                        Err(err) => {
+                            logger::error!(value_conversion_err=?err);
+                            continue;
+                        }
+                    };
+
+                    let key = match key {
+                        CacheKind::Config(key) => {
+                            CONFIG_CACHE
+                                .remove(CacheKey {
+                                    key: key.to_string(),
+                                    prefix: self.key_prefix.clone(),
+                                })
+                                .await;
+                            key
+                        }
+                        CacheKind::Accounts(key) => {
+                            ACCOUNTS_CACHE
+                                .remove(CacheKey {
+                                    key: key.to_string(),
+                                    prefix: self.key_prefix.clone(),
+                                })
+                                .await;
+                            key
+                        }
+                        CacheKind::CGraph(key) => {
+                            CGRAPH_CACHE
+                                .remove(CacheKey {
+                                    key: key.to_string(),
+                                    prefix: self.key_prefix.clone(),
+                                })
+                                .await;
+                            key
+                        }
+                        CacheKind::Routing(key) => {
+                            ROUTING_CACHE
+                                .remove(CacheKey {
+                                    key: key.to_string(),
+                                    prefix: self.key_prefix.clone(),
+                                })
+                                .await;
+                            key
+                        }
+                        CacheKind::DecisionManager(key) => {
+                            DECISION_MANAGER_CACHE
+                                .remove(CacheKey {
+                                    key: key.to_string(),
+                                    prefix: self.key_prefix.clone(),
+                                })
+                                .await;
+                            key
+                        }
+                        CacheKind::Surcharge(key) => {
+                            SURCHARGE_CACHE
+                                .remove(CacheKey {
+                                    key: key.to_string(),
+                                    prefix: self.key_prefix.clone(),
+                                })
+                                .await;
+                            key
+                        }
+                        CacheKind::All(key) => {
+                            CONFIG_CACHE
+                                .remove(CacheKey {
+                                    key: key.to_string(),
+                                    prefix: self.key_prefix.clone(),
+                                })
+                                .await;
+                            ACCOUNTS_CACHE
+                                .remove(CacheKey {
+                                    key: key.to_string(),
+                                    prefix: self.key_prefix.clone(),
+                                })
+                                .await;
+                            CGRAPH_CACHE
+                                .remove(CacheKey {
+                                    key: key.to_string(),
+                                    prefix: self.key_prefix.clone(),
+                                })
+                                .await;
+                            ROUTING_CACHE
+                                .remove(CacheKey {
+                                    key: key.to_string(),
+                                    prefix: self.key_prefix.clone(),
+                                })
+                                .await;
+                            DECISION_MANAGER_CACHE
+                                .remove(CacheKey {
+                                    key: key.to_string(),
+                                    prefix: self.key_prefix.clone(),
+                                })
+                                .await;
+                            SURCHARGE_CACHE
+                                .remove(CacheKey {
+                                    key: key.to_string(),
+                                    prefix: self.key_prefix.clone(),
+                                })
+                                .await;
 
-            self.delete_key(key.as_ref())
-                .await
-                .map_err(|err| logger::error!("Error while deleting redis key: {err:?}"))
-                .ok();
+                            key
+                        }
+                    };
 
-            logger::debug!("Done invalidating {key}");
+                    self.delete_key(key.as_ref())
+                        .await
+                        .map_err(|err| logger::error!("Error while deleting redis key: {err:?}"))
+                        .ok();
+
+                    logger::debug!(
+                        "Handled message on channel {channel_name} - Done invalidating {key}"
+                    );
+                }
+                _ => {
+                    logger::debug!("Received message from unknown channel: {channel_name}");
+                }
+            }
         }
         Ok(())
     }