Bug: [FIX] batch encrypt decrypt for mca fetch/insert/update



This adds batch encryption/decryption support for fetch/insert/update of the Merchant Connector Account.
diff --git a/crates/hyperswitch_domain_models/src/merchant_connector_account.rs b/crates/hyperswitch_domain_models/src/merchant_connector_account.rs
index a281c553a13..66b98389818 100644
--- a/crates/hyperswitch_domain_models/src/merchant_connector_account.rs
+++ b/crates/hyperswitch_domain_models/src/merchant_connector_account.rs
@@ -4,14 +4,15 @@ use common_utils::{
     encryption::Encryption,
     errors::{CustomResult, ValidationError},
     id_type, pii, type_name,
-    types::keymanager::{Identifier, KeyManagerState},
+    types::keymanager::{Identifier, KeyManagerState, ToEncryptable},
 };
 use diesel_models::{enums, merchant_connector_account::MerchantConnectorAccountUpdateInternal};
 use error_stack::ResultExt;
 use masking::{PeekInterface, Secret};
+use rustc_hash::FxHashMap;
 
 use super::behaviour;
-use crate::type_encryption::{crypto_operation, AsyncLift, CryptoOperation};
+use crate::type_encryption::{crypto_operation, CryptoOperation};
 
 #[cfg(feature = "v1")]
 #[derive(Clone, Debug)]
@@ -175,21 +176,33 @@ impl behaviour::Conversion for MerchantConnectorAccount {
         _key_manager_identifier: Identifier,
     ) -> CustomResult<Self, ValidationError> {
         let identifier = Identifier::Merchant(other.merchant_id.clone());
+        let decrypted_data = crypto_operation(
+            state,
+            type_name!(Self::DstType),
+            CryptoOperation::BatchDecrypt(EncryptedMca::to_encryptable(EncryptedMca {
+                connector_account_details: other.connector_account_details,
+                additional_merchant_data: other.additional_merchant_data,
+                connector_wallets_details: other.connector_wallets_details,
+            })),
+            identifier.clone(),
+            key.peek(),
+        )
+        .await
+        .and_then(|val| val.try_into_batchoperation())
+        .change_context(ValidationError::InvalidValue {
+            message: "Failed while decrypting connector account details".to_string(),
+        })?;
+
+        let decrypted_data = EncryptedMca::from_encryptable(decrypted_data).change_context(
+            ValidationError::InvalidValue {
+                message: "Failed while decrypting connector account details".to_string(),
+            },
+        )?;
+
         Ok(Self {
             merchant_id: other.merchant_id,
             connector_name: other.connector_name,
-            connector_account_details: crypto_operation(
-                state,
-                type_name!(Self::DstType),
-                CryptoOperation::Decrypt(other.connector_account_details),
-                identifier.clone(),
-                key.peek(),
-            )
-            .await
-            .and_then(|val| val.try_into_operation())
-            .change_context(ValidationError::InvalidValue {
-                message: "Failed while decrypting connector account details".to_string(),
-            })?,
+            connector_account_details: decrypted_data.connector_account_details,
             test_mode: other.test_mode,
             disabled: other.disabled,
             merchant_connector_id: other.merchant_connector_id,
@@ -213,41 +226,8 @@ impl behaviour::Conversion for MerchantConnectorAccount {
             applepay_verified_domains: other.applepay_verified_domains,
             pm_auth_config: other.pm_auth_config,
             status: other.status,
-            connector_wallets_details: other
-                .connector_wallets_details
-                .async_lift(|inner| async {
-                    crypto_operation(
-                        state,
-                        type_name!(Self::DstType),
-                        CryptoOperation::DecryptOptional(inner),
-                        identifier.clone(),
-                        key.peek(),
-                    )
-                    .await
-                    .and_then(|val| val.try_into_optionaloperation())
-                })
-                .await
-                .change_context(ValidationError::InvalidValue {
-                    message: "Failed while decrypting connector wallets details".to_string(),
-                })?,
-            additional_merchant_data: if let Some(data) = other.additional_merchant_data {
-                Some(
-                    crypto_operation(
-                        state,
-                        type_name!(Self::DstType),
-                        CryptoOperation::Decrypt(data),
-                        identifier,
-                        key.peek(),
-                    )
-                    .await
-                    .and_then(|val| val.try_into_operation())
-                    .change_context(ValidationError::InvalidValue {
-                        message: "Failed while decrypting additional_merchant_data".to_string(),
-                    })?,
-                )
-            } else {
-                None
-            },
+            connector_wallets_details: decrypted_data.connector_wallets_details,
+            additional_merchant_data: decrypted_data.additional_merchant_data,
             version: other.version,
         })
     }
@@ -324,22 +304,35 @@ impl behaviour::Conversion for MerchantConnectorAccount {
         _key_manager_identifier: Identifier,
     ) -> CustomResult<Self, ValidationError> {
         let identifier = Identifier::Merchant(other.merchant_id.clone());
+
+        let decrypted_data = crypto_operation(
+            state,
+            type_name!(Self::DstType),
+            CryptoOperation::BatchDecrypt(EncryptedMca::to_encryptable(EncryptedMca {
+                connector_account_details: other.connector_account_details,
+                additional_merchant_data: other.additional_merchant_data,
+                connector_wallets_details: other.connector_wallets_details,
+            })),
+            identifier.clone(),
+            key.peek(),
+        )
+        .await
+        .and_then(|val| val.try_into_batchoperation())
+        .change_context(ValidationError::InvalidValue {
+            message: "Failed while decrypting connector account details".to_string(),
+        })?;
+
+        let decrypted_data = EncryptedMca::from_encryptable(decrypted_data).change_context(
+            ValidationError::InvalidValue {
+                message: "Failed while decrypting connector account details".to_string(),
+            },
+        )?;
+
         Ok(Self {
             id: other.id,
             merchant_id: other.merchant_id,
             connector_name: other.connector_name,
-            connector_account_details: crypto_operation(
-                state,
-                type_name!(Self::DstType),
-                CryptoOperation::Decrypt(other.connector_account_details),
-                identifier.clone(),
-                key.peek(),
-            )
-            .await
-            .and_then(|val| val.try_into_operation())
-            .change_context(ValidationError::InvalidValue {
-                message: "Failed while decrypting connector account details".to_string(),
-            })?,
+            connector_account_details: decrypted_data.connector_account_details,
             disabled: other.disabled,
             payment_methods_enabled: other.payment_methods_enabled,
             connector_type: other.connector_type,
@@ -354,41 +347,8 @@ impl behaviour::Conversion for MerchantConnectorAccount {
             applepay_verified_domains: other.applepay_verified_domains,
             pm_auth_config: other.pm_auth_config,
             status: other.status,
-            connector_wallets_details: other
-                .connector_wallets_details
-                .async_lift(|inner| async {
-                    crypto_operation(
-                        state,
-                        type_name!(Self::DstType),
-                        CryptoOperation::DecryptOptional(inner),
-                        identifier.clone(),
-                        key.peek(),
-                    )
-                    .await
-                    .and_then(|val| val.try_into_optionaloperation())
-                })
-                .await
-                .change_context(ValidationError::InvalidValue {
-                    message: "Failed while decrypting connector wallets details".to_string(),
-                })?,
-            additional_merchant_data: if let Some(data) = other.additional_merchant_data {
-                Some(
-                    crypto_operation(
-                        state,
-                        type_name!(Self::DstType),
-                        CryptoOperation::Decrypt(data),
-                        identifier,
-                        key.peek(),
-                    )
-                    .await
-                    .and_then(|val| val.try_into_operation())
-                    .change_context(ValidationError::InvalidValue {
-                        message: "Failed while decrypting additional_merchant_data".to_string(),
-                    })?,
-                )
-            } else {
-                None
-            },
+            connector_wallets_details: decrypted_data.connector_wallets_details,
+            additional_merchant_data: decrypted_data.additional_merchant_data,
             version: other.version,
         })
     }
@@ -542,3 +502,121 @@ impl From<MerchantConnectorAccountUpdate> for MerchantConnectorAccountUpdateInte
         }
     }
 }
+
+pub struct McaFromRequestfromUpdate {
+    pub connector_account_details: Option<pii::SecretSerdeValue>,
+    pub connector_wallets_details: Option<pii::SecretSerdeValue>,
+    pub additional_merchant_data: Option<pii::SecretSerdeValue>,
+}
+pub struct McaFromRequest {
+    pub connector_account_details: pii::SecretSerdeValue,
+    pub connector_wallets_details: Option<pii::SecretSerdeValue>,
+    pub additional_merchant_data: Option<pii::SecretSerdeValue>,
+}
+
+pub struct DecryptedMca {
+    pub connector_account_details: Encryptable<pii::SecretSerdeValue>,
+    pub connector_wallets_details: Option<Encryptable<pii::SecretSerdeValue>>,
+    pub additional_merchant_data: Option<Encryptable<pii::SecretSerdeValue>>,
+}
+
+pub struct EncryptedMca {
+    pub connector_account_details: Encryption,
+    pub connector_wallets_details: Option<Encryption>,
+    pub additional_merchant_data: Option<Encryption>,
+}
+
+pub struct DecryptedUpdateMca {
+    pub connector_account_details: Option<Encryptable<pii::SecretSerdeValue>>,
+    pub connector_wallets_details: Option<Encryptable<pii::SecretSerdeValue>>,
+    pub additional_merchant_data: Option<Encryptable<pii::SecretSerdeValue>>,
+}
+
+impl ToEncryptable<DecryptedMca, Secret<serde_json::Value>, Encryption> for EncryptedMca {
+    fn from_encryptable(
+        mut hashmap: FxHashMap<String, Encryptable<Secret<serde_json::Value>>>,
+    ) -> CustomResult<DecryptedMca, common_utils::errors::ParsingError> {
+        Ok(DecryptedMca {
+            connector_account_details: hashmap.remove("connector_account_details").ok_or(
+                error_stack::report!(common_utils::errors::ParsingError::EncodeError(
+                    "Unable to convert from HashMap to DecryptedMca",
+                )),
+            )?,
+            connector_wallets_details: hashmap.remove("connector_wallets_details"),
+            additional_merchant_data: hashmap.remove("additional_merchant_data"),
+        })
+    }
+
+    fn to_encryptable(self) -> FxHashMap<String, Encryption> {
+        let mut map = FxHashMap::with_capacity_and_hasher(3, Default::default());
+
+        map.insert(
+            "connector_account_details".to_string(),
+            self.connector_account_details,
+        );
+        self.connector_wallets_details
+            .map(|s| map.insert("connector_wallets_details".to_string(), s));
+        self.additional_merchant_data
+            .map(|s| map.insert("additional_merchant_data".to_string(), s));
+        map
+    }
+}
+
+impl ToEncryptable<DecryptedUpdateMca, Secret<serde_json::Value>, Secret<serde_json::Value>>
+    for McaFromRequestfromUpdate
+{
+    fn from_encryptable(
+        mut hashmap: FxHashMap<String, Encryptable<Secret<serde_json::Value>>>,
+    ) -> CustomResult<DecryptedUpdateMca, common_utils::errors::ParsingError> {
+        Ok(DecryptedUpdateMca {
+            connector_account_details: hashmap.remove("connector_account_details"),
+            connector_wallets_details: hashmap.remove("connector_wallets_details"),
+            additional_merchant_data: hashmap.remove("additional_merchant_data"),
+        })
+    }
+
+    fn to_encryptable(self) -> FxHashMap<String, Secret<serde_json::Value>> {
+        let mut map = FxHashMap::with_capacity_and_hasher(3, Default::default());
+
+        self.connector_account_details
+            .map(|cad| map.insert("connector_account_details".to_string(), cad));
+
+        self.connector_wallets_details
+            .map(|s| map.insert("connector_wallets_details".to_string(), s));
+        self.additional_merchant_data
+            .map(|s| map.insert("additional_merchant_data".to_string(), s));
+        map
+    }
+}
+
+impl ToEncryptable<DecryptedMca, Secret<serde_json::Value>, Secret<serde_json::Value>>
+    for McaFromRequest
+{
+    fn from_encryptable(
+        mut hashmap: FxHashMap<String, Encryptable<Secret<serde_json::Value>>>,
+    ) -> CustomResult<DecryptedMca, common_utils::errors::ParsingError> {
+        Ok(DecryptedMca {
+            connector_account_details: hashmap.remove("connector_account_details").ok_or(
+                error_stack::report!(common_utils::errors::ParsingError::EncodeError(
+                    "Unable to convert from HashMap to DecryptedMca",
+                )),
+            )?,
+            connector_wallets_details: hashmap.remove("connector_wallets_details"),
+            additional_merchant_data: hashmap.remove("additional_merchant_data"),
+        })
+    }
+
+    fn to_encryptable(self) -> FxHashMap<String, Secret<serde_json::Value>> {
+        let mut map = FxHashMap::with_capacity_and_hasher(3, Default::default());
+
+        map.insert(
+            "connector_account_details".to_string(),
+            self.connector_account_details,
+        );
+        self.connector_wallets_details
+            .map(|s| map.insert("connector_wallets_details".to_string(), s));
+        self.additional_merchant_data
+            .map(|s| map.insert("additional_merchant_data".to_string(), s));
+        map
+    }
+}
diff --git a/crates/router/src/core/admin.rs b/crates/router/src/core/admin.rs
index b988578d35e..fed284955b3 100644
--- a/crates/router/src/core/admin.rs
+++ b/crates/router/src/core/admin.rs
@@ -8,12 +8,15 @@ use common_utils::{
     date_time,
     ext_traits::{AsyncExt, Encode, OptionExt, ValueExt},
     id_type, pii, type_name,
-    types::keymanager::{self as km_types, KeyManagerState},
+    types::keymanager::{self as km_types, KeyManagerState, ToEncryptable},
 };
 use diesel_models::configs;
 #[cfg(all(any(feature = "v1", feature = "v2"), feature = "olap"))]
 use diesel_models::organization::OrganizationBridge;
 use error_stack::{report, FutureExt, ResultExt};
+use hyperswitch_domain_models::merchant_connector_account::{
+    McaFromRequest, McaFromRequestfromUpdate,
+};
 use masking::{ExposeInterface, PeekInterface, Secret};
 use pm_auth::{connector::plaid::transformers::PlaidAuthType, types as pm_auth_types};
 use regex::Regex;
@@ -2089,25 +2092,37 @@ impl MerchantConnectorAccountUpdateBridge for api_models::admin::MerchantConnect
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Failed to serialize MerchantRecipientData")?;
 
+        let encrypted_data = domain_types::crypto_operation(
+            key_manager_state,
+            type_name!(domain::MerchantConnectorAccount),
+            domain_types::CryptoOperation::BatchEncrypt(McaFromRequestfromUpdate::to_encryptable(
+                McaFromRequestfromUpdate {
+                    connector_account_details: self.connector_account_details,
+                    connector_wallets_details:
+                        helpers::get_connector_wallets_details_with_apple_pay_certificates(
+                            &self.metadata,
+                            &self.connector_wallets_details,
+                        )
+                        .await?,
+                    additional_merchant_data: merchant_recipient_data.map(Secret::new),
+                },
+            )),
+            km_types::Identifier::Merchant(key_store.merchant_id.clone()),
+            key_store.key.peek(),
+        )
+        .await
+        .and_then(|val| val.try_into_batchoperation())
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed while decrypting connector account details".to_string())?;
+
+        let encrypted_data = McaFromRequestfromUpdate::from_encryptable(encrypted_data)
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed while decrypting connector account details")?;
+
         Ok(storage::MerchantConnectorAccountUpdate::Update {
             connector_type: Some(self.connector_type),
             connector_label: self.connector_label.clone(),
-            connector_account_details: self
-                .connector_account_details
-                .async_lift(|inner| async {
-                    domain_types::crypto_operation(
-                        key_manager_state,
-                        type_name!(storage::MerchantConnectorAccount),
-                        domain_types::CryptoOperation::EncryptOptional(inner),
-                        km_types::Identifier::Merchant(key_store.merchant_id.clone()),
-                        key_store.key.get_inner().peek(),
-                    )
-                    .await
-                    .and_then(|val| val.try_into_optionaloperation())
-                })
-                .await
-                .change_context(errors::ApiErrorResponse::InternalServerError)
-                .attach_printable("Failed while encrypting data")?,
+            connector_account_details: encrypted_data.connector_account_details,
             disabled,
             payment_methods_enabled,
             metadata: self.metadata,
@@ -2123,31 +2138,8 @@ impl MerchantConnectorAccountUpdateBridge for api_models::admin::MerchantConnect
             applepay_verified_domains: None,
             pm_auth_config: self.pm_auth_config,
             status: Some(connector_status),
-            additional_merchant_data: if let Some(mcd) = merchant_recipient_data {
-                Some(
-                    domain_types::crypto_operation(
-                        key_manager_state,
-                        type_name!(domain::MerchantConnectorAccount),
-                        domain_types::CryptoOperation::Encrypt(Secret::new(mcd)),
-                        km_types::Identifier::Merchant(key_store.merchant_id.clone()),
-                        key_store.key.peek(),
-                    )
-                    .await
-                    .and_then(|val| val.try_into_operation())
-                    .change_context(errors::ApiErrorResponse::InternalServerError)
-                    .attach_printable("Unable to encrypt additional_merchant_data")?,
-                )
-            } else {
-                None
-            },
-            connector_wallets_details:
-                helpers::get_encrypted_connector_wallets_details_with_apple_pay_certificates(
-                    state,
-                    &key_store,
-                    &metadata,
-                    &self.connector_wallets_details,
-                )
-                .await?,
+            additional_merchant_data: encrypted_data.additional_merchant_data,
+            connector_wallets_details: encrypted_data.connector_wallets_details,
         })
     }
 }
@@ -2267,27 +2259,39 @@ impl MerchantConnectorAccountUpdateBridge for api_models::admin::MerchantConnect
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Failed to serialize MerchantRecipientData")?;
 
+        let encrypted_data = domain_types::crypto_operation(
+            key_manager_state,
+            type_name!(domain::MerchantConnectorAccount),
+            domain_types::CryptoOperation::BatchEncrypt(McaFromRequestfromUpdate::to_encryptable(
+                McaFromRequestfromUpdate {
+                    connector_account_details: self.connector_account_details,
+                    connector_wallets_details:
+                        helpers::get_connector_wallets_details_with_apple_pay_certificates(
+                            &self.metadata,
+                            &self.connector_wallets_details,
+                        )
+                        .await?,
+                    additional_merchant_data: merchant_recipient_data.map(Secret::new),
+                },
+            )),
+            km_types::Identifier::Merchant(key_store.merchant_id.clone()),
+            key_store.key.peek(),
+        )
+        .await
+        .and_then(|val| val.try_into_batchoperation())
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed while decrypting connector account details".to_string())?;
+
+        let encrypted_data = McaFromRequestfromUpdate::from_encryptable(encrypted_data)
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed while decrypting connector account details")?;
+
         Ok(storage::MerchantConnectorAccountUpdate::Update {
             connector_type: Some(self.connector_type),
             connector_name: None,
             merchant_connector_id: None,
             connector_label: self.connector_label.clone(),
-            connector_account_details: self
-                .connector_account_details
-                .async_lift(|inner| async {
-                    domain_types::crypto_operation(
-                        key_manager_state,
-                        type_name!(storage::MerchantConnectorAccount),
-                        domain_types::CryptoOperation::EncryptOptional(inner),
-                        km_types::Identifier::Merchant(key_store.merchant_id.clone()),
-                        key_store.key.get_inner().peek(),
-                    )
-                    .await
-                    .and_then(|val| val.try_into_optionaloperation())
-                })
-                .await
-                .change_context(errors::ApiErrorResponse::InternalServerError)
-                .attach_printable("Failed while encrypting data")?,
+            connector_account_details: encrypted_data.connector_account_details,
             test_mode: self.test_mode,
             disabled,
             payment_methods_enabled,
@@ -2304,31 +2308,8 @@ impl MerchantConnectorAccountUpdateBridge for api_models::admin::MerchantConnect
             applepay_verified_domains: None,
             pm_auth_config: self.pm_auth_config,
             status: Some(connector_status),
-            additional_merchant_data: if let Some(mcd) = merchant_recipient_data {
-                Some(
-                    domain_types::crypto_operation(
-                        key_manager_state,
-                        type_name!(domain::MerchantConnectorAccount),
-                        domain_types::CryptoOperation::Encrypt(Secret::new(mcd)),
-                        km_types::Identifier::Merchant(key_store.merchant_id.clone()),
-                        key_store.key.peek(),
-                    )
-                    .await
-                    .and_then(|val| val.try_into_operation())
-                    .change_context(errors::ApiErrorResponse::InternalServerError)
-                    .attach_printable("Unable to encrypt additional_merchant_data")?,
-                )
-            } else {
-                None
-            },
-            connector_wallets_details:
-                helpers::get_encrypted_connector_wallets_details_with_apple_pay_certificates(
-                    state,
-                    &key_store,
-                    &metadata,
-                    &self.connector_wallets_details,
-                )
-                .await?,
+            additional_merchant_data: encrypted_data.additional_merchant_data,
+            connector_wallets_details: encrypted_data.connector_wallets_details,
         })
     }
 }
@@ -2417,25 +2398,43 @@ impl MerchantConnectorAccountCreateBridge for api::MerchantConnectorCreate {
         .transpose()
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Failed to serialize MerchantRecipientData")?;
+
+        let encrypted_data = domain_types::crypto_operation(
+            key_manager_state,
+            type_name!(domain::MerchantConnectorAccount),
+            domain_types::CryptoOperation::BatchEncrypt(McaFromRequest::to_encryptable(
+                McaFromRequest {
+                    connector_account_details: self.connector_account_details.ok_or(
+                        errors::ApiErrorResponse::MissingRequiredField {
+                            field_name: "connector_account_details",
+                        },
+                    )?,
+                    connector_wallets_details:
+                        helpers::get_connector_wallets_details_with_apple_pay_certificates(
+                            &self.metadata,
+                            &self.connector_wallets_details,
+                        )
+                        .await?,
+                    additional_merchant_data: merchant_recipient_data.map(Secret::new),
+                },
+            )),
+            identifier.clone(),
+            key_store.key.peek(),
+        )
+        .await
+        .and_then(|val| val.try_into_batchoperation())
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed while decrypting connector account details".to_string())?;
+
+        let encrypted_data = McaFromRequest::from_encryptable(encrypted_data)
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed while decrypting connector account details")?;
+
         Ok(domain::MerchantConnectorAccount {
             merchant_id: business_profile.merchant_id.clone(),
             connector_type: self.connector_type,
             connector_name: self.connector_name.to_string(),
-            connector_account_details: domain_types::crypto_operation(
-                key_manager_state,
-                type_name!(domain::MerchantConnectorAccount),
-                domain_types::CryptoOperation::Encrypt(self.connector_account_details.ok_or(
-                    errors::ApiErrorResponse::MissingRequiredField {
-                        field_name: "connector_account_details",
-                    },
-                )?),
-                identifier.clone(),
-                key_store.key.peek(),
-            )
-            .await
-            .and_then(|val| val.try_into_operation())
-            .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("Unable to encrypt connector account details")?,
+            connector_account_details: encrypted_data.connector_account_details,
             payment_methods_enabled,
             disabled,
             metadata: self.metadata.clone(),
@@ -2459,22 +2458,8 @@ impl MerchantConnectorAccountCreateBridge for api::MerchantConnectorCreate {
             applepay_verified_domains: None,
             pm_auth_config: self.pm_auth_config.clone(),
             status: connector_status,
-            connector_wallets_details: helpers::get_encrypted_connector_wallets_details_with_apple_pay_certificates(state, &key_store, &self.metadata, &self.connector_wallets_details).await?,
-            additional_merchant_data: if let Some(mcd) =  merchant_recipient_data {
-                Some(domain_types::crypto_operation(
-                    key_manager_state,
-                    type_name!(domain::MerchantConnectorAccount),
-                    domain_types::CryptoOperation::Encrypt(Secret::new(mcd)),
-                    km_types::Identifier::Merchant(key_store.merchant_id.clone()),
-                    key_store.key.peek(),
-                )
-                .await
-                .and_then(|val| val.try_into_operation())
-                .change_context(errors::ApiErrorResponse::InternalServerError)
-                .attach_printable("Unable to encrypt additional_merchant_data")?)
-            } else {
-                None
-            },
+            connector_wallets_details: encrypted_data.connector_wallets_details,
+            additional_merchant_data: encrypted_data.additional_merchant_data,
             version: hyperswitch_domain_models::consts::API_VERSION,
         })
     }
@@ -2582,26 +2567,44 @@ impl MerchantConnectorAccountCreateBridge for api::MerchantConnectorCreate {
         .transpose()
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Failed to serialize MerchantRecipientData")?;
+
+        let encrypted_data = domain_types::crypto_operation(
+            key_manager_state,
+            type_name!(domain::MerchantConnectorAccount),
+            domain_types::CryptoOperation::BatchEncrypt(McaFromRequest::to_encryptable(
+                McaFromRequest {
+                    connector_account_details: self.connector_account_details.ok_or(
+                        errors::ApiErrorResponse::MissingRequiredField {
+                            field_name: "connector_account_details",
+                        },
+                    )?,
+                    connector_wallets_details:
+                        helpers::get_connector_wallets_details_with_apple_pay_certificates(
+                            &self.metadata,
+                            &self.connector_wallets_details,
+                        )
+                        .await?,
+                    additional_merchant_data: merchant_recipient_data.map(Secret::new),
+                },
+            )),
+            identifier.clone(),
+            key_store.key.peek(),
+        )
+        .await
+        .and_then(|val| val.try_into_batchoperation())
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed while decrypting connector account details".to_string())?;
+
+        let encrypted_data = McaFromRequest::from_encryptable(encrypted_data)
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed while decrypting connector account details")?;
+
         Ok(domain::MerchantConnectorAccount {
             merchant_id: business_profile.merchant_id.clone(),
             connector_type: self.connector_type,
             connector_name: self.connector_name.to_string(),
             merchant_connector_id: common_utils::generate_merchant_connector_account_id_of_default_length(),
-            connector_account_details: domain_types::crypto_operation(
-                key_manager_state,
-                type_name!(domain::MerchantConnectorAccount),
-                domain_types::CryptoOperation::Encrypt(self.connector_account_details.ok_or(
-                    errors::ApiErrorResponse::MissingRequiredField {
-                        field_name: "connector_account_details",
-                    },
-                )?),
-                identifier.clone(),
-                key_store.key.peek(),
-            )
-            .await
-            .and_then(|val| val.try_into_operation())
-            .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("Unable to encrypt connector account details")?,
+            connector_account_details: encrypted_data.connector_account_details,
             payment_methods_enabled,
             disabled,
             metadata: self.metadata.clone(),
@@ -2624,26 +2627,12 @@ impl MerchantConnectorAccountCreateBridge for api::MerchantConnectorCreate {
             applepay_verified_domains: None,
             pm_auth_config: self.pm_auth_config.clone(),
             status: connector_status,
-            connector_wallets_details: helpers::get_encrypted_connector_wallets_details_with_apple_pay_certificates(state, &key_store, &self.metadata, &self.connector_wallets_details).await?,
+            connector_wallets_details: encrypted_data.connector_wallets_details,
             test_mode: self.test_mode,
             business_country: self.business_country,
             business_label: self.business_label.clone(),
             business_sub_label: self.business_sub_label.clone(),
-            additional_merchant_data: if let Some(mcd) =  merchant_recipient_data {
-                Some(domain_types::crypto_operation(
-                    key_manager_state,
-                    type_name!(domain::MerchantConnectorAccount),
-                    domain_types::CryptoOperation::Encrypt(Secret::new(mcd)),
-                    identifier,
-                    key_store.key.peek(),
-                )
-                .await
-                .and_then(|val| val.try_into_operation())
-                .change_context(errors::ApiErrorResponse::InternalServerError)
-                .attach_printable("Unable to encrypt additional_merchant_data")?)
-            } else {
-                None
-            },
+            additional_merchant_data: encrypted_data.additional_merchant_data,
             version: hyperswitch_domain_models::consts::API_VERSION,
         })
     }
diff --git a/crates/router/src/core/payments/helpers.rs b/crates/router/src/core/payments/helpers.rs
index 3782026835b..f0b1e9ca86e 100644
--- a/crates/router/src/core/payments/helpers.rs
+++ b/crates/router/src/core/payments/helpers.rs
@@ -76,10 +76,7 @@ use crate::{
     services,
     types::{
         api::{self, admin, enums as api_enums, MandateValidationFieldsExt},
-        domain::{
-            self,
-            types::{self, AsyncLift},
-        },
+        domain::{self, types},
         storage::{self, enums as storage_enums, ephemeral_key, CardTokenData},
         transformers::{ForeignFrom, ForeignTryFrom},
         AdditionalMerchantData, AdditionalPaymentMethodConnectorResponse, ErrorResponse,
@@ -4534,12 +4531,10 @@ pub fn is_apple_pay_simplified_flow(
 // As part of migration fallback this function checks apple pay details are present in connector_wallets_details
 // If yes, it will encrypt connector_wallets_details and store it in the database.
 // If no, it will check if apple pay details are present in metadata and merge it with connector_wallets_details, encrypt and store it.
-pub async fn get_encrypted_connector_wallets_details_with_apple_pay_certificates(
-    state: &SessionState,
-    key_store: &domain::MerchantKeyStore,
+pub async fn get_connector_wallets_details_with_apple_pay_certificates(
     connector_metadata: &Option<masking::Secret<tera::Value>>,
     connector_wallets_details_optional: &Option<api_models::admin::ConnectorWalletDetails>,
-) -> RouterResult<Option<Encryptable<masking::Secret<serde_json::Value>>>> {
+) -> RouterResult<Option<masking::Secret<serde_json::Value>>> {
     let connector_wallet_details_with_apple_pay_metadata_optional =
         get_apple_pay_metadata_if_needed(connector_metadata, connector_wallets_details_optional)
             .await?;
@@ -4553,25 +4548,7 @@ pub async fn get_encrypted_connector_wallets_details_with_apple_pay_certificates
         .transpose()?
         .map(masking::Secret::new);
 
-    let key_manager_state: KeyManagerState = state.into();
-    let encrypted_connector_wallets_details = connector_wallets_details
-        .clone()
-        .async_lift(|wallets_details| async {
-            types::crypto_operation(
-                &key_manager_state,
-                type_name!(domain::MerchantConnectorAccount),
-                types::CryptoOperation::EncryptOptional(wallets_details),
-                Identifier::Merchant(key_store.merchant_id.clone()),
-                key_store.key.get_inner().peek(),
-            )
-            .await
-            .and_then(|val| val.try_into_optionaloperation())
-        })
-        .await
-        .change_context(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable("Failed while encrypting connector wallets details")?;
-
-    Ok(encrypted_connector_wallets_details)
+    Ok(connector_wallets_details)
 }
 
 async fn get_apple_pay_metadata_if_needed(

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

## Description
<!-- Describe your changes in detail -->
This adds batch encrypt/decrypt support for Merchant Connector account create/fetch/update

## 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).
-->
Closes #6205


## 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)?
-->
1. Create MerchantConnector Account
```bash
{
    "connector_type": "fiz_operations",
    "connector_name": "checkout_test",
    "connector_label": "checkout_test_US_default",
    "merchant_connector_id": "mca_1nx2lpMSrHoJZ0Xiw3CM",
    "profile_id": "pro_1tS3symGpqUT5zGglYyI",
    "connector_account_details": {
        "auth_type": "HeaderKey",
        "api_key": "pk*******************************u5"
    },
    "payment_methods_enabled": [
        {
            "payment_method": "card",
            "payment_method_types": [
                {
                    "payment_method_type": "credit",
                    "payment_experience": null,
                    "card_networks": [
                        "Visa",
                        "Mastercard"
                    ],
                    "accepted_currencies": null,
                    "accepted_countries": null,
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                },
                {
                    "payment_method_type": "debit",
                    "payment_experience": null,
                    "card_networks": [
                        "Visa",
                        "Mastercard"
                    ],
                    "accepted_currencies": null,
                    "accepted_countries": null,
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                }
            ]
        },
        {
            "payment_method": "wallet",
            "payment_method_types": [
                {
                    "payment_method_type": "apple_pay",
                    "payment_experience": "invoke_sdk_client",
                    "card_networks": null,
                    "accepted_currencies": null,
                    "accepted_countries": null,
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                },
                {
                    "payment_method_type": "google_pay",
                    "payment_experience": "invoke_sdk_client",
                    "card_networks": null,
                    "accepted_currencies": null,
                    "accepted_countries": null,
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                }
            ]
        },
        {
            "payment_method": "bank_redirect",
            "payment_method_types": [
                {
                    "payment_method_type": "eps",
                    "payment_experience": null,
                    "card_networks": null,
                    "accepted_currencies": null,
                    "accepted_countries": null,
                    "minimum_amount": null,
                    "maximum_amount": null,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                },
                {
                    "payment_method_type": "sofort",
                    "payment_experience": null,
                    "card_networks": null,
                    "accepted_currencies": null,
                    "accepted_countries": null,
                    "minimum_amount": null,
                    "maximum_amount": null,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                },
                {
                    "payment_method_type": "giropay",
                    "payment_experience": null,
                    "card_networks": null,
                    "accepted_currencies": null,
                    "accepted_countries": null,
                    "minimum_amount": null,
                    "maximum_amount": null,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                },
                {
                    "payment_method_type": "ideal",
                    "payment_experience": null,
                    "card_networks": null,
                    "accepted_currencies": null,
                    "accepted_countries": null,
                    "minimum_amount": null,
                    "maximum_amount": null,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                }
            ]
        }
    ],
    "connector_webhook_details": null,
    "metadata": {
        "google_pay": {
            "merchant_info": {
                "merchant_id": "sundari",
                "merchant_name": "gpay"
            },
            "allowed_payment_methods": [
                {
                    "type": "CARD",
                    "parameters": {
                        "allowed_auth_methods": [
                            "PAN_ONLY",
                            "CRYPTOGRAM_3DS"
                        ],
                        "allowed_card_networks": [
                            "AMEX",
                            "DISCOVER",
                            "INTERAC",
                            "JCB",
                            "MASTERCARD",
                            "VISA"
                        ]
                    },
                    "tokenization_specification": {
                        "type": "PAYMENT_GATEWAY",
                        "parameters": {
                            "gateway": "checkoutltd",
                            "gateway_merchant_id": "pk_sbox_ivbndwrrb35t67wrveqic347ii4"
                        }
                    }
                }
            ]
        },
                "payment_request_data": {
                    "label": "applepay",
                    "supported_networks": [
                        "visa",
                        "masterCard",
                        "amex",
                        "discover"
                    ],
                    "merchant_capabilities": [
                        "supports3DS"
                    ]
                }
            }
        }
    },
    "test_mode": false,
    "disabled": false,
    "frm_configs": null,
    "business_country": "US",
    "business_label": "default",
    "business_sub_label": null,
    "applepay_verified_domains": null,
    "pm_auth_config": null,
    "status": "active",
    "additional_merchant_data": null,
    "connector_wallets_details": {
        "apple_pay_combined": {
            "manual": {
                "payment_request_data": {
                    "supported_networks": [
                        "visa",
                        "masterCard",
                        "amex",
                        "discover"
                    ],
                    "merchant_capabilities": [
                        "supports3DS"
                    ],
                    "label": "applepay"
                },
                    "merchant_identifier": "merchant.com.hyperswitch.checkout",
                    "display_name": "checkoutapple",
                    "initiative": "web",
                    "initiative_context": "sdk-test-app.netlify.app",
                    "merchant_business_country": null
                }
            }
        }
    }
}
```
2. Update MerchantConnectorAccount
```bash
curl --location 'http://localhost:8080/account/merchant_123_1727938790/connectors/mca_1nx2lpMSrHoJZ0Xiw3CM' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: test_admin' \
--data '{
    "connector_type": "fiz_operations",
    "connector_account_details": {
        "auth_type": "HeaderKey",
        "api_key": ""
    },
    "test_mode": false,
    "disabled": false,
    "payment_methods_enabled": [
        {
            "payment_method": "card",
            "payment_method_types": [
                {
                    "payment_method_type": "credit",
                    "card_networks": [
                        "Visa",
                        "Mastercard"
                    ],
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true,
                    "accepted_countries": {
                        "type": "disable_only",
                        "list": [
                            "US",
                            "GB",
                            "IN"
                        ]
                    },
                    "accepted_currencies": {
                        "type": "enable_only",
                        "list": [
                            "USD",
                            "GBP",
                            "INR"
                        ]
                    }
                },
                {
                    "payment_method_type": "debit",
                    "card_networks": [
                        "Visa",
                        "Mastercard"
                    ],
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true,
                    "accepted_countries": {
                        "type": "enable_only",
                        "list": [
                            "US",
                            "GB",
                            "IN"
                        ]
                    },
                    "accepted_currencies": {
                        "type": "disable_only",
                        "list": [
                            "USD",
                            "GBP",
                            "INR"
                        ]
                    }
                }
            ]
        },
        {
            "payment_method": "pay_later",
            "payment_method_types": [
                {
                    "payment_method_type": "klarna",
                    "payment_experience": "redirect_to_url",
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                },
                {
                    "payment_method_type": "affirm",
                    "payment_experience": "redirect_to_url",
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                },
                {
                    "payment_method_type": "afterpay_clearpay",
                    "payment_experience": "redirect_to_url",
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                }
            ]
        }
    ],
    "metadata": {
        "city": "NY",
        "unit": "245"
    },
    "connector_webhook_details": {
        "merchant_secret": "MyWebhookSecret"
    },
    "business_country": "US",
    "business_label": "default"
}'
```

3. Sanity test on `/payments`
```bash
curl --location 'http://localhost:8080/payments' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'x-feature: router-custom' \
--header 'api-key: dev_BPQpf0oItObooOxXtmyACTUJD9TpwU3jjs3YCkStXbCDwMKYCauROqOqqRBQpJLi' \
--data-raw '{
    "amount": 600,
    "currency": "USD",
    "confirm": true,
    "capture_method": "automatic",
    "customer_id": "HScustomer1234",
    "email": "p143@example.com",
    "amount_to_capture": 600,
    "description": "Its my first payment request",
    "capture_on": "2022-09-10T10:11:12Z",
    "return_url": "https://google.com",
    "name": "Preetam",
    "phone": "999999999",
    "setup_future_usage": "off_session",
    "phone_country_code": "+65",
    "authentication_type": "three_ds",
    "payment_method": "card",
    "payment_method_type": "debit",
    "payment_method_data": {
        "card": {
            "card_number": "4111111111111111",
            "card_exp_month": "03",
            "card_exp_year": "2030",
            "card_holder_name": "joseph Doe",
            "card_cvc": "737"
        }
    },
    "connector_metadata": {
        "noon": {
            "order_category": "applepay"
        }
    },
    "browser_info": {
        "user_agent": "Mozilla\/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit\/537.36 (KHTML, like Gecko) Chrome\/70.0.3538.110 Safari\/537.36",
        "accept_header": "text\/html,application\/xhtml+xml,application\/xml;q=0.9,image\/webp,image\/apng,*\/*;q=0.8",
        "language": "nl-NL",
        "color_depth": 24,
        "screen_height": 723,
        "screen_width": 1536,
        "time_zone": 0,
        "java_enabled": true,
        "java_script_enabled": true,
        "ip_address": "128.0.0.1"
    },
    "billing": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "preetam",
            "last_name": "revankar"
        },
        "phone": {
            "number": "8056594427",
            "country_code": "+91"
        }
    },
    "shipping": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": {
            "number": "8056594427",
            "country_code": "+91"
        }
    },
    "order_details": [
        {
            "product_name": "Apple iphone 15",
            "quantity": 1,
            "amount": 6540,
            "account_name": "transaction_processing"
        }
    ],
    "statement_descriptor_name": "joseph",
    "statement_descriptor_suffix": "JS",
    "metadata": {
        "udf1": "value1",
        "new_customer": "true",
        "login_date": "2019-09-10T10:11:12Z"
    }
}'
```
## 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