Bug: [REFACTOR]: use `merchant_connector_id` for storing `access_token`



The access tokens are currently stored in the format of `access_token_{merchant_id}_{connector_name}`. This cannot support multiple connector accounts.

In order to store access tokens for multiple connector accounts in a single business profile or a merchant account, the new format in which this should be stored is

`access_token_{merchant_id}_{merchant_connector_id}`

But there are cases when `merchant_connector_id` may not be present
- In cases where straight through routing algorithm is used
- In cases where the credentials are supplied directly using the creds identifier

In order to support these use cases, a fallback should be provided. 
diff --git a/crates/common_utils/src/access_token.rs b/crates/common_utils/src/access_token.rs
new file mode 100644
index 00000000000..20bac0dafbb
--- /dev/null
+++ b/crates/common_utils/src/access_token.rs
@@ -0,0 +1,11 @@
+//! Commonly used utilities for access token
+
+use std::fmt::Display;
+
+/// Create a key for fetching the access token from redis
+pub fn create_access_token_key(
+    merchant_id: impl Display,
+    merchant_connector_id: impl Display,
+) -> String {
+    format!("access_token_{merchant_id}_{merchant_connector_id}")
+}
diff --git a/crates/common_utils/src/lib.rs b/crates/common_utils/src/lib.rs
index e30d596d9a7..f24b67fde1b 100644
--- a/crates/common_utils/src/lib.rs
+++ b/crates/common_utils/src/lib.rs
@@ -2,6 +2,7 @@
 #![warn(missing_docs, missing_debug_implementations)]
 #![doc = include_str!(concat!(env!("CARGO_MANIFEST_DIR" ), "/", "README.md"))]
 
+pub mod access_token;
 pub mod consts;
 pub mod crypto;
 pub mod custom_serde;
diff --git a/crates/router/src/core/payments/access_token.rs b/crates/router/src/core/payments/access_token.rs
index 47f6b7e2ae4..49d27d81509 100644
--- a/crates/router/src/core/payments/access_token.rs
+++ b/crates/router/src/core/payments/access_token.rs
@@ -10,7 +10,7 @@ use crate::{
         payments,
     },
     routes::{metrics, AppState},
-    services,
+    services::{self, logger},
     types::{self, api as api_types, domain},
 };
 
@@ -64,8 +64,14 @@ pub async fn add_access_token<
     {
         let merchant_id = &merchant_account.merchant_id;
         let store = &*state.store;
+        let merchant_connector_id = connector
+            .merchant_connector_id
+            .as_ref()
+            .ok_or(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Missing merchant_connector_id in ConnectorData")?;
+
         let old_access_token = store
-            .get_access_token(merchant_id, connector.connector.id())
+            .get_access_token(merchant_id, merchant_connector_id)
             .await
             .change_context(errors::ApiErrorResponse::InternalServerError)
             .attach_printable("DB error when accessing the access token")?;
@@ -103,19 +109,24 @@ pub async fn add_access_token<
                 )
                 .await?
                 .async_map(|access_token| async {
-                    //Store the access token in db
+                    // Store the access token in redis with expiry
+                    // The expiry should be adjusted for network delays from the connector
                     let store = &*state.store;
-                    // This error should not be propagated, we don't want payments to fail once we have
-                    // the access token, the next request will create new access token
-                    let _ = store
+                    if let Err(access_token_set_error) = store
                         .set_access_token(
                             merchant_id,
-                            connector.connector.id(),
+                            merchant_connector_id.as_str(),
                             access_token.clone(),
                         )
                         .await
                         .change_context(errors::ApiErrorResponse::InternalServerError)
-                        .attach_printable("DB error when setting the access token");
+                        .attach_printable("DB error when setting the access token")
+                    {
+                        // If we are not able to set the access token in redis, the error should just be logged and proceed with the payment
+                        // Payments should not fail, once the access token is successfully created
+                        // The next request will create new access token, if required
+                        logger::error!(access_token_set_error=?access_token_set_error);
+                    }
                     Some(access_token)
                 })
                 .await
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 9787e366e79..a4e73f7d2c6 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -859,21 +859,21 @@ impl ConnectorAccessToken for KafkaStore {
     async fn get_access_token(
         &self,
         merchant_id: &str,
-        connector_name: &str,
+        merchant_connector_id: &str,
     ) -> CustomResult<Option<AccessToken>, errors::StorageError> {
         self.diesel_store
-            .get_access_token(merchant_id, connector_name)
+            .get_access_token(merchant_id, merchant_connector_id)
             .await
     }
 
     async fn set_access_token(
         &self,
         merchant_id: &str,
-        connector_name: &str,
+        merchant_connector_id: &str,
         access_token: AccessToken,
     ) -> CustomResult<(), errors::StorageError> {
         self.diesel_store
-            .set_access_token(merchant_id, connector_name, access_token)
+            .set_access_token(merchant_id, merchant_connector_id, access_token)
             .await
     }
 }
diff --git a/crates/router/src/db/merchant_connector_account.rs b/crates/router/src/db/merchant_connector_account.rs
index b7c85560804..06d1e505c6b 100644
--- a/crates/router/src/db/merchant_connector_account.rs
+++ b/crates/router/src/db/merchant_connector_account.rs
@@ -24,13 +24,13 @@ pub trait ConnectorAccessToken {
     async fn get_access_token(
         &self,
         merchant_id: &str,
-        connector_name: &str,
+        merchant_connector_id: &str,
     ) -> CustomResult<Option<types::AccessToken>, errors::StorageError>;
 
     async fn set_access_token(
         &self,
         merchant_id: &str,
-        connector_name: &str,
+        merchant_connector_id: &str,
         access_token: types::AccessToken,
     ) -> CustomResult<(), errors::StorageError>;
 }
@@ -41,12 +41,14 @@ impl ConnectorAccessToken for Store {
     async fn get_access_token(
         &self,
         merchant_id: &str,
-        connector_name: &str,
+        merchant_connector_id: &str,
     ) -> CustomResult<Option<types::AccessToken>, errors::StorageError> {
         //TODO: Handle race condition
         // This function should acquire a global lock on some resource, if access token is already
         // being refreshed by other request then wait till it finishes and use the same access token
-        let key = format!("access_token_{merchant_id}_{connector_name}");
+        let key =
+            common_utils::access_token::create_access_token_key(merchant_id, merchant_connector_id);
+
         let maybe_token = self
             .get_redis_conn()
             .map_err(Into::<errors::StorageError>::into)?
@@ -55,10 +57,9 @@ impl ConnectorAccessToken for Store {
             .change_context(errors::StorageError::KVError)
             .attach_printable("DB error when getting access token")?;
 
-        let access_token: Option<types::AccessToken> = maybe_token
-            .map(|token| token.parse_struct("AccessToken"))
+        let access_token = maybe_token
+            .map(|token| token.parse_struct::<types::AccessToken>("AccessToken"))
             .transpose()
-            .change_context(errors::ParsingError::UnknownError)
             .change_context(errors::StorageError::DeserializationFailed)?;
 
         Ok(access_token)
@@ -68,10 +69,11 @@ impl ConnectorAccessToken for Store {
     async fn set_access_token(
         &self,
         merchant_id: &str,
-        connector_name: &str,
+        merchant_connector_id: &str,
         access_token: types::AccessToken,
     ) -> CustomResult<(), errors::StorageError> {
-        let key = format!("access_token_{merchant_id}_{connector_name}");
+        let key =
+            common_utils::access_token::create_access_token_key(merchant_id, merchant_connector_id);
         let serialized_access_token = access_token
             .encode_to_string_of_json()
             .change_context(errors::StorageError::SerializationFailed)?;
@@ -88,7 +90,7 @@ impl ConnectorAccessToken for MockDb {
     async fn get_access_token(
         &self,
         _merchant_id: &str,
-        _connector_name: &str,
+        _merchant_connector_id: &str,
     ) -> CustomResult<Option<types::AccessToken>, errors::StorageError> {
         Ok(None)
     }
@@ -96,7 +98,7 @@ impl ConnectorAccessToken for MockDb {
     async fn set_access_token(
         &self,
         _merchant_id: &str,
-        _connector_name: &str,
+        _merchant_connector_id: &str,
         _access_token: types::AccessToken,
     ) -> CustomResult<(), errors::StorageError> {
         Ok(())

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

- [x] Refactoring

## Description
<!-- Describe your changes in detail -->
This PR modifies the way in which access tokens are stored in the redis. The key format is changed to 
`access_token_{merchant_id}_{merchant_connector_id}` More information can be found in the linked issue.

<!--
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).
-->
In order to support multiple connector accounts

## 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)?
-->
- Run the Trustpay postman collection locally and check whether the access token is stored in redis
<img width="2056" alt="Screenshot 2024-04-25 at 1 56 18 PM" src="https://github.com/juspay/hyperswitch/assets/48803246/f527d6d4-4af9-49cf-8a3f-bf2fa6952d2c">

## Impact
- All connectors using access token auth.
- Trustpay BankRedirect, Volt and Iatapay

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