Bug: [BUG] Merchant Connector Account uses application decryption which should changed to API



### Bug Description

Due to recent v2 related changes Merchant Connector Account uses application decryption instead it should be changed API call to encryption service

### Expected Behavior

all encryption and decryption should happen in the encryption service

### Actual Behavior

Only of MCA decryption is happening in Application

### Steps To Reproduce

Create MCA, application throws 5xx due to decryption failure

### Context For The Bug

_No response_

### Environment

Are you using hyperswitch hosted version? Yes
If yes, please provide the value of the `x-request-id` response header to help us debug your issue.

If not (or if building/running locally), please provide the following details:
1. Operating System or Linux distribution:
2. Rust version (output of `rustc --version`): ``
3. App version (output of `cargo r --features vergen -- --version`): ``


### Have you spent some time checking if this bug has been raised before?

- [X] I checked and didn't find a similar issue

### Have you read the Contributing Guidelines?

- [X] I have read the [Contributing Guidelines](https://github.com/juspay/hyperswitch/blob/main/docs/CONTRIBUTING.md)

### Are you willing to submit a PR?

Yes, I am willing to submit a PR!
diff --git a/crates/hyperswitch_domain_models/src/merchant_account.rs b/crates/hyperswitch_domain_models/src/merchant_account.rs
index a430673c59a..47ced6573f8 100644
--- a/crates/hyperswitch_domain_models/src/merchant_account.rs
+++ b/crates/hyperswitch_domain_models/src/merchant_account.rs
@@ -14,7 +14,7 @@ use error_stack::ResultExt;
 use masking::{PeekInterface, Secret};
 use router_env::logger;
 
-use crate::type_encryption::{decrypt, AsyncLift};
+use crate::type_encryption::{decrypt_optional, AsyncLift};
 
 #[derive(Clone, Debug, serde::Serialize)]
 pub struct MerchantAccount {
@@ -220,11 +220,15 @@ impl super::behaviour::Conversion for MerchantAccount {
                 redirect_to_merchant_with_http_post: item.redirect_to_merchant_with_http_post,
                 merchant_name: item
                     .merchant_name
-                    .async_lift(|inner| decrypt(state, inner, identifier.clone(), key.peek()))
+                    .async_lift(|inner| {
+                        decrypt_optional(state, inner, identifier.clone(), key.peek())
+                    })
                     .await?,
                 merchant_details: item
                     .merchant_details
-                    .async_lift(|inner| decrypt(state, inner, identifier.clone(), key.peek()))
+                    .async_lift(|inner| {
+                        decrypt_optional(state, inner, identifier.clone(), key.peek())
+                    })
                     .await?,
                 webhook_details: item.webhook_details,
                 sub_merchants_enabled: item.sub_merchants_enabled,
diff --git a/crates/hyperswitch_domain_models/src/merchant_key_store.rs b/crates/hyperswitch_domain_models/src/merchant_key_store.rs
index 76acf0f1404..d48403b62ff 100644
--- a/crates/hyperswitch_domain_models/src/merchant_key_store.rs
+++ b/crates/hyperswitch_domain_models/src/merchant_key_store.rs
@@ -1,5 +1,5 @@
 use common_utils::{
-    crypto::{Encryptable, GcmAes256},
+    crypto::Encryptable,
     custom_serde, date_time,
     errors::{CustomResult, ValidationError},
     types::keymanager::{Identifier, KeyManagerState},
@@ -8,7 +8,7 @@ use error_stack::ResultExt;
 use masking::{PeekInterface, Secret};
 use time::PrimitiveDateTime;
 
-use crate::type_encryption::TypeEncryption;
+use crate::type_encryption::decrypt;
 
 #[derive(Clone, Debug, serde::Serialize)]
 pub struct MerchantKeyStore {
@@ -41,7 +41,7 @@ impl super::behaviour::Conversion for MerchantKeyStore {
     {
         let identifier = Identifier::Merchant(item.merchant_id.clone());
         Ok(Self {
-            key: Encryptable::decrypt_via_api(state, item.key, identifier, key.peek(), GcmAes256)
+            key: decrypt(state, item.key, identifier, key.peek())
                 .await
                 .change_context(ValidationError::InvalidValue {
                     message: "Failed while decrypting customer data".to_string(),
diff --git a/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs b/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
index 5f8dd3934a2..1f0dc3ca0b3 100644
--- a/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
+++ b/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
@@ -15,7 +15,7 @@ use super::PaymentIntent;
 use crate::{
     behaviour, errors,
     mandates::{MandateDataType, MandateDetails},
-    type_encryption::{decrypt, AsyncLift},
+    type_encryption::{decrypt_optional, AsyncLift},
     ForeignIDRef, RemoteStorageObject,
 };
 
@@ -552,7 +552,7 @@ impl behaviour::Conversion for PaymentIntent {
     {
         async {
             let inner_decrypt = |inner| {
-                decrypt(
+                decrypt_optional(
                     state,
                     inner,
                     common_utils::types::keymanager::Identifier::Merchant(key_store_ref_id.clone()),
diff --git a/crates/hyperswitch_domain_models/src/type_encryption.rs b/crates/hyperswitch_domain_models/src/type_encryption.rs
index d656afc9251..ed36ad59aec 100644
--- a/crates/hyperswitch_domain_models/src/type_encryption.rs
+++ b/crates/hyperswitch_domain_models/src/type_encryption.rs
@@ -908,25 +908,38 @@ where
 }
 
 #[inline]
-pub async fn decrypt<T: Clone, S: masking::Strategy<T>>(
+pub async fn decrypt_optional<T: Clone, S: masking::Strategy<T>>(
     state: &KeyManagerState,
     inner: Option<Encryption>,
     identifier: Identifier,
     key: &[u8],
 ) -> CustomResult<Option<crypto::Encryptable<Secret<T, S>>>, errors::CryptoError>
+where
+    crypto::Encryptable<Secret<T, S>>: TypeEncryption<T, crypto::GcmAes256, S>,
+{
+    inner
+        .async_map(|item| decrypt(state, item, identifier, key))
+        .await
+        .transpose()
+}
+
+#[inline]
+pub async fn decrypt<T: Clone, S: masking::Strategy<T>>(
+    state: &KeyManagerState,
+    inner: Encryption,
+    identifier: Identifier,
+    key: &[u8],
+) -> CustomResult<crypto::Encryptable<Secret<T, S>>, errors::CryptoError>
 where
     crypto::Encryptable<Secret<T, S>>: TypeEncryption<T, crypto::GcmAes256, S>,
 {
     record_operation_time(
-        inner.async_map(|item| {
-            crypto::Encryptable::decrypt_via_api(state, item, identifier, key, crypto::GcmAes256)
-        }),
+        crypto::Encryptable::decrypt_via_api(state, inner, identifier, key, crypto::GcmAes256),
         &metrics::DECRYPTION_TIME,
         &metrics::CONTEXT,
         &[],
     )
     .await
-    .transpose()
 }
 
 #[inline]
diff --git a/crates/router/src/core/admin.rs b/crates/router/src/core/admin.rs
index bfefe893977..84c5df6a316 100644
--- a/crates/router/src/core/admin.rs
+++ b/crates/router/src/core/admin.rs
@@ -111,7 +111,10 @@ pub async fn create_merchant_account(
     req: api::MerchantAccountCreate,
 ) -> RouterResponse<api::MerchantAccountResponse> {
     #[cfg(feature = "keymanager_create")]
-    use common_utils::keymanager;
+    use {
+        base64::Engine,
+        common_utils::{keymanager, types::keymanager::EncryptionTransferRequest},
+    };
 
     let db = state.store.as_ref();
 
@@ -124,13 +127,13 @@ pub async fn create_merchant_account(
     let key_manager_state = &(&state).into();
     let merchant_id = req.get_merchant_reference_id().get_string_repr().to_owned();
     let identifier = km_types::Identifier::Merchant(merchant_id.clone());
-
     #[cfg(feature = "keymanager_create")]
     {
-        keymanager::create_key_in_key_manager(
+        keymanager::transfer_key_to_key_manager(
             key_manager_state,
-            km_types::EncryptionCreateRequest {
+            EncryptionTransferRequest {
                 identifier: identifier.clone(),
+                key: consts::BASE64_ENGINE.encode(key),
             },
         )
         .await
diff --git a/crates/router/src/core/customers.rs b/crates/router/src/core/customers.rs
index e9126200acc..c0dc969db10 100644
--- a/crates/router/src/core/customers.rs
+++ b/crates/router/src/core/customers.rs
@@ -1,11 +1,12 @@
 use api_models::customers::CustomerRequestWithEmail;
 use common_utils::{
-    crypto::{Encryptable, GcmAes256},
+    crypto::Encryptable,
     errors::ReportSwitchExt,
     ext_traits::OptionExt,
     types::keymanager::{Identifier, ToEncryptable},
 };
 use error_stack::{report, ResultExt};
+use hyperswitch_domain_models::type_encryption::encrypt;
 use masking::{Secret, SwitchStrategy};
 use router_env::{instrument, tracing};
 
@@ -19,10 +20,7 @@ use crate::{
     services,
     types::{
         api::customers,
-        domain::{
-            self,
-            types::{self, TypeEncryption},
-        },
+        domain::{self, types},
         storage::{self, enums},
     },
     utils::CustomerAddress,
@@ -277,12 +275,11 @@ pub async fn delete_customer(
 
     let key = key_store.key.get_inner().peek();
     let identifier = Identifier::Merchant(key_store.merchant_id.clone());
-    let redacted_encrypted_value: Encryptable<Secret<_>> = Encryptable::encrypt_via_api(
+    let redacted_encrypted_value: Encryptable<Secret<_>> = encrypt(
         key_manager_state,
         REDACTED.to_string().into(),
         identifier.clone(),
         key,
-        GcmAes256,
     )
     .await
     .switch()?;
@@ -336,12 +333,11 @@ pub async fn delete_customer(
     let updated_customer = storage::CustomerUpdate::Update {
         name: Some(redacted_encrypted_value.clone()),
         email: Some(
-            Encryptable::encrypt_via_api(
+            encrypt(
                 key_manager_state,
                 REDACTED.to_string().into(),
                 identifier,
                 key,
-                GcmAes256,
             )
             .await
             .switch()?,
diff --git a/crates/router/src/core/payment_methods/cards.rs b/crates/router/src/core/payment_methods/cards.rs
index 431826fbdb0..e96f2da158c 100644
--- a/crates/router/src/core/payment_methods/cards.rs
+++ b/crates/router/src/core/payment_methods/cards.rs
@@ -71,7 +71,7 @@ use crate::{
     services,
     types::{
         api::{self, routing as routing_types, PaymentMethodCreateExt},
-        domain::{self, types::decrypt},
+        domain::{self, types::decrypt_optional},
         storage::{self, enums, PaymentMethodListContext, PaymentTokenData},
         transformers::{ForeignFrom, ForeignTryFrom},
     },
@@ -1209,7 +1209,7 @@ pub async fn update_customer_payment_method(
         }
 
         // Fetch the existing payment method data from db
-        let existing_card_data = decrypt::<serde_json::Value, masking::WithType>(
+        let existing_card_data = decrypt_optional::<serde_json::Value, masking::WithType>(
             &(&state).into(),
             pm.payment_method_data.clone(),
             Identifier::Merchant(key_store.merchant_id.clone()),
@@ -1683,7 +1683,7 @@ pub async fn decode_and_decrypt_locker_data(
         .change_context(errors::VaultError::ResponseDeserializationFailed)
         .attach_printable("Failed to decode hex string into bytes")?;
     // Decrypt
-    decrypt(
+    decrypt_optional(
         &state.into(),
         Some(Encryption::new(decoded_bytes.into())),
         Identifier::Merchant(key_store.merchant_id.clone()),
@@ -4061,12 +4061,16 @@ where
 {
     let key = key_store.key.get_inner().peek();
     let identifier = Identifier::Merchant(key_store.merchant_id.clone());
-    let decrypted_data =
-        decrypt::<serde_json::Value, masking::WithType>(&state.into(), data, identifier, key)
-            .await
-            .change_context(errors::StorageError::DecryptionError)
-            .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("unable to decrypt data")?;
+    let decrypted_data = decrypt_optional::<serde_json::Value, masking::WithType>(
+        &state.into(),
+        data,
+        identifier,
+        key,
+    )
+    .await
+    .change_context(errors::StorageError::DecryptionError)
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .attach_printable("unable to decrypt data")?;
 
     decrypted_data
         .map(|decrypted_data| decrypted_data.into_inner().expose())
@@ -4083,7 +4087,7 @@ pub async fn get_card_details_with_locker_fallback(
 ) -> errors::RouterResult<Option<api::CardDetailFromLocker>> {
     let key = key_store.key.get_inner().peek();
     let identifier = Identifier::Merchant(key_store.merchant_id.clone());
-    let card_decrypted = decrypt::<serde_json::Value, masking::WithType>(
+    let card_decrypted = decrypt_optional::<serde_json::Value, masking::WithType>(
         &state.into(),
         pm.payment_method_data.clone(),
         identifier,
@@ -4119,7 +4123,7 @@ pub async fn get_card_details_without_locker_fallback(
 ) -> errors::RouterResult<api::CardDetailFromLocker> {
     let key = key_store.key.get_inner().peek();
     let identifier = Identifier::Merchant(key_store.merchant_id.clone());
-    let card_decrypted = decrypt::<serde_json::Value, masking::WithType>(
+    let card_decrypted = decrypt_optional::<serde_json::Value, masking::WithType>(
         &state.into(),
         pm.payment_method_data.clone(),
         identifier,
@@ -4194,7 +4198,7 @@ async fn get_masked_bank_details(
 ) -> errors::RouterResult<Option<MaskedBankDetails>> {
     let key = key_store.key.get_inner().peek();
     let identifier = Identifier::Merchant(key_store.merchant_id.clone());
-    let payment_method_data = decrypt::<serde_json::Value, masking::WithType>(
+    let payment_method_data = decrypt_optional::<serde_json::Value, masking::WithType>(
         &state.into(),
         pm.payment_method_data.clone(),
         identifier,
@@ -4234,7 +4238,7 @@ async fn get_bank_account_connector_details(
 ) -> errors::RouterResult<Option<BankAccountTokenData>> {
     let key = key_store.key.get_inner().peek();
     let identifier = Identifier::Merchant(key_store.merchant_id.clone());
-    let payment_method_data = decrypt::<serde_json::Value, masking::WithType>(
+    let payment_method_data = decrypt_optional::<serde_json::Value, masking::WithType>(
         &state.into(),
         pm.payment_method_data.clone(),
         identifier,
diff --git a/crates/router/src/core/payments/operations/payment_confirm.rs b/crates/router/src/core/payments/operations/payment_confirm.rs
index addd2f30670..ad3f8fb013e 100644
--- a/crates/router/src/core/payments/operations/payment_confirm.rs
+++ b/crates/router/src/core/payments/operations/payment_confirm.rs
@@ -39,7 +39,7 @@ use crate::{
     types::{
         self,
         api::{self, ConnectorCallType, PaymentIdTypeExt},
-        domain::{self, types::decrypt},
+        domain::{self, types::decrypt_optional},
         storage::{self, enums as storage_enums},
     },
     utils::{self, OptionExt},
@@ -1084,7 +1084,7 @@ impl<F: Clone> UpdateTracker<F, PaymentData<F>, api::PaymentsRequest> for Paymen
             let key = key_store.key.get_inner().peek();
 
             let card_detail_from_locker: Option<api::CardDetailFromLocker> =
-                decrypt::<serde_json::Value, masking::WithType>(
+                decrypt_optional::<serde_json::Value, masking::WithType>(
                     key_manager_state,
                     pm.payment_method_data.clone(),
                     Identifier::Merchant(key_store.merchant_id.clone()),
diff --git a/crates/router/src/core/payments/operations/payment_create.rs b/crates/router/src/core/payments/operations/payment_create.rs
index 929f404eae9..f264a056f6c 100644
--- a/crates/router/src/core/payments/operations/payment_create.rs
+++ b/crates/router/src/core/payments/operations/payment_create.rs
@@ -13,7 +13,7 @@ use error_stack::{self, ResultExt};
 use hyperswitch_domain_models::{
     mandates::{MandateData, MandateDetails},
     payments::{payment_attempt::PaymentAttempt, payment_intent::CustomerData},
-    type_encryption::decrypt,
+    type_encryption::decrypt_optional,
 };
 use masking::{ExposeInterface, PeekInterface, Secret};
 use router_derive::PaymentOperation;
@@ -847,7 +847,7 @@ impl PaymentCreate {
             additional_pm_data = payment_method_info
                 .as_ref()
                 .async_map(|pm_info| async {
-                    decrypt::<serde_json::Value, masking::WithType>(
+                    decrypt_optional::<serde_json::Value, masking::WithType>(
                         &state.into(),
                         pm_info.payment_method_data.clone(),
                         Identifier::Merchant(key_store.merchant_id.clone()),
diff --git a/crates/router/src/core/pm_auth.rs b/crates/router/src/core/pm_auth.rs
index d6f5ffd9fb9..df28d257b36 100644
--- a/crates/router/src/core/pm_auth.rs
+++ b/crates/router/src/core/pm_auth.rs
@@ -45,7 +45,7 @@ use crate::{
     services::{pm_auth as pm_auth_services, ApplicationResponse},
     types::{
         self,
-        domain::{self, types::decrypt},
+        domain::{self, types::decrypt_optional},
         storage,
         transformers::ForeignTryFrom,
     },
@@ -327,7 +327,7 @@ async fn store_bank_details_in_payment_methods(
 
     for pm in payment_methods {
         if pm.payment_method == Some(enums::PaymentMethod::BankDebit) {
-            let bank_details_pm_data = decrypt::<serde_json::Value, masking::WithType>(
+            let bank_details_pm_data = decrypt_optional::<serde_json::Value, masking::WithType>(
                 &(&state).into(),
                 pm.payment_method_data.clone(),
                 Identifier::Merchant(key_store.merchant_id.clone()),
diff --git a/crates/router/src/core/webhooks/outgoing.rs b/crates/router/src/core/webhooks/outgoing.rs
index ed2b41dc64d..7f30a58e208 100644
--- a/crates/router/src/core/webhooks/outgoing.rs
+++ b/crates/router/src/core/webhooks/outgoing.rs
@@ -7,7 +7,7 @@ use api_models::{
 use common_utils::{ext_traits::Encode, request::RequestContent, types::keymanager::Identifier};
 use diesel_models::process_tracker::business_status;
 use error_stack::{report, ResultExt};
-use hyperswitch_domain_models::type_encryption::decrypt;
+use hyperswitch_domain_models::type_encryption::decrypt_optional;
 use masking::{ExposeInterface, Mask, PeekInterface, Secret};
 use router_env::{
     instrument,
@@ -575,7 +575,7 @@ pub(crate) async fn get_outgoing_webhook_request(
 
         let transformed_outgoing_webhook = WebhookType::from(outgoing_webhook);
         let payment_response_hash_key = business_profile.payment_response_hash_key.clone();
-        let custom_headers = decrypt::<serde_json::Value, masking::WithType>(
+        let custom_headers = decrypt_optional::<serde_json::Value, masking::WithType>(
             &state.into(),
             business_profile
                 .outgoing_webhook_custom_http_headers
diff --git a/crates/router/src/types/api/admin.rs b/crates/router/src/types/api/admin.rs
index 054888d1dfe..6cd259f3e7b 100644
--- a/crates/router/src/types/api/admin.rs
+++ b/crates/router/src/types/api/admin.rs
@@ -13,7 +13,9 @@ use common_utils::{
     types::keymanager::Identifier,
 };
 use error_stack::{report, ResultExt};
-use hyperswitch_domain_models::{merchant_key_store::MerchantKeyStore, type_encryption::decrypt};
+use hyperswitch_domain_models::{
+    merchant_key_store::MerchantKeyStore, type_encryption::decrypt_optional,
+};
 use masking::{ExposeInterface, PeekInterface, Secret};
 
 use crate::{
@@ -92,24 +94,25 @@ pub async fn business_profile_response(
     item: storage::business_profile::BusinessProfile,
     key_store: &MerchantKeyStore,
 ) -> Result<BusinessProfileResponse, error_stack::Report<errors::ParsingError>> {
-    let outgoing_webhook_custom_http_headers = decrypt::<serde_json::Value, masking::WithType>(
-        &state.into(),
-        item.outgoing_webhook_custom_http_headers.clone(),
-        Identifier::Merchant(key_store.merchant_id.clone()),
-        key_store.key.get_inner().peek(),
-    )
-    .await
-    .change_context(errors::ParsingError::StructParseFailure(
-        "Outgoing webhook custom HTTP headers",
-    ))
-    .attach_printable("Failed to decrypt outgoing webhook custom HTTP headers")?
-    .map(|decrypted_value| {
-        decrypted_value
-            .into_inner()
-            .expose()
-            .parse_value::<HashMap<String, String>>("HashMap<String,String>")
-    })
-    .transpose()?;
+    let outgoing_webhook_custom_http_headers =
+        decrypt_optional::<serde_json::Value, masking::WithType>(
+            &state.into(),
+            item.outgoing_webhook_custom_http_headers.clone(),
+            Identifier::Merchant(key_store.merchant_id.clone()),
+            key_store.key.get_inner().peek(),
+        )
+        .await
+        .change_context(errors::ParsingError::StructParseFailure(
+            "Outgoing webhook custom HTTP headers",
+        ))
+        .attach_printable("Failed to decrypt outgoing webhook custom HTTP headers")?
+        .map(|decrypted_value| {
+            decrypted_value
+                .into_inner()
+                .expose()
+                .parse_value::<HashMap<String, String>>("HashMap<String,String>")
+        })
+        .transpose()?;
 
     Ok(BusinessProfileResponse {
         merchant_id: item.merchant_id,
diff --git a/crates/router/src/types/domain/merchant_connector_account.rs b/crates/router/src/types/domain/merchant_connector_account.rs
index 6f654848f60..bd0bfe15e53 100644
--- a/crates/router/src/types/domain/merchant_connector_account.rs
+++ b/crates/router/src/types/domain/merchant_connector_account.rs
@@ -1,5 +1,5 @@
 use common_utils::{
-    crypto::{Encryptable, GcmAes256},
+    crypto::Encryptable,
     date_time,
     encryption::Encryption,
     errors::{CustomResult, ValidationError},
@@ -12,7 +12,7 @@ use masking::{PeekInterface, Secret};
 
 use super::{
     behaviour,
-    types::{self, AsyncLift, TypeEncryption},
+    types::{decrypt, decrypt_optional, AsyncLift},
 };
 #[derive(Clone, Debug)]
 pub struct MerchantConnectorAccount {
@@ -117,12 +117,11 @@ impl behaviour::Conversion for MerchantConnectorAccount {
             id: Some(other.id),
             merchant_id: other.merchant_id,
             connector_name: other.connector_name,
-            connector_account_details: Encryptable::decrypt_via_api(
+            connector_account_details: decrypt(
                 state,
                 other.connector_account_details,
                 identifier.clone(),
                 key.peek(),
-                GcmAes256,
             )
             .await
             .change_context(ValidationError::InvalidValue {
@@ -149,14 +148,14 @@ impl behaviour::Conversion for MerchantConnectorAccount {
             status: other.status,
             connector_wallets_details: other
                 .connector_wallets_details
-                .async_lift(|inner| types::decrypt(state, inner, identifier.clone(), key.peek()))
+                .async_lift(|inner| decrypt_optional(state, inner, identifier.clone(), key.peek()))
                 .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(
-                    Encryptable::decrypt(data, key.peek(), GcmAes256)
+                    decrypt(state, data, identifier, key.peek())
                         .await
                         .change_context(ValidationError::InvalidValue {
                             message: "Failed while decrypting additional_merchant_data".to_string(),
diff --git a/crates/router/src/types/domain/types.rs b/crates/router/src/types/domain/types.rs
index 53239cd27cf..fa9053a096f 100644
--- a/crates/router/src/types/domain/types.rs
+++ b/crates/router/src/types/domain/types.rs
@@ -1,7 +1,7 @@
 use common_utils::types::keymanager::KeyManagerState;
 pub use hyperswitch_domain_models::type_encryption::{
-    batch_decrypt, batch_encrypt, decrypt, encrypt, encrypt_optional, AsyncLift, Lift,
-    TypeEncryption,
+    batch_decrypt, batch_encrypt, decrypt, decrypt_optional, encrypt, encrypt_optional, AsyncLift,
+    Lift,
 };
 
 impl From<&crate::SessionState> for KeyManagerState {
diff --git a/crates/router/src/types/domain/user.rs b/crates/router/src/types/domain/user.rs
index ec44103a8f0..6988cdf6136 100644
--- a/crates/router/src/types/domain/user.rs
+++ b/crates/router/src/types/domain/user.rs
@@ -6,8 +6,6 @@ use api_models::{
 use common_enums::TokenPurpose;
 #[cfg(not(feature = "v2"))]
 use common_utils::id_type;
-#[cfg(feature = "keymanager_create")]
-use common_utils::types::keymanager::EncryptionCreateRequest;
 use common_utils::{
     crypto::Encryptable, errors::CustomResult, new_type::MerchantName, pii,
     types::keymanager::Identifier,
@@ -25,6 +23,8 @@ use once_cell::sync::Lazy;
 use rand::distributions::{Alphanumeric, DistString};
 use router_env::env;
 use unicode_segmentation::UnicodeSegmentation;
+#[cfg(feature = "keymanager_create")]
+use {base64::Engine, common_utils::types::keymanager::EncryptionTransferRequest};
 
 use crate::{
     consts,
@@ -975,6 +975,19 @@ impl UserFromStorage {
                 .change_context(UserErrors::InternalServerError)
                 .attach_printable("Unable to generate aes 256 key")?;
 
+            #[cfg(feature = "keymanager_create")]
+            {
+                common_utils::keymanager::transfer_key_to_key_manager(
+                    key_manager_state,
+                    EncryptionTransferRequest {
+                        identifier: Identifier::User(self.get_user_id().to_string()),
+                        key: consts::BASE64_ENGINE.encode(key),
+                    },
+                )
+                .await
+                .change_context(UserErrors::InternalServerError)?;
+            }
+
             let key_store = UserKeyStore {
                 user_id: self.get_user_id().to_string(),
                 key: domain_types::encrypt(
@@ -988,18 +1001,6 @@ impl UserFromStorage {
                 created_at: common_utils::date_time::now(),
             };
 
-            #[cfg(feature = "keymanager_create")]
-            {
-                common_utils::keymanager::create_key_in_key_manager(
-                    key_manager_state,
-                    EncryptionCreateRequest {
-                        identifier: Identifier::User(key_store.user_id.clone()),
-                    },
-                )
-                .await
-                .change_context(UserErrors::InternalServerError)?;
-            }
-
             state
                 .global_store
                 .insert_user_key_store(key_manager_state, key_store, &master_key.to_vec().into())
@@ -1039,7 +1040,7 @@ impl UserFromStorage {
             .await
             .change_context(UserErrors::InternalServerError)?;
 
-        Ok(domain_types::decrypt::<String, masking::WithType>(
+        Ok(domain_types::decrypt_optional::<String, masking::WithType>(
             key_manager_state,
             self.0.totp_secret.clone(),
             Identifier::User(user_key_store.user_id.clone()),
diff --git a/crates/router/src/types/domain/user_key_store.rs b/crates/router/src/types/domain/user_key_store.rs
index 3a1a9a60e95..40c8acd5695 100644
--- a/crates/router/src/types/domain/user_key_store.rs
+++ b/crates/router/src/types/domain/user_key_store.rs
@@ -1,16 +1,14 @@
 use common_utils::{
-    crypto::{Encryptable, GcmAes256},
+    crypto::Encryptable,
     date_time,
     types::keymanager::{Identifier, KeyManagerState},
 };
 use error_stack::ResultExt;
+use hyperswitch_domain_models::type_encryption::decrypt;
 use masking::{PeekInterface, Secret};
 use time::PrimitiveDateTime;
 
-use crate::{
-    errors::{CustomResult, ValidationError},
-    types::domain::types::TypeEncryption,
-};
+use crate::errors::{CustomResult, ValidationError};
 
 #[derive(Clone, Debug, serde::Serialize)]
 pub struct UserKeyStore {
@@ -43,7 +41,7 @@ impl super::behaviour::Conversion for UserKeyStore {
     {
         let identifier = Identifier::User(item.user_id.clone());
         Ok(Self {
-            key: Encryptable::decrypt_via_api(state, item.key, identifier, key.peek(), GcmAes256)
+            key: decrypt(state, item.key, identifier, key.peek())
                 .await
                 .change_context(ValidationError::InvalidValue {
                     message: "Failed while decrypting customer data".to_string(),
diff --git a/crates/router/src/utils/user.rs b/crates/router/src/utils/user.rs
index 9734122a260..ccb9ed7d37b 100644
--- a/crates/router/src/utils/user.rs
+++ b/crates/router/src/utils/user.rs
@@ -283,7 +283,7 @@ pub async fn decrypt_oidc_private_config(
     .change_context(UserErrors::InternalServerError)
     .attach_printable("Failed to decode DEK")?;
 
-    let private_config = domain::types::decrypt::<serde_json::Value, masking::WithType>(
+    let private_config = domain::types::decrypt_optional::<serde_json::Value, masking::WithType>(
         &state.into(),
         encrypted_config,
         Identifier::UserAuth(id),

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

- [x] Bugfix
- [ ] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
encrypt/decrypt should happen only via encryption service and DEK should be same in application and encryption service


### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
encrypt/decrypt should happen only via encryption service and DEK should be same in application and encryption service


## How did you test it?
Changes can be verified only through logs. Look for "Fall back to Application Encryption" or "Fall back to Application Decryption" in the logs post deployment after running below tests. If logs are present encryption/decryption failed with encryption service and fall back to the application encryption which has to be investigated.

Test with merchant created before the changes deployed
1. Create Payment
2. Retrieve Payment
3. Create Refund
4. Retrieve Payment
Test with merchant created after the changes deployed
1. Create Merchant
2. Create API Key
3. Create MCA
4. Create Payment
6. Retrieve Payment
7. Create Refund
8. Retrieve Payment
Requires sanity on all the flows


## 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
- [ ] I added unit tests for my changes where possible
