Bug: [FIX] use batch encryption/decryption for payment intent table




diff --git a/crates/hyperswitch_domain_models/src/payments/payment_intent.rs b/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
index bd4707243ce..562e5d205c0 100644
--- a/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
+++ b/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
@@ -1,14 +1,14 @@
 use common_enums as storage_enums;
 use common_utils::{
     consts::{PAYMENTS_LIST_MAX_LIMIT_V1, PAYMENTS_LIST_MAX_LIMIT_V2},
-    crypto::Encryptable,
+    crypto::{self, Encryptable},
     encryption::Encryption,
     errors::{CustomResult, ValidationError},
     id_type,
     pii::{self, Email},
     type_name,
     types::{
-        keymanager::{self, KeyManagerState},
+        keymanager::{self, KeyManagerState, ToEncryptable},
         MinorUnit,
     },
 };
@@ -17,6 +17,7 @@ use diesel_models::{
 };
 use error_stack::ResultExt;
 use masking::{Deserialize, PeekInterface, Secret};
+use rustc_hash::FxHashMap;
 use serde::Serialize;
 use time::PrimitiveDateTime;
 
@@ -26,7 +27,7 @@ use super::PaymentIntent;
 use crate::{
     behaviour, errors,
     merchant_key_store::MerchantKeyStore,
-    type_encryption::{crypto_operation, AsyncLift, CryptoOperation},
+    type_encryption::{crypto_operation, CryptoOperation},
     RemoteStorageObject,
 };
 #[async_trait::async_trait]
@@ -1568,17 +1569,25 @@ impl behaviour::Conversion for PaymentIntent {
         Self: Sized,
     {
         async {
-            let inner_decrypt = |inner| async {
-                crypto_operation(
-                    state,
-                    type_name!(Self::DstType),
-                    CryptoOperation::DecryptOptional(inner),
-                    key_manager_identifier.clone(),
-                    key.peek(),
-                )
-                .await
-                .and_then(|val| val.try_into_optionaloperation())
-            };
+            let decrypted_data = crypto_operation(
+                state,
+                type_name!(Self::DstType),
+                CryptoOperation::BatchDecrypt(EncryptedPaymentIntentAddress::to_encryptable(
+                    EncryptedPaymentIntentAddress {
+                        billing: storage_model.billing_address,
+                        shipping: storage_model.shipping_address,
+                        customer_details: storage_model.customer_details,
+                    },
+                )),
+                key_manager_identifier,
+                key.peek(),
+            )
+            .await
+            .and_then(|val| val.try_into_batchoperation())?;
+
+            let data = EncryptedPaymentIntentAddress::from_encryptable(decrypted_data)
+                .change_context(common_utils::errors::CryptoError::DecodingFailed)
+                .attach_printable("Invalid batch operation data")?;
 
             let amount_details = super::AmountDetails {
                 order_amount: storage_model.amount,
@@ -1628,18 +1637,9 @@ impl behaviour::Conversion for PaymentIntent {
                         storage_model.request_external_three_ds_authentication,
                     ),
                 frm_metadata: storage_model.frm_metadata,
-                customer_details: storage_model
-                    .customer_details
-                    .async_lift(inner_decrypt)
-                    .await?,
-                billing_address: storage_model
-                    .billing_address
-                    .async_lift(inner_decrypt)
-                    .await?,
-                shipping_address: storage_model
-                    .shipping_address
-                    .async_lift(inner_decrypt)
-                    .await?,
+                customer_details: data.customer_details,
+                billing_address: data.billing,
+                shipping_address: data.shipping,
                 capture_method: storage_model.capture_method,
                 id: storage_model.id,
                 merchant_reference_id: storage_model.merchant_reference_id,
@@ -1796,17 +1796,26 @@ impl behaviour::Conversion for PaymentIntent {
         Self: Sized,
     {
         async {
-            let inner_decrypt = |inner| async {
-                crypto_operation(
-                    state,
-                    type_name!(Self::DstType),
-                    CryptoOperation::DecryptOptional(inner),
-                    key_manager_identifier.clone(),
-                    key.peek(),
-                )
-                .await
-                .and_then(|val| val.try_into_optionaloperation())
-            };
+            let decrypted_data = crypto_operation(
+                state,
+                type_name!(Self::DstType),
+                CryptoOperation::BatchDecrypt(EncryptedPaymentIntentAddress::to_encryptable(
+                    EncryptedPaymentIntentAddress {
+                        billing: storage_model.billing_details,
+                        shipping: storage_model.shipping_details,
+                        customer_details: storage_model.customer_details,
+                    },
+                )),
+                key_manager_identifier,
+                key.peek(),
+            )
+            .await
+            .and_then(|val| val.try_into_batchoperation())?;
+
+            let data = EncryptedPaymentIntentAddress::from_encryptable(decrypted_data)
+                .change_context(common_utils::errors::CryptoError::DecodingFailed)
+                .attach_printable("Invalid batch operation data")?;
+
             Ok::<Self, error_stack::Report<common_utils::errors::CryptoError>>(Self {
                 payment_id: storage_model.payment_id,
                 merchant_id: storage_model.merchant_id,
@@ -1854,19 +1863,10 @@ impl behaviour::Conversion for PaymentIntent {
                 frm_metadata: storage_model.frm_metadata,
                 shipping_cost: storage_model.shipping_cost,
                 tax_details: storage_model.tax_details,
-                customer_details: storage_model
-                    .customer_details
-                    .async_lift(inner_decrypt)
-                    .await?,
-                billing_details: storage_model
-                    .billing_details
-                    .async_lift(inner_decrypt)
-                    .await?,
+                customer_details: data.customer_details,
+                billing_details: data.billing,
                 merchant_order_reference_id: storage_model.merchant_order_reference_id,
-                shipping_details: storage_model
-                    .shipping_details
-                    .async_lift(inner_decrypt)
-                    .await?,
+                shipping_details: data.shipping,
                 is_payment_processor_token_flow: storage_model.is_payment_processor_token_flow,
                 organization_id: storage_model.organization_id,
                 skip_external_tax_calculation: storage_model.skip_external_tax_calculation,
@@ -1935,3 +1935,73 @@ impl behaviour::Conversion for PaymentIntent {
         })
     }
 }
+
+pub struct EncryptedPaymentIntentAddress {
+    pub shipping: Option<Encryption>,
+    pub billing: Option<Encryption>,
+    pub customer_details: Option<Encryption>,
+}
+
+pub struct PaymentAddressFromRequest {
+    pub shipping: Option<Secret<serde_json::Value>>,
+    pub billing: Option<Secret<serde_json::Value>>,
+    pub customer_details: Option<Secret<serde_json::Value>>,
+}
+
+pub struct DecryptedPaymentIntentAddress {
+    pub shipping: crypto::OptionalEncryptableValue,
+    pub billing: crypto::OptionalEncryptableValue,
+    pub customer_details: crypto::OptionalEncryptableValue,
+}
+
+impl ToEncryptable<DecryptedPaymentIntentAddress, Secret<serde_json::Value>, Encryption>
+    for EncryptedPaymentIntentAddress
+{
+    fn from_encryptable(
+        mut hashmap: FxHashMap<String, Encryptable<Secret<serde_json::Value>>>,
+    ) -> CustomResult<DecryptedPaymentIntentAddress, common_utils::errors::ParsingError> {
+        Ok(DecryptedPaymentIntentAddress {
+            shipping: hashmap.remove("shipping"),
+            billing: hashmap.remove("billing"),
+            customer_details: hashmap.remove("customer_details"),
+        })
+    }
+
+    fn to_encryptable(self) -> FxHashMap<String, Encryption> {
+        let mut map = FxHashMap::with_capacity_and_hasher(9, Default::default());
+
+        self.shipping.map(|s| map.insert("shipping".to_string(), s));
+        self.billing.map(|s| map.insert("billing".to_string(), s));
+        self.customer_details
+            .map(|s| map.insert("customer_details".to_string(), s));
+        map
+    }
+}
+
+impl
+    ToEncryptable<
+        DecryptedPaymentIntentAddress,
+        Secret<serde_json::Value>,
+        Secret<serde_json::Value>,
+    > for PaymentAddressFromRequest
+{
+    fn from_encryptable(
+        mut hashmap: FxHashMap<String, Encryptable<Secret<serde_json::Value>>>,
+    ) -> CustomResult<DecryptedPaymentIntentAddress, common_utils::errors::ParsingError> {
+        Ok(DecryptedPaymentIntentAddress {
+            shipping: hashmap.remove("shipping"),
+            billing: hashmap.remove("billing"),
+            customer_details: hashmap.remove("customer_details"),
+        })
+    }
+
+    fn to_encryptable(self) -> FxHashMap<String, Secret<serde_json::Value>> {
+        let mut map = FxHashMap::with_capacity_and_hasher(9, Default::default());
+
+        self.shipping.map(|s| map.insert("shipping".to_string(), s));
+        self.billing.map(|s| map.insert("billing".to_string(), s));
+        self.customer_details
+            .map(|s| map.insert("customer_details".to_string(), s));
+        map
+    }
+}
diff --git a/crates/router/Cargo.toml b/crates/router/Cargo.toml
index 31cfae49875..5becef839d5 100644
--- a/crates/router/Cargo.toml
+++ b/crates/router/Cargo.toml
@@ -17,7 +17,7 @@ email = ["external_services/email", "scheduler/email", "olap"]
 # keymanager_create, keymanager_mtls, encryption_service should not be removed or added to default feature. Once this features were enabled it can't be disabled as these are breaking changes.
 keymanager_create = []
 keymanager_mtls = ["reqwest/rustls-tls", "common_utils/keymanager_mtls"]
-encryption_service = ["hyperswitch_domain_models/encryption_service", "common_utils/encryption_service"]
+encryption_service = ["keymanager_create","hyperswitch_domain_models/encryption_service", "common_utils/encryption_service"]
 km_forward_x_request_id = ["common_utils/km_forward_x_request_id"]
 frm = ["api_models/frm", "hyperswitch_domain_models/frm", "hyperswitch_connectors/frm", "hyperswitch_interfaces/frm"]
 stripe = []
diff --git a/crates/router/src/core/admin.rs b/crates/router/src/core/admin.rs
index d27b6b82fd4..099ce4aced1 100644
--- a/crates/router/src/core/admin.rs
+++ b/crates/router/src/core/admin.rs
@@ -194,7 +194,7 @@ pub async fn create_merchant_account(
 
     let master_key = db.get_master_key();
 
-    let key_manager_state = &(&state).into();
+    let key_manager_state: &KeyManagerState = &(&state).into();
     let merchant_id = req.get_merchant_reference_id();
     let identifier = km_types::Identifier::Merchant(merchant_id.clone());
     #[cfg(feature = "keymanager_create")]
@@ -203,16 +203,18 @@ pub async fn create_merchant_account(
 
         use crate::consts::BASE64_ENGINE;
 
-        keymanager::transfer_key_to_key_manager(
-            key_manager_state,
-            EncryptionTransferRequest {
-                identifier: identifier.clone(),
-                key: BASE64_ENGINE.encode(key),
-            },
-        )
-        .await
-        .change_context(errors::ApiErrorResponse::DuplicateMerchantAccount)
-        .attach_printable("Failed to insert key to KeyManager")?;
+        if key_manager_state.enabled {
+            keymanager::transfer_key_to_key_manager(
+                key_manager_state,
+                EncryptionTransferRequest {
+                    identifier: identifier.clone(),
+                    key: BASE64_ENGINE.encode(key),
+                },
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::DuplicateMerchantAccount)
+            .attach_printable("Failed to insert key to KeyManager")?;
+        }
     }
 
     let key_store = domain::MerchantKeyStore {
diff --git a/crates/router/src/core/payments/operations/payment_create.rs b/crates/router/src/core/payments/operations/payment_create.rs
index 51f2de51f26..0921e38317a 100644
--- a/crates/router/src/core/payments/operations/payment_create.rs
+++ b/crates/router/src/core/payments/operations/payment_create.rs
@@ -7,13 +7,20 @@ use api_models::{
 use async_trait::async_trait;
 use common_utils::{
     ext_traits::{AsyncExt, Encode, ValueExt},
-    types::{keymanager::KeyManagerState, MinorUnit},
+    type_name,
+    types::{
+        keymanager::{Identifier, KeyManagerState, ToEncryptable},
+        MinorUnit,
+    },
 };
 use diesel_models::ephemeral_key;
 use error_stack::{self, ResultExt};
 use hyperswitch_domain_models::{
     mandates::{MandateData, MandateDetails},
-    payments::{payment_attempt::PaymentAttempt, payment_intent::CustomerData},
+    payments::{
+        payment_attempt::PaymentAttempt,
+        payment_intent::{CustomerData, PaymentAddressFromRequest},
+    },
 };
 use masking::{ExposeInterface, PeekInterface, Secret};
 use router_derive::PaymentOperation;
@@ -1280,28 +1287,6 @@ impl PaymentCreate {
             .change_context(errors::ApiErrorResponse::InternalServerError)?
             .map(Secret::new);
 
-        // Derivation of directly supplied Billing Address data in our Payment Create Request
-        // Encrypting our Billing Address Details to be stored in Payment Intent
-        let billing_details = request
-            .billing
-            .clone()
-            .async_map(|billing_details| create_encrypted_data(state, key_store, billing_details))
-            .await
-            .transpose()
-            .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("Unable to encrypt billing details")?;
-
-        // Derivation of directly supplied Shipping Address data in our Payment Create Request
-        // Encrypting our Shipping Address Details to be stored in Payment Intent
-        let shipping_details = request
-            .shipping
-            .clone()
-            .async_map(|shipping_details| create_encrypted_data(state, key_store, shipping_details))
-            .await
-            .transpose()
-            .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("Unable to encrypt shipping details")?;
-
         // Derivation of directly supplied Customer data in our Payment Create Request
         let raw_customer_details = if request.customer_id.is_none()
             && (request.name.is_some()
@@ -1325,13 +1310,53 @@ impl PaymentCreate {
             },
         );
 
-        // Encrypting our Customer Details to be stored in Payment Intent
-        let customer_details = raw_customer_details
-            .async_map(|customer_details| create_encrypted_data(state, key_store, customer_details))
-            .await
+        let key = key_store.key.get_inner().peek();
+        let identifier = Identifier::Merchant(key_store.merchant_id.clone());
+        let key_manager_state: KeyManagerState = state.into();
+
+        let shipping_details_encoded = request
+            .shipping
+            .clone()
+            .map(|shipping| Encode::encode_to_value(&shipping).map(Secret::new))
             .transpose()
             .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("Unable to encrypt customer details")?;
+            .attach_printable("Unable to encode billing details to serde_json::Value")?;
+
+        let billing_details_encoded = request
+            .billing
+            .clone()
+            .map(|billing| Encode::encode_to_value(&billing).map(Secret::new))
+            .transpose()
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Unable to encode billing details to serde_json::Value")?;
+
+        let customer_details_encoded = raw_customer_details
+            .map(|customer| Encode::encode_to_value(&customer).map(Secret::new))
+            .transpose()
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Unable to encode shipping details to serde_json::Value")?;
+
+        let encrypted_data = domain::types::crypto_operation(
+            &key_manager_state,
+            type_name!(storage::PaymentIntent),
+            domain::types::CryptoOperation::BatchEncrypt(
+                PaymentAddressFromRequest::to_encryptable(PaymentAddressFromRequest {
+                    shipping: shipping_details_encoded,
+                    billing: billing_details_encoded,
+                    customer_details: customer_details_encoded,
+                }),
+            ),
+            identifier.clone(),
+            key,
+        )
+        .await
+        .and_then(|val| val.try_into_batchoperation())
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Unable to encrypt data")?;
+
+        let encrypted_data = PaymentAddressFromRequest::from_encryptable(encrypted_data)
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Unable to encrypt the payment intent data")?;
 
         let skip_external_tax_calculation = request.skip_external_tax_calculation;
 
@@ -1382,10 +1407,10 @@ impl PaymentCreate {
                 .request_external_three_ds_authentication,
             charges,
             frm_metadata: request.frm_metadata.clone(),
-            billing_details,
-            customer_details,
+            billing_details: encrypted_data.billing,
+            customer_details: encrypted_data.customer_details,
             merchant_order_reference_id: request.merchant_order_reference_id.clone(),
-            shipping_details,
+            shipping_details: encrypted_data.shipping,
             is_payment_processor_token_flow,
             organization_id: merchant_account.organization_id.clone(),
             shipping_cost: request.shipping_cost,
diff --git a/crates/router/src/services/api.rs b/crates/router/src/services/api.rs
index 7444fea0c3b..197c1e4c52a 100644
--- a/crates/router/src/services/api.rs
+++ b/crates/router/src/services/api.rs
@@ -496,7 +496,7 @@ pub async fn send_request(
         ))
     };
 
-    // We cannot clone the request type, because it has Form trait which is not clonable. So we are cloning the request builder here.
+    // We cannot clone the request type, because it has Form trait which is not cloneable. So we are cloning the request builder here.
     let cloned_send_request = request.try_clone().map(|cloned_request| async {
         cloned_request
             .send()
@@ -570,7 +570,7 @@ pub async fn send_request(
                     .await
                 }
                 None => {
-                    logger::info!("Retrying request due to connection closed before message could complete failed as request is not clonable");
+                    logger::info!("Retrying request due to connection closed before message could complete failed as request is not cloneable");
                     Err(error)
                 }
             }

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

## Description
<!-- Describe your changes in detail -->
Batch Encrypt/Decrypt fields in Payment Intent table

## 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).
-->
Currently all the encryption/Decryption operation is happening serially. Which adds additional latency, So this PR fixes it by using the Batch encryption/decryption API of Keymanager

## 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)?
-->
### How to test it in deployed environments
Basic sanity testing of `PaymentsCreate` and `PaymentsList`.

### How did I test it locally

1. Call /payments from postman
```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"
    }
}'
```
3. Check if the encryption/decryption call happening only once in the logs.
<img width="1354" alt="Screenshot 2024-10-01 at 1 19 45 PM" src="https://github.com/user-attachments/assets/09ba719a-4ef7-4c17-9dd4-cc9cbb6dbe57">
<img width="1354" alt="Screenshot 2024-10-01 at 1 18 59 PM" src="https://github.com/user-attachments/assets/b8e185fd-22ca-4b11-aab0-74da4a3cd604">

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