feat: encryption service integration to support batch encryption and decryption (#5164)

Co-authored-by: dracarys18 <karthikey.hegde@juspay.in>
Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>

crates/storage_impl/src/mock_db/payment_intent.rs

@@ -1,4 +1,4 @@
-use common_utils::errors::CustomResult;
+use common_utils::{errors::CustomResult, types::keymanager::KeyManagerState};
 use diesel_models::enums as storage_enums;
 use error_stack::ResultExt;
 use hyperswitch_domain_models::{
@@ -19,6 +19,7 @@ impl PaymentIntentInterface for MockDb {
     #[cfg(feature = "olap")]
     async fn filter_payment_intent_by_constraints(
         &self,
+        _state: &KeyManagerState,
         _merchant_id: &str,
         _filters: &hyperswitch_domain_models::payments::payment_intent::PaymentIntentFetchConstraints,
         _key_store: &MerchantKeyStore,
@@ -30,6 +31,7 @@ impl PaymentIntentInterface for MockDb {
     #[cfg(feature = "olap")]
     async fn filter_payment_intents_by_time_range_constraints(
         &self,
+        _state: &KeyManagerState,
         _merchant_id: &str,
         _time_range: &api_models::payments::TimeRange,
         _key_store: &MerchantKeyStore,
@@ -51,6 +53,7 @@ impl PaymentIntentInterface for MockDb {
     #[cfg(feature = "olap")]
     async fn get_filtered_payment_intents_attempt(
         &self,
+        _state: &KeyManagerState,
         _merchant_id: &str,
         _constraints: &hyperswitch_domain_models::payments::payment_intent::PaymentIntentFetchConstraints,
         _key_store: &MerchantKeyStore,
@@ -63,6 +66,7 @@ impl PaymentIntentInterface for MockDb {
     #[allow(clippy::panic)]
     async fn insert_payment_intent(
         &self,
+        _state: &KeyManagerState,
         new: PaymentIntent,
         _key_store: &MerchantKeyStore,
         _storage_scheme: storage_enums::MerchantStorageScheme,
@@ -76,6 +80,7 @@ impl PaymentIntentInterface for MockDb {
     #[allow(clippy::unwrap_used)]
     async fn update_payment_intent(
         &self,
+        state: &KeyManagerState,
         this: PaymentIntent,
         update: PaymentIntentUpdate,
         key_store: &MerchantKeyStore,
@@ -95,8 +100,10 @@ impl PaymentIntentInterface for MockDb {
             .change_context(StorageError::EncryptionError)?;
 
         *payment_intent = PaymentIntent::convert_back(
+            state,
             diesel_payment_intent_update.apply_changeset(diesel_payment_intent),
             key_store.key.get_inner(),
+            key_store.merchant_id.clone(),
         )
         .await
         .change_context(StorageError::DecryptionError)?;
@@ -108,6 +115,7 @@ impl PaymentIntentInterface for MockDb {
     #[allow(clippy::unwrap_used)]
     async fn find_payment_intent_by_payment_id_merchant_id(
         &self,
+        _state: &KeyManagerState,
         payment_id: &str,
         merchant_id: &str,
         _key_store: &MerchantKeyStore,

crates/storage_impl/src/payments/payment_intent.rs

@@ -4,7 +4,10 @@ use api_models::payments::AmountFilter;
 use async_bb8_diesel::{AsyncConnection, AsyncRunQueryDsl};
 #[cfg(feature = "olap")]
 use common_utils::errors::ReportSwitchExt;
-use common_utils::ext_traits::{AsyncExt, Encode};
+use common_utils::{
+    ext_traits::{AsyncExt, Encode},
+    types::keymanager::KeyManagerState,
+};
 #[cfg(feature = "olap")]
 use diesel::{associations::HasTable, ExpressionMethods, JoinOnDsl, QueryDsl};
 use diesel_models::{
@@ -53,6 +56,7 @@ use crate::{
 impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
     async fn insert_payment_intent(
         &self,
+        state: &KeyManagerState,
         payment_intent: PaymentIntent,
         merchant_key_store: &MerchantKeyStore,
         storage_scheme: MerchantStorageScheme,
@@ -69,7 +73,12 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
         match storage_scheme {
             MerchantStorageScheme::PostgresOnly => {
                 self.router_store
-                    .insert_payment_intent(payment_intent, merchant_key_store, storage_scheme)
+                    .insert_payment_intent(
+                        state,
+                        payment_intent,
+                        merchant_key_store,
+                        storage_scheme,
+                    )
                     .await
             }
 
@@ -121,6 +130,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
     #[instrument(skip_all)]
     async fn update_payment_intent(
         &self,
+        state: &KeyManagerState,
         this: PaymentIntent,
         payment_intent_update: PaymentIntentUpdate,
         merchant_key_store: &MerchantKeyStore,
@@ -143,6 +153,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
             MerchantStorageScheme::PostgresOnly => {
                 self.router_store
                     .update_payment_intent(
+                        state,
                         this,
                         payment_intent_update,
                         merchant_key_store,
@@ -189,10 +200,14 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
                 .try_into_hset()
                 .change_context(StorageError::KVError)?;
 
-                let payment_intent =
-                    PaymentIntent::convert_back(diesel_intent, merchant_key_store.key.get_inner())
-                        .await
-                        .change_context(StorageError::DecryptionError)?;
+                let payment_intent = PaymentIntent::convert_back(
+                    state,
+                    diesel_intent,
+                    merchant_key_store.key.get_inner(),
+                    merchant_id,
+                )
+                .await
+                .change_context(StorageError::DecryptionError)?;
 
                 Ok(payment_intent)
             }
@@ -202,6 +217,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
     #[instrument(skip_all)]
     async fn find_payment_intent_by_payment_id_merchant_id(
         &self,
+        state: &KeyManagerState,
         payment_id: &str,
         merchant_id: &str,
         merchant_key_store: &MerchantKeyStore,
@@ -243,9 +259,14 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
             }
         }?;
 
-        PaymentIntent::convert_back(diesel_payment_intent, merchant_key_store.key.get_inner())
-            .await
-            .change_context(StorageError::DecryptionError)
+        PaymentIntent::convert_back(
+            state,
+            diesel_payment_intent,
+            merchant_key_store.key.get_inner(),
+            merchant_id.to_string(),
+        )
+        .await
+        .change_context(StorageError::DecryptionError)
     }
 
     async fn get_active_payment_attempt(
@@ -278,6 +299,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
     #[cfg(feature = "olap")]
     async fn filter_payment_intent_by_constraints(
         &self,
+        state: &KeyManagerState,
         merchant_id: &str,
         filters: &PaymentIntentFetchConstraints,
         merchant_key_store: &MerchantKeyStore,
@@ -285,6 +307,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
     ) -> error_stack::Result<Vec<PaymentIntent>, StorageError> {
         self.router_store
             .filter_payment_intent_by_constraints(
+                state,
                 merchant_id,
                 filters,
                 merchant_key_store,
@@ -296,6 +319,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
     #[cfg(feature = "olap")]
     async fn filter_payment_intents_by_time_range_constraints(
         &self,
+        state: &KeyManagerState,
         merchant_id: &str,
         time_range: &api_models::payments::TimeRange,
         merchant_key_store: &MerchantKeyStore,
@@ -303,6 +327,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
     ) -> error_stack::Result<Vec<PaymentIntent>, StorageError> {
         self.router_store
             .filter_payment_intents_by_time_range_constraints(
+                state,
                 merchant_id,
                 time_range,
                 merchant_key_store,
@@ -314,6 +339,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
     #[cfg(feature = "olap")]
     async fn get_filtered_payment_intents_attempt(
         &self,
+        state: &KeyManagerState,
         merchant_id: &str,
         filters: &PaymentIntentFetchConstraints,
         merchant_key_store: &MerchantKeyStore,
@@ -321,6 +347,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
     ) -> error_stack::Result<Vec<(PaymentIntent, PaymentAttempt)>, StorageError> {
         self.router_store
             .get_filtered_payment_intents_attempt(
+                state,
                 merchant_id,
                 filters,
                 merchant_key_store,
@@ -351,6 +378,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
     #[instrument(skip_all)]
     async fn insert_payment_intent(
         &self,
+        state: &KeyManagerState,
         payment_intent: PaymentIntent,
         merchant_key_store: &MerchantKeyStore,
         _storage_scheme: MerchantStorageScheme,
@@ -367,14 +395,20 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
                 er.change_context(new_err)
             })?;
 
-        PaymentIntent::convert_back(diesel_payment_intent, merchant_key_store.key.get_inner())
-            .await
-            .change_context(StorageError::DecryptionError)
+        PaymentIntent::convert_back(
+            state,
+            diesel_payment_intent,
+            merchant_key_store.key.get_inner(),
+            merchant_key_store.merchant_id.clone(),
+        )
+        .await
+        .change_context(StorageError::DecryptionError)
     }
 
     #[instrument(skip_all)]
     async fn update_payment_intent(
         &self,
+        state: &KeyManagerState,
         this: PaymentIntent,
         payment_intent: PaymentIntentUpdate,
         merchant_key_store: &MerchantKeyStore,
@@ -394,14 +428,20 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
                 er.change_context(new_err)
             })?;
 
-        PaymentIntent::convert_back(diesel_payment_intent, merchant_key_store.key.get_inner())
-            .await
-            .change_context(StorageError::DecryptionError)
+        PaymentIntent::convert_back(
+            state,
+            diesel_payment_intent,
+            merchant_key_store.key.get_inner(),
+            merchant_key_store.merchant_id.clone(),
+        )
+        .await
+        .change_context(StorageError::DecryptionError)
     }
 
     #[instrument(skip_all)]
     async fn find_payment_intent_by_payment_id_merchant_id(
         &self,
+        state: &KeyManagerState,
         payment_id: &str,
         merchant_id: &str,
         merchant_key_store: &MerchantKeyStore,
@@ -417,8 +457,10 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
             })
             .async_and_then(|diesel_payment_intent| async {
                 PaymentIntent::convert_back(
+                    state,
                     diesel_payment_intent,
                     merchant_key_store.key.get_inner(),
+                    merchant_key_store.merchant_id.clone(),
                 )
                 .await
                 .change_context(StorageError::DecryptionError)
@@ -458,6 +500,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
     #[instrument(skip_all)]
     async fn filter_payment_intent_by_constraints(
         &self,
+        state: &KeyManagerState,
         merchant_id: &str,
         filters: &PaymentIntentFetchConstraints,
         merchant_key_store: &MerchantKeyStore,
@@ -498,6 +541,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
                         // TODO: Fetch partial columns for this query since we only need some columns
                         let starting_at = self
                             .find_payment_intent_by_payment_id_merchant_id(
+                                state,
                                 starting_after_id,
                                 merchant_id,
                                 merchant_key_store,
@@ -516,6 +560,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
                         // TODO: Fetch partial columns for this query since we only need some columns
                         let ending_at = self
                             .find_payment_intent_by_payment_id_merchant_id(
+                                state,
                                 ending_before_id,
                                 merchant_id,
                                 merchant_key_store,
@@ -560,8 +605,10 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
         .map(|payment_intents| {
             try_join_all(payment_intents.into_iter().map(|diesel_payment_intent| {
                 PaymentIntent::convert_back(
+                    state,
                     diesel_payment_intent,
                     merchant_key_store.key.get_inner(),
+                    merchant_key_store.merchant_id.clone(),
                 )
             }))
             .map(|join_result| join_result.change_context(StorageError::DecryptionError))
@@ -579,6 +626,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
     #[instrument(skip_all)]
     async fn filter_payment_intents_by_time_range_constraints(
         &self,
+        state: &KeyManagerState,
         merchant_id: &str,
         time_range: &api_models::payments::TimeRange,
         merchant_key_store: &MerchantKeyStore,
@@ -587,6 +635,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
         // TODO: Remove this redundant function
         let payment_filters = (*time_range).into();
         self.filter_payment_intent_by_constraints(
+            state,
             merchant_id,
             &payment_filters,
             merchant_key_store,
@@ -599,6 +648,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
     #[instrument(skip_all)]
     async fn get_filtered_payment_intents_attempt(
         &self,
+        state: &KeyManagerState,
         merchant_id: &str,
         constraints: &PaymentIntentFetchConstraints,
         merchant_key_store: &MerchantKeyStore,
@@ -640,6 +690,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
                         // TODO: Fetch partial columns for this query since we only need some columns
                         let starting_at = self
                             .find_payment_intent_by_payment_id_merchant_id(
+                                state,
                                 starting_after_id,
                                 merchant_id,
                                 merchant_key_store,
@@ -658,6 +709,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
                         // TODO: Fetch partial columns for this query since we only need some columns
                         let ending_at = self
                             .find_payment_intent_by_payment_id_merchant_id(
+                                state,
                                 ending_before_id,
                                 merchant_id,
                                 merchant_key_store,
@@ -745,13 +797,17 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
             .await
             .map(|results| {
                 try_join_all(results.into_iter().map(|(pi, pa)| {
-                    PaymentIntent::convert_back(pi, merchant_key_store.key.get_inner()).map(
-                        |payment_intent| {
-                            payment_intent.map(|payment_intent| {
-                                (payment_intent, PaymentAttempt::from_storage_model(pa))
-                            })
-                        },
+                    PaymentIntent::convert_back(
+                        state,
+                        pi,
+                        merchant_key_store.key.get_inner(),
+                        merchant_id.to_string(),
                     )
+                    .map(|payment_intent| {
+                        payment_intent.map(|payment_intent| {
+                            (payment_intent, PaymentAttempt::from_storage_model(pa))
+                        })
+                    })
                 }))
                 .map(|join_result| join_result.change_context(StorageError::DecryptionError))
             })