Bug: [FEATURE] Replace Routing cache with Moka



### Feature Description

Have to replace static routing cache with moka cache in order to use extra features like TTL

### Possible Implementation

Just need to replace the current cache.

### Have you spent some time checking if this feature request has been raised before?

- [X] I checked and didn't find a similar issue

### Have you read the Contributing Guidelines?

- [X] I have read the [Contributing Guidelines](https://github.com/juspay/hyperswitch/blob/main/docs/CONTRIBUTING.md)

### Are you willing to submit a PR?

Yes, I am willing to submit a PR!
diff --git a/crates/router/src/core/admin.rs b/crates/router/src/core/admin.rs
index 4c2ebb516e6..b50c5b69681 100644
--- a/crates/router/src/core/admin.rs
+++ b/crates/router/src/core/admin.rs
@@ -1245,17 +1245,6 @@ pub async fn update_payment_connector(
         }
     }
 
-    // The purpose of this merchant account update is just to update the
-    // merchant account `modified_at` field for KGraph cache invalidation
-    db.update_specific_fields_in_merchant(
-        merchant_id,
-        storage::MerchantAccountUpdate::ModifiedAtUpdate,
-        &key_store,
-    )
-    .await
-    .change_context(errors::ApiErrorResponse::InternalServerError)
-    .attach_printable("error updating the merchant account when updating payment connector")?;
-
     let payment_connector = storage::MerchantConnectorAccountUpdate::Update {
         merchant_id: None,
         connector_type: Some(req.connector_type),
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index ff48d181a77..175400c54bf 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -3192,7 +3192,6 @@ where
             connectors = routing::perform_eligibility_analysis_with_fallback(
                 &state.clone(),
                 key_store,
-                merchant_account.modified_at.assume_utc().unix_timestamp(),
                 connectors,
                 &TransactionData::Payment(payment_data),
                 eligible_connectors,
@@ -3250,7 +3249,6 @@ where
             connectors = routing::perform_eligibility_analysis_with_fallback(
                 &state,
                 key_store,
-                merchant_account.modified_at.assume_utc().unix_timestamp(),
                 connectors,
                 &TransactionData::Payment(payment_data),
                 eligible_connectors,
@@ -3680,7 +3678,6 @@ where
     let connectors = routing::perform_eligibility_analysis_with_fallback(
         &state.clone(),
         key_store,
-        merchant_account.modified_at.assume_utc().unix_timestamp(),
         connectors,
         &transaction_data,
         eligible_connectors,
diff --git a/crates/router/src/core/payments/routing.rs b/crates/router/src/core/payments/routing.rs
index 802ba524cc6..d45f804268b 100644
--- a/crates/router/src/core/payments/routing.rs
+++ b/crates/router/src/core/payments/routing.rs
@@ -13,7 +13,6 @@ use api_models::{
     payments::Address,
     routing::ConnectorSelection,
 };
-use common_utils::static_cache::StaticCache;
 use diesel_models::enums as storage_enums;
 use error_stack::ResultExt;
 use euclid::{
@@ -33,6 +32,7 @@ use rand::{
     SeedableRng,
 };
 use rustc_hash::FxHashMap;
+use storage_impl::redis::cache::{CGRAPH_CACHE, ROUTING_CACHE};
 
 #[cfg(feature = "payouts")]
 use crate::core::payouts;
@@ -53,7 +53,7 @@ use crate::{
     AppState,
 };
 
-pub(super) enum CachedAlgorithm {
+pub enum CachedAlgorithm {
     Single(Box<routing_types::RoutableConnectorChoice>),
     Priority(Vec<routing_types::RoutableConnectorChoice>),
     VolumeSplit(Vec<routing_types::ConnectorVolumeSplit>),
@@ -73,7 +73,6 @@ pub struct SessionFlowRoutingInput<'a> {
 pub struct SessionRoutingPmTypeInput<'a> {
     state: &'a AppState,
     key_store: &'a domain::MerchantKeyStore,
-    merchant_last_modified: i64,
     attempt_id: &'a str,
     routing_algorithm: &'a MerchantAccountRoutingAlgorithm,
     backend_input: dsl_inputs::BackendInput,
@@ -84,10 +83,6 @@ pub struct SessionRoutingPmTypeInput<'a> {
     ))]
     profile_id: Option<String>,
 }
-static ROUTING_CACHE: StaticCache<CachedAlgorithm> = StaticCache::new();
-static KGRAPH_CACHE: StaticCache<
-    hyperswitch_constraint_graph::ConstraintGraph<'_, euclid_dir::DirValue>,
-> = StaticCache::new();
 
 type RoutingResult<O> = oss_errors::CustomResult<O, errors::RoutingError>;
 
@@ -302,20 +297,15 @@ pub async fn perform_static_routing_v1<F: Clone>(
 
         return Ok(fallback_config);
     };
-    let key = ensure_algorithm_cached_v1(
+    let cached_algorithm = ensure_algorithm_cached_v1(
         state,
         merchant_id,
-        algorithm_ref.timestamp,
         &algorithm_id,
         #[cfg(feature = "business_profile_routing")]
         Some(profile_id).cloned(),
         &api_enums::TransactionType::from(transaction_data),
     )
     .await?;
-    let cached_algorithm: Arc<CachedAlgorithm> = ROUTING_CACHE
-        .retrieve(&key)
-        .change_context(errors::RoutingError::CacheMiss)
-        .attach_printable("Unable to retrieve cached routing algorithm even after refresh")?;
 
     Ok(match cached_algorithm.as_ref() {
         CachedAlgorithm::Single(conn) => vec![(**conn).clone()],
@@ -342,11 +332,10 @@ pub async fn perform_static_routing_v1<F: Clone>(
 async fn ensure_algorithm_cached_v1(
     state: &AppState,
     merchant_id: &str,
-    timestamp: i64,
     algorithm_id: &str,
     #[cfg(feature = "business_profile_routing")] profile_id: Option<String>,
     transaction_type: &api_enums::TransactionType,
-) -> RoutingResult<String> {
+) -> RoutingResult<Arc<CachedAlgorithm>> {
     #[cfg(feature = "business_profile_routing")]
     let key = {
         let profile_id = profile_id
@@ -376,29 +365,24 @@ async fn ensure_algorithm_cached_v1(
         }
     };
 
-    let present = ROUTING_CACHE
-        .present(&key)
-        .change_context(errors::RoutingError::DslCachePoisoned)
-        .attach_printable("Error checking presence of DSL")?;
+    let cached_algorithm = ROUTING_CACHE
+        .get_val::<Arc<CachedAlgorithm>>(key.as_str())
+        .await;
 
-    let expired = ROUTING_CACHE
-        .expired(&key, timestamp)
-        .change_context(errors::RoutingError::DslCachePoisoned)
-        .attach_printable("Error checking expiry of DSL in cache")?;
-
-    if !present || expired {
+    let algorithm = if let Some(algo) = cached_algorithm {
+        algo
+    } else {
         refresh_routing_cache_v1(
             state,
             key.clone(),
             algorithm_id,
-            timestamp,
             #[cfg(feature = "business_profile_routing")]
             profile_id,
         )
-        .await?;
+        .await?
     };
 
-    Ok(key)
+    Ok(algorithm)
 }
 
 pub fn perform_straight_through_routing(
@@ -447,9 +431,8 @@ pub async fn refresh_routing_cache_v1(
     state: &AppState,
     key: String,
     algorithm_id: &str,
-    timestamp: i64,
     #[cfg(feature = "business_profile_routing")] profile_id: Option<String>,
-) -> RoutingResult<()> {
+) -> RoutingResult<Arc<CachedAlgorithm>> {
     #[cfg(feature = "business_profile_routing")]
     let algorithm = {
         let algorithm = state
@@ -498,12 +481,11 @@ pub async fn refresh_routing_cache_v1(
         }
     };
 
-    ROUTING_CACHE
-        .save(key, cached_algorithm, timestamp)
-        .change_context(errors::RoutingError::DslCachePoisoned)
-        .attach_printable("Error saving DSL to cache")?;
+    let arc_cached_algorithm = Arc::new(cached_algorithm);
+
+    ROUTING_CACHE.push(key, arc_cached_algorithm.clone()).await;
 
-    Ok(())
+    Ok(arc_cached_algorithm)
 }
 
 pub fn perform_volume_split(
@@ -540,10 +522,9 @@ pub fn perform_volume_split(
     Ok(splits.into_iter().map(|sp| sp.connector).collect())
 }
 
-pub async fn get_merchant_kgraph<'a>(
+pub async fn get_merchant_cgraph<'a>(
     state: &AppState,
     key_store: &domain::MerchantKeyStore,
-    merchant_last_modified: i64,
     #[cfg(feature = "business_profile_routing")] profile_id: Option<String>,
     transaction_type: &api_enums::TransactionType,
 ) -> RoutingResult<Arc<hyperswitch_constraint_graph::ConstraintGraph<'a, euclid_dir::DirValue>>> {
@@ -556,10 +537,10 @@ pub async fn get_merchant_kgraph<'a>(
             .get_required_value("profile_id")
             .change_context(errors::RoutingError::ProfileIdMissing)?;
         match transaction_type {
-            api_enums::TransactionType::Payment => format!("kgraph_{}_{}", merchant_id, profile_id),
+            api_enums::TransactionType::Payment => format!("cgraph_{}_{}", merchant_id, profile_id),
             #[cfg(feature = "payouts")]
             api_enums::TransactionType::Payout => {
-                format!("kgraph_po_{}_{}", merchant_id, profile_id)
+                format!("cgraph_po_{}_{}", merchant_id, profile_id)
             }
         }
     };
@@ -571,45 +552,36 @@ pub async fn get_merchant_kgraph<'a>(
         api_enums::TransactionType::Payout => format!("kgraph_po_{}", merchant_id),
     };
 
-    let kgraph_present = KGRAPH_CACHE
-        .present(&key)
-        .change_context(errors::RoutingError::KgraphCacheFailure)
-        .attach_printable("when checking kgraph presence")?;
-
-    let kgraph_expired = KGRAPH_CACHE
-        .expired(&key, merchant_last_modified)
-        .change_context(errors::RoutingError::KgraphCacheFailure)
-        .attach_printable("when checking kgraph expiry")?;
+    let cached_cgraph = CGRAPH_CACHE
+        .get_val::<Arc<hyperswitch_constraint_graph::ConstraintGraph<'_, euclid_dir::DirValue>>>(
+            key.as_str(),
+        )
+        .await;
 
-    if !kgraph_present || kgraph_expired {
-        refresh_kgraph_cache(
+    let cgraph = if let Some(graph) = cached_cgraph {
+        graph
+    } else {
+        refresh_cgraph_cache(
             state,
             key_store,
-            merchant_last_modified,
             key.clone(),
             #[cfg(feature = "business_profile_routing")]
             profile_id,
             transaction_type,
         )
-        .await?;
-    }
-
-    let cached_kgraph = KGRAPH_CACHE
-        .retrieve(&key)
-        .change_context(errors::RoutingError::CacheMiss)
-        .attach_printable("when retrieving kgraph")?;
+        .await?
+    };
 
-    Ok(cached_kgraph)
+    Ok(cgraph)
 }
 
-pub async fn refresh_kgraph_cache(
+pub async fn refresh_cgraph_cache<'a>(
     state: &AppState,
     key_store: &domain::MerchantKeyStore,
-    timestamp: i64,
     key: String,
     #[cfg(feature = "business_profile_routing")] profile_id: Option<String>,
     transaction_type: &api_enums::TransactionType,
-) -> RoutingResult<()> {
+) -> RoutingResult<Arc<hyperswitch_constraint_graph::ConstraintGraph<'a, euclid_dir::DirValue>>> {
     let mut merchant_connector_accounts = state
         .store
         .find_merchant_connector_account_by_merchant_id_and_disabled_list(
@@ -672,23 +644,21 @@ pub async fn refresh_kgraph_cache(
         connector_configs,
         default_configs,
     };
-    let kgraph = mca_graph::make_mca_graph(api_mcas, &config_pm_filters)
-        .change_context(errors::RoutingError::KgraphCacheRefreshFailed)
-        .attach_printable("when construction kgraph")?;
+    let cgraph = Arc::new(
+        mca_graph::make_mca_graph(api_mcas, &config_pm_filters)
+            .change_context(errors::RoutingError::KgraphCacheRefreshFailed)
+            .attach_printable("when construction cgraph")?,
+    );
 
-    KGRAPH_CACHE
-        .save(key, kgraph, timestamp)
-        .change_context(errors::RoutingError::KgraphCacheRefreshFailed)
-        .attach_printable("when saving kgraph to cache")?;
+    CGRAPH_CACHE.push(key, Arc::clone(&cgraph)).await;
 
-    Ok(())
+    Ok(cgraph)
 }
 
 #[allow(clippy::too_many_arguments)]
-async fn perform_kgraph_filtering(
+async fn perform_cgraph_filtering(
     state: &AppState,
     key_store: &domain::MerchantKeyStore,
-    merchant_last_modified: i64,
     chosen: Vec<routing_types::RoutableConnectorChoice>,
     backend_input: dsl_inputs::BackendInput,
     eligible_connectors: Option<&Vec<api_enums::RoutableConnectors>>,
@@ -700,10 +670,9 @@ async fn perform_kgraph_filtering(
             .into_context()
             .change_context(errors::RoutingError::KgraphAnalysisError)?,
     );
-    let cached_kgraph = get_merchant_kgraph(
+    let cached_cgraph = get_merchant_cgraph(
         state,
         key_store,
-        merchant_last_modified,
         #[cfg(feature = "business_profile_routing")]
         profile_id,
         transaction_type,
@@ -717,7 +686,7 @@ async fn perform_kgraph_filtering(
         let dir_val = euclid_choice
             .into_dir_value()
             .change_context(errors::RoutingError::KgraphAnalysisError)?;
-        let kgraph_eligible = cached_kgraph
+        let cgraph_eligible = cached_cgraph
             .check_value_validity(
                 dir_val,
                 &context,
@@ -730,7 +699,7 @@ async fn perform_kgraph_filtering(
         let filter_eligible =
             eligible_connectors.map_or(true, |list| list.contains(&routable_connector));
 
-        if kgraph_eligible && filter_eligible {
+        if cgraph_eligible && filter_eligible {
             final_selection.push(choice);
         }
     }
@@ -741,7 +710,6 @@ async fn perform_kgraph_filtering(
 pub async fn perform_eligibility_analysis<F: Clone>(
     state: &AppState,
     key_store: &domain::MerchantKeyStore,
-    merchant_last_modified: i64,
     chosen: Vec<routing_types::RoutableConnectorChoice>,
     transaction_data: &routing::TransactionData<'_, F>,
     eligible_connectors: Option<&Vec<api_enums::RoutableConnectors>>,
@@ -753,10 +721,9 @@ pub async fn perform_eligibility_analysis<F: Clone>(
         routing::TransactionData::Payout(payout_data) => make_dsl_input_for_payouts(payout_data)?,
     };
 
-    perform_kgraph_filtering(
+    perform_cgraph_filtering(
         state,
         key_store,
-        merchant_last_modified,
         chosen,
         backend_input,
         eligible_connectors,
@@ -770,7 +737,6 @@ pub async fn perform_eligibility_analysis<F: Clone>(
 pub async fn perform_fallback_routing<F: Clone>(
     state: &AppState,
     key_store: &domain::MerchantKeyStore,
-    merchant_last_modified: i64,
     transaction_data: &routing::TransactionData<'_, F>,
     eligible_connectors: Option<&Vec<api_enums::RoutableConnectors>>,
     #[cfg(feature = "business_profile_routing")] profile_id: Option<String>,
@@ -801,10 +767,9 @@ pub async fn perform_fallback_routing<F: Clone>(
         routing::TransactionData::Payout(payout_data) => make_dsl_input_for_payouts(payout_data)?,
     };
 
-    perform_kgraph_filtering(
+    perform_cgraph_filtering(
         state,
         key_store,
-        merchant_last_modified,
         fallback_config,
         backend_input,
         eligible_connectors,
@@ -818,7 +783,6 @@ pub async fn perform_fallback_routing<F: Clone>(
 pub async fn perform_eligibility_analysis_with_fallback<F: Clone>(
     state: &AppState,
     key_store: &domain::MerchantKeyStore,
-    merchant_last_modified: i64,
     chosen: Vec<routing_types::RoutableConnectorChoice>,
     transaction_data: &routing::TransactionData<'_, F>,
     eligible_connectors: Option<Vec<api_enums::RoutableConnectors>>,
@@ -827,7 +791,6 @@ pub async fn perform_eligibility_analysis_with_fallback<F: Clone>(
     let mut final_selection = perform_eligibility_analysis(
         state,
         key_store,
-        merchant_last_modified,
         chosen,
         transaction_data,
         eligible_connectors.as_ref(),
@@ -839,7 +802,6 @@ pub async fn perform_eligibility_analysis_with_fallback<F: Clone>(
     let fallback_selection = perform_fallback_routing(
         state,
         key_store,
-        merchant_last_modified,
         transaction_data,
         eligible_connectors.as_ref(),
         #[cfg(feature = "business_profile_routing")]
@@ -873,11 +835,6 @@ pub async fn perform_session_flow_routing(
 ) -> RoutingResult<FxHashMap<api_enums::PaymentMethodType, routing_types::SessionRoutingChoice>> {
     let mut pm_type_map: FxHashMap<api_enums::PaymentMethodType, FxHashMap<String, api::GetToken>> =
         FxHashMap::default();
-    let merchant_last_modified = session_input
-        .merchant_account
-        .modified_at
-        .assume_utc()
-        .unix_timestamp();
 
     #[cfg(feature = "business_profile_routing")]
     let routing_algorithm: MerchantAccountRoutingAlgorithm = {
@@ -995,7 +952,6 @@ pub async fn perform_session_flow_routing(
         let session_pm_input = SessionRoutingPmTypeInput {
             state: session_input.state,
             key_store: session_input.key_store,
-            merchant_last_modified,
             attempt_id: &session_input.payment_attempt.attempt_id,
             routing_algorithm: &routing_algorithm,
             backend_input: backend_input.clone(),
@@ -1035,10 +991,9 @@ async fn perform_session_routing_for_pm_type(
     let chosen_connectors = match session_pm_input.routing_algorithm {
         MerchantAccountRoutingAlgorithm::V1(algorithm_ref) => {
             if let Some(ref algorithm_id) = algorithm_ref.algorithm_id {
-                let key = ensure_algorithm_cached_v1(
+                let cached_algorithm = ensure_algorithm_cached_v1(
                     &session_pm_input.state.clone(),
                     merchant_id,
-                    algorithm_ref.timestamp,
                     algorithm_id,
                     #[cfg(feature = "business_profile_routing")]
                     session_pm_input.profile_id.clone(),
@@ -1046,11 +1001,6 @@ async fn perform_session_routing_for_pm_type(
                 )
                 .await?;
 
-                let cached_algorithm = ROUTING_CACHE
-                    .retrieve(&key)
-                    .change_context(errors::RoutingError::CacheMiss)
-                    .attach_printable("unable to retrieve cached routing algorithm")?;
-
                 match cached_algorithm.as_ref() {
                     CachedAlgorithm::Single(conn) => vec![(**conn).clone()],
                     CachedAlgorithm::Priority(plist) => plist.clone(),
@@ -1084,10 +1034,9 @@ async fn perform_session_routing_for_pm_type(
         }
     };
 
-    let mut final_selection = perform_kgraph_filtering(
+    let mut final_selection = perform_cgraph_filtering(
         &session_pm_input.state.clone(),
         session_pm_input.key_store,
-        session_pm_input.merchant_last_modified,
         chosen_connectors,
         session_pm_input.backend_input.clone(),
         None,
@@ -1115,10 +1064,9 @@ async fn perform_session_routing_for_pm_type(
         .await
         .change_context(errors::RoutingError::FallbackConfigFetchFailed)?;
 
-        final_selection = perform_kgraph_filtering(
+        final_selection = perform_cgraph_filtering(
             &session_pm_input.state.clone(),
             session_pm_input.key_store,
-            session_pm_input.merchant_last_modified,
             fallback,
             session_pm_input.backend_input,
             None,
diff --git a/crates/router/src/core/payouts/helpers.rs b/crates/router/src/core/payouts/helpers.rs
index 46c3c8c9aab..dad0739879f 100644
--- a/crates/router/src/core/payouts/helpers.rs
+++ b/crates/router/src/core/payouts/helpers.rs
@@ -669,7 +669,6 @@ pub async fn decide_payout_connector(
             connectors = routing::perform_eligibility_analysis_with_fallback(
                 state,
                 key_store,
-                merchant_account.modified_at.assume_utc().unix_timestamp(),
                 connectors,
                 &TransactionData::<()>::Payout(payout_data),
                 eligible_connectors,
@@ -728,7 +727,6 @@ pub async fn decide_payout_connector(
             connectors = routing::perform_eligibility_analysis_with_fallback(
                 state,
                 key_store,
-                merchant_account.modified_at.assume_utc().unix_timestamp(),
                 connectors,
                 &TransactionData::<()>::Payout(payout_data),
                 eligible_connectors,
diff --git a/crates/router/src/core/routing/helpers.rs b/crates/router/src/core/routing/helpers.rs
index 894e09ab60e..1d46e54bccf 100644
--- a/crates/router/src/core/routing/helpers.rs
+++ b/crates/router/src/core/routing/helpers.rs
@@ -10,10 +10,11 @@ use diesel_models::{
 };
 use error_stack::ResultExt;
 use rustc_hash::FxHashSet;
+use storage_impl::redis::cache as redis_cache;
 
 use crate::{
     core::errors::{self, RouterResult},
-    db::StorageInterface,
+    db::{cache, StorageInterface},
     types::{domain, storage},
     utils::StringExt,
 };
@@ -239,6 +240,17 @@ pub async fn update_business_profile_active_algorithm_ref(
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Failed to convert routing ref to value")?;
 
+    let merchant_id = current_business_profile.merchant_id.clone();
+
+    #[cfg(feature = "business_profile_routing")]
+    let profile_id = current_business_profile.profile_id.clone();
+    #[cfg(feature = "business_profile_routing")]
+    let routing_cache_key = redis_cache::CacheKind::Routing(
+        format!("routing_config_{merchant_id}_{profile_id}").into(),
+    );
+
+    #[cfg(not(feature = "business_profile_routing"))]
+    let routing_cache_key = redis_cache::CacheKind::Routing(format!("dsl_{merchant_id}").into());
     let (routing_algorithm, payout_routing_algorithm) = match transaction_type {
         storage::enums::TransactionType::Payment => (Some(ref_val), None),
         #[cfg(feature = "payouts")]
@@ -272,6 +284,11 @@ pub async fn update_business_profile_active_algorithm_ref(
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Failed to update routing algorithm ref in business profile")?;
+
+    cache::publish_into_redact_channel(db, [routing_cache_key])
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to invalidate routing cache")?;
     Ok(())
 }
 
diff --git a/crates/router/src/db/merchant_account.rs b/crates/router/src/db/merchant_account.rs
index c323ca4abb1..0cecb5e8d76 100644
--- a/crates/router/src/db/merchant_account.rs
+++ b/crates/router/src/db/merchant_account.rs
@@ -513,11 +513,29 @@ async fn publish_and_redact_merchant_account_cache(
         .as_ref()
         .map(|publishable_key| CacheKind::Accounts(publishable_key.into()));
 
+    #[cfg(feature = "business_profile_routing")]
+    let kgraph_key = merchant_account.default_profile.as_ref().map(|profile_id| {
+        CacheKind::CGraph(
+            format!(
+                "kgraph_{}_{}",
+                merchant_account.merchant_id.clone(),
+                profile_id,
+            )
+            .into(),
+        )
+    });
+
+    #[cfg(not(feature = "business_profile_routing"))]
+    let kgraph_key = Some(CacheKind::CGraph(
+        format!("kgraph_{}", merchant_account.merchant_id.clone()).into(),
+    ));
+
     let mut cache_keys = vec![CacheKind::Accounts(
         merchant_account.merchant_id.as_str().into(),
     )];
 
     cache_keys.extend(publishable_key.into_iter());
+    cache_keys.extend(kgraph_key.into_iter());
 
     super::cache::publish_into_redact_channel(store, cache_keys).await?;
     Ok(())
diff --git a/crates/router/src/db/merchant_connector_account.rs b/crates/router/src/db/merchant_connector_account.rs
index 8f50f5f5426..31d965c8cd6 100644
--- a/crates/router/src/db/merchant_connector_account.rs
+++ b/crates/router/src/db/merchant_connector_account.rs
@@ -427,6 +427,9 @@ impl MerchantConnectorAccountInterface for Store {
                     cache::CacheKind::Accounts(
                         format!("{}_{}", _merchant_id, _merchant_connector_id).into(),
                     ),
+                    cache::CacheKind::CGraph(
+                        format!("cgraph_{}_{_profile_id}", _merchant_id).into(),
+                    ),
                 ],
                 update_call,
             )
@@ -475,9 +478,16 @@ impl MerchantConnectorAccountInterface for Store {
                 "profile_id".to_string(),
             ))?;
 
-            super::cache::publish_and_redact(
+            super::cache::publish_and_redact_multiple(
                 self,
-                cache::CacheKind::Accounts(format!("{}_{}", mca.merchant_id, _profile_id).into()),
+                [
+                    cache::CacheKind::Accounts(
+                        format!("{}_{}", mca.merchant_id, _profile_id).into(),
+                    ),
+                    cache::CacheKind::CGraph(
+                        format!("cgraph_{}_{_profile_id}", mca.merchant_id).into(),
+                    ),
+                ],
                 delete_call,
             )
             .await
diff --git a/crates/storage_impl/src/redis/cache.rs b/crates/storage_impl/src/redis/cache.rs
index 4cd2fc0c504..80a6847e376 100644
--- a/crates/storage_impl/src/redis/cache.rs
+++ b/crates/storage_impl/src/redis/cache.rs
@@ -21,6 +21,12 @@ 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 kgraph cache key
+const CGRAPH_CACHE_PREFIX: &str = "cgraph";
+
 /// Prefix for all kinds of cache key
 const ALL_CACHE_PREFIX: &str = "all_cache_kind";
 
@@ -40,6 +46,14 @@ pub static CONFIG_CACHE: Lazy<Cache> = Lazy::new(|| Cache::new(CACHE_TTL, CACHE_
 pub static ACCOUNTS_CACHE: Lazy<Cache> =
     Lazy::new(|| Cache::new(CACHE_TTL, CACHE_TTI, Some(MAX_CAPACITY)));
 
+/// Routing Cache
+pub static ROUTING_CACHE: Lazy<Cache> =
+    Lazy::new(|| Cache::new(CACHE_TTL, CACHE_TTI, Some(MAX_CAPACITY)));
+
+/// CGraph Cache
+pub static CGRAPH_CACHE: Lazy<Cache> =
+    Lazy::new(|| Cache::new(CACHE_TTL, CACHE_TTI, Some(MAX_CAPACITY)));
+
 /// Trait which defines the behaviour of types that's gonna be stored in Cache
 pub trait Cacheable: Any + Send + Sync + DynClone {
     fn as_any(&self) -> &dyn Any;
@@ -48,6 +62,8 @@ pub trait Cacheable: Any + Send + Sync + DynClone {
 pub enum CacheKind<'a> {
     Config(Cow<'a, str>),
     Accounts(Cow<'a, str>),
+    Routing(Cow<'a, str>),
+    CGraph(Cow<'a, str>),
     All(Cow<'a, str>),
 }
 
@@ -56,6 +72,8 @@ impl<'a> From<CacheKind<'a>> for RedisValue {
         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::CGraph(s) => format!("{CGRAPH_CACHE_PREFIX},{s}"),
             CacheKind::All(s) => format!("{ALL_CACHE_PREFIX},{s}"),
         };
         Self::from_string(value)
@@ -73,6 +91,8 @@ impl<'a> TryFrom<RedisValue> for CacheKind<'a> {
         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()))),
+            CGRAPH_CACHE_PREFIX => Ok(Self::CGraph(Cow::Owned(split.1.to_string()))),
             ALL_CACHE_PREFIX => Ok(Self::All(Cow::Owned(split.1.to_string()))),
             _ => Err(validation_err.into()),
         }
@@ -123,6 +143,11 @@ impl Cache {
         (*val).as_any().downcast_ref::<T>().cloned()
     }
 
+    /// Check if a key exists in cache
+    pub async fn exists(&self, key: &str) -> bool {
+        self.inner.contains_key(key)
+    }
+
     pub async fn remove(&self, key: &str) {
         self.inner.invalidate(key).await;
     }
diff --git a/crates/storage_impl/src/redis/pub_sub.rs b/crates/storage_impl/src/redis/pub_sub.rs
index 2922dbcadba..9669e826663 100644
--- a/crates/storage_impl/src/redis/pub_sub.rs
+++ b/crates/storage_impl/src/redis/pub_sub.rs
@@ -2,7 +2,7 @@ use error_stack::ResultExt;
 use redis_interface::{errors as redis_errors, PubsubInterface, RedisValue};
 use router_env::logger;
 
-use crate::redis::cache::{CacheKind, ACCOUNTS_CACHE, CONFIG_CACHE};
+use crate::redis::cache::{CacheKind, ACCOUNTS_CACHE, CGRAPH_CACHE, CONFIG_CACHE, ROUTING_CACHE};
 
 #[async_trait::async_trait]
 pub trait PubSubInterface {
@@ -67,9 +67,19 @@ impl PubSubInterface for redis_interface::RedisConnectionPool {
                     ACCOUNTS_CACHE.remove(key.as_ref()).await;
                     key
                 }
+                CacheKind::CGraph(key) => {
+                    CGRAPH_CACHE.remove(key.as_ref()).await;
+                    key
+                }
+                CacheKind::Routing(key) => {
+                    ROUTING_CACHE.remove(key.as_ref()).await;
+                    key
+                }
                 CacheKind::All(key) => {
                     CONFIG_CACHE.remove(key.as_ref()).await;
                     ACCOUNTS_CACHE.remove(key.as_ref()).await;
+                    CGRAPH_CACHE.remove(key.as_ref()).await;
+                    ROUTING_CACHE.remove(key.as_ref()).await;
                     key
                 }
             };

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

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
Enabled Moka cache for routing with cache invalidation


### 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)?
-->
Look for the pub-sub cache invalidation log
![image](https://github.com/juspay/hyperswitch/assets/76486416/e739ac71-2b2b-4103-9804-b4478c5d0e87)
![image](https://github.com/juspay/hyperswitch/assets/76486416/068743c0-a9d0-49af-859d-f23ab32eb2ec)

1. Update an existing Merchant connector account and look for Cgrpah cache invalidation log
```
curl --location --request POST 'http://localhost:8080/account/sarthak/connectors/mca_VaZpO5COiFDUIiMq7DFV' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: test_admin' \
--data-raw '{
    "connector_type": "fiz_operations",
    "metadata": {
        "city": "NY",
        "unit": "248"
    }
}'
```
2. Activate a routing config from dashboard, then search for routing cache invalidation using routing config 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`
- [ ] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
- [ ] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable
