refactor(access_token): use merchant_connector_id for storing access token (#4462)

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}")
+}

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;

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
+                    if let Err(access_token_set_error) = store
-                    // the access token, the next request will create new access token
-                    let _ = 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

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

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
+        let access_token = maybe_token
-            .map(|token| token.parse_struct("AccessToken"))
+            .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(())