Issue:
Bug: [fix] add  offset and limit parameters to key transfer API





Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->
- [x] Enhancement

## Description
<!-- Describe your changes in detail -->
Add offset and limit to key transfer API

## 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).
-->
This adds `from` and `limit` parameters to Key transfer API and thus providing a way to transfer the keys in batches.

## 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. Run `/accounts/transfer` with `from` and `limit` parameters. It should respond with the `total_transferred` same as `limit`.

```bash
curl --location 'http://localhost:8080/accounts/transfer' \
--header 'Content-Type: application/json' \
--header 'api-key: test_admin' \
--data '{
    "from": 0,
    "limit": 2
}'
```

2. Do the same thing for `user/key/transfer`
```bash
curl --location 'http://localhost:8080/user/key/transfer' \
--header 'Content-Type: application/json' \
--header 'api-key: test_admin' \
--data '{
    "from": 0,
    "limit": 2
}'
````

## 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

Testing:

1. Run `/accounts/transfer` with `from` and `limit` parameters. It should respond with the `total_transferred` same as `limit`.

```bash
curl --location 'http://localhost:8080/accounts/transfer' \
--header 'Content-Type: application/json' \
--header 'api-key: test_admin' \
--data '{
    "from": 0,
    "limit": 2
}'
```

2. Do the same thing for `user/key/transfer`
```bash
curl --location 'http://localhost:8080/user/key/transfer' \
--header 'Content-Type: application/json' \
--header 'api-key: test_admin' \
--data '{
    "from": 0,
    "limit": 2
}'
````

Patch:
diff --git a/crates/analytics/src/types.rs b/crates/analytics/src/types.rs
index d2d3604661f..6593e5d6eb0 100644
--- a/crates/analytics/src/types.rs
+++ b/crates/analytics/src/types.rs
@@ -3,7 +3,7 @@ use std::{fmt::Display, str::FromStr};
 use common_utils::{
     errors::{CustomResult, ErrorSwitch, ParsingError},
     events::{ApiEventMetric, ApiEventsType},
-    impl_misc_api_event_type,
+    impl_api_event_type,
 };
 use error_stack::{report, Report, ResultExt};
 
@@ -150,4 +150,4 @@ impl ErrorSwitch<AnalyticsError> for FiltersError {
     }
 }
 
-impl_misc_api_event_type!(AnalyticsDomain);
+impl_api_event_type!(Miscellaneous, (AnalyticsDomain));
diff --git a/crates/api_models/src/admin.rs b/crates/api_models/src/admin.rs
index 853f4308b17..3e90a48ff0b 100644
--- a/crates/api_models/src/admin.rs
+++ b/crates/api_models/src/admin.rs
@@ -1053,6 +1053,16 @@ pub struct ToggleKVResponse {
     pub kv_enabled: bool,
 }
 
+#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize, ToSchema)]
+pub struct MerchantKeyTransferRequest {
+    /// Offset for merchant account
+    #[schema(example = 32)]
+    pub from: u32,
+    /// Limit for merchant account
+    #[schema(example = 32)]
+    pub limit: u32,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct TransferKeyResponse {
     /// The identifier for the Merchant Account
diff --git a/crates/api_models/src/events.rs b/crates/api_models/src/events.rs
index 3757dd98d78..32cdd83eb9f 100644
--- a/crates/api_models/src/events.rs
+++ b/crates/api_models/src/events.rs
@@ -16,7 +16,7 @@ pub mod user_role;
 
 use common_utils::{
     events::{ApiEventMetric, ApiEventsType},
-    impl_misc_api_event_type,
+    impl_api_event_type,
 };
 
 #[allow(unused_imports)]
@@ -33,6 +33,7 @@ use crate::{
     mandates::*,
     payment_methods::*,
     payments::*,
+    user::{UserKeyTransferRequest, UserTransferKeyResponse},
     verifications::*,
 };
 
@@ -56,87 +57,102 @@ impl ApiEventMetric for PaymentIntentFiltersResponse {
     }
 }
 
-impl_misc_api_event_type!(
-    PaymentMethodId,
-    PaymentMethodCreate,
-    PaymentLinkInitiateRequest,
-    RetrievePaymentLinkResponse,
-    MandateListConstraints,
-    CreateFileResponse,
-    MerchantConnectorResponse,
-    MerchantConnectorId,
-    MandateResponse,
-    MandateRevokedResponse,
-    RetrievePaymentLinkRequest,
-    PaymentLinkListConstraints,
-    MandateId,
-    DisputeListConstraints,
-    RetrieveApiKeyResponse,
-    BusinessProfileResponse,
-    BusinessProfileUpdate,
-    BusinessProfileCreate,
-    RevokeApiKeyResponse,
-    ToggleKVResponse,
-    ToggleKVRequest,
-    ToggleAllKVRequest,
-    ToggleAllKVResponse,
-    TransferKeyResponse,
-    MerchantAccountDeleteResponse,
-    MerchantAccountUpdate,
-    CardInfoResponse,
-    CreateApiKeyResponse,
-    CreateApiKeyRequest,
-    MerchantConnectorDeleteResponse,
-    MerchantConnectorUpdate,
-    MerchantConnectorCreate,
-    MerchantId,
-    CardsInfoRequest,
-    MerchantAccountResponse,
-    MerchantAccountListRequest,
-    MerchantAccountCreate,
-    PaymentsSessionRequest,
-    ApplepayMerchantVerificationRequest,
-    ApplepayMerchantResponse,
-    ApplepayVerifiedDomainsResponse,
-    UpdateApiKeyRequest,
-    GetApiEventFiltersRequest,
-    ApiEventFiltersResponse,
-    GetInfoResponse,
-    GetPaymentMetricRequest,
-    GetRefundMetricRequest,
-    GetActivePaymentsMetricRequest,
-    GetSdkEventMetricRequest,
-    GetAuthEventMetricRequest,
-    GetPaymentFiltersRequest,
-    PaymentFiltersResponse,
-    GetRefundFilterRequest,
-    RefundFiltersResponse,
-    GetSdkEventFiltersRequest,
-    SdkEventFiltersResponse,
-    ApiLogsRequest,
-    GetApiEventMetricRequest,
-    SdkEventsRequest,
-    ReportRequest,
-    ConnectorEventsRequest,
-    OutgoingWebhookLogsRequest,
-    GetGlobalSearchRequest,
-    GetSearchRequest,
-    GetSearchResponse,
-    GetSearchRequestWithIndex,
-    GetDisputeFilterRequest,
-    DisputeFiltersResponse,
-    GetDisputeMetricRequest
+impl_api_event_type!(
+    Miscellaneous,
+    (
+        PaymentMethodId,
+        PaymentMethodCreate,
+        PaymentLinkInitiateRequest,
+        RetrievePaymentLinkResponse,
+        MandateListConstraints,
+        CreateFileResponse,
+        MerchantConnectorResponse,
+        MerchantConnectorId,
+        MandateResponse,
+        MandateRevokedResponse,
+        RetrievePaymentLinkRequest,
+        PaymentLinkListConstraints,
+        MandateId,
+        DisputeListConstraints,
+        RetrieveApiKeyResponse,
+        BusinessProfileResponse,
+        BusinessProfileUpdate,
+        BusinessProfileCreate,
+        RevokeApiKeyResponse,
+        ToggleKVResponse,
+        ToggleKVRequest,
+        ToggleAllKVRequest,
+        ToggleAllKVResponse,
+        MerchantAccountDeleteResponse,
+        MerchantAccountUpdate,
+        CardInfoResponse,
+        CreateApiKeyResponse,
+        CreateApiKeyRequest,
+        MerchantConnectorDeleteResponse,
+        MerchantConnectorUpdate,
+        MerchantConnectorCreate,
+        MerchantId,
+        CardsInfoRequest,
+        MerchantAccountResponse,
+        MerchantAccountListRequest,
+        MerchantAccountCreate,
+        PaymentsSessionRequest,
+        ApplepayMerchantVerificationRequest,
+        ApplepayMerchantResponse,
+        ApplepayVerifiedDomainsResponse,
+        UpdateApiKeyRequest,
+        GetApiEventFiltersRequest,
+        ApiEventFiltersResponse,
+        GetInfoResponse,
+        GetPaymentMetricRequest,
+        GetRefundMetricRequest,
+        GetActivePaymentsMetricRequest,
+        GetSdkEventMetricRequest,
+        GetAuthEventMetricRequest,
+        GetPaymentFiltersRequest,
+        PaymentFiltersResponse,
+        GetRefundFilterRequest,
+        RefundFiltersResponse,
+        GetSdkEventFiltersRequest,
+        SdkEventFiltersResponse,
+        ApiLogsRequest,
+        GetApiEventMetricRequest,
+        SdkEventsRequest,
+        ReportRequest,
+        ConnectorEventsRequest,
+        OutgoingWebhookLogsRequest,
+        GetGlobalSearchRequest,
+        GetSearchRequest,
+        GetSearchResponse,
+        GetSearchRequestWithIndex,
+        GetDisputeFilterRequest,
+        DisputeFiltersResponse,
+        GetDisputeMetricRequest
+    )
+);
+
+impl_api_event_type!(
+    Keymanager,
+    (
+        TransferKeyResponse,
+        MerchantKeyTransferRequest,
+        UserKeyTransferRequest,
+        UserTransferKeyResponse
+    )
 );
 
 #[cfg(feature = "stripe")]
-impl_misc_api_event_type!(
-    StripeSetupIntentResponse,
-    StripeRefundResponse,
-    StripePaymentIntentListResponse,
-    StripePaymentIntentResponse,
-    CustomerDeleteResponse,
-    CustomerPaymentMethodListResponse,
-    CreateCustomerResponse
+impl_api_event_type!(
+    Miscellaneous,
+    (
+        StripeSetupIntentResponse,
+        StripeRefundResponse,
+        StripePaymentIntentListResponse,
+        StripePaymentIntentResponse,
+        CustomerDeleteResponse,
+        CustomerPaymentMethodListResponse,
+        CreateCustomerResponse
+    )
 );
 
 impl<T> ApiEventMetric for MetricsResponse<T> {
diff --git a/crates/api_models/src/events/connector_onboarding.rs b/crates/api_models/src/events/connector_onboarding.rs
index 0da89f61da7..c7fc2cd7de7 100644
--- a/crates/api_models/src/events/connector_onboarding.rs
+++ b/crates/api_models/src/events/connector_onboarding.rs
@@ -5,10 +5,13 @@ use crate::connector_onboarding::{
     ResetTrackingIdRequest,
 };
 
-common_utils::impl_misc_api_event_type!(
-    ActionUrlRequest,
-    ActionUrlResponse,
-    OnboardingSyncRequest,
-    OnboardingStatus,
-    ResetTrackingIdRequest
+common_utils::impl_api_event_type!(
+    Miscellaneous,
+    (
+        ActionUrlRequest,
+        ActionUrlResponse,
+        OnboardingSyncRequest,
+        OnboardingStatus,
+        ResetTrackingIdRequest
+    )
 );
diff --git a/crates/api_models/src/events/user.rs b/crates/api_models/src/events/user.rs
index b8c893ae25d..862c361d465 100644
--- a/crates/api_models/src/events/user.rs
+++ b/crates/api_models/src/events/user.rs
@@ -20,7 +20,7 @@ use crate::user::{
     SsoSignInRequest, SwitchMerchantIdRequest, TokenOrPayloadResponse, TokenResponse,
     TwoFactorAuthStatusResponse, UpdateUserAccountDetailsRequest,
     UpdateUserAuthenticationMethodRequest, UserFromEmailRequest, UserMerchantCreate,
-    UserTransferKeyResponse, VerifyEmailRequest, VerifyRecoveryCodeRequest, VerifyTotpRequest,
+    VerifyEmailRequest, VerifyRecoveryCodeRequest, VerifyTotpRequest,
 };
 
 impl ApiEventMetric for DashboardEntryResponse {
@@ -46,48 +46,50 @@ impl<T> ApiEventMetric for TokenOrPayloadResponse<T> {
     }
 }
 
-common_utils::impl_misc_api_event_type!(
-    SignUpRequest,
-    SignUpWithMerchantIdRequest,
-    ChangePasswordRequest,
-    GetMultipleMetaDataPayload,
-    GetMetaDataResponse,
-    GetMetaDataRequest,
-    SetMetaDataRequest,
-    SwitchMerchantIdRequest,
-    CreateInternalUserRequest,
-    UserMerchantCreate,
-    ListUsersResponse,
-    AuthorizeResponse,
-    ConnectAccountRequest,
-    ForgotPasswordRequest,
-    ResetPasswordRequest,
-    RotatePasswordRequest,
-    InviteUserRequest,
-    ReInviteUserRequest,
-    VerifyEmailRequest,
-    SendVerifyEmailRequest,
-    AcceptInviteFromEmailRequest,
-    SignInResponse,
-    UpdateUserAccountDetailsRequest,
-    GetUserDetailsResponse,
-    GetUserRoleDetailsRequest,
-    GetUserRoleDetailsResponse,
-    TokenResponse,
-    TwoFactorAuthStatusResponse,
-    UserFromEmailRequest,
-    BeginTotpResponse,
-    VerifyRecoveryCodeRequest,
-    VerifyTotpRequest,
-    RecoveryCodes,
-    GetUserAuthenticationMethodsRequest,
-    CreateUserAuthenticationMethodRequest,
-    UpdateUserAuthenticationMethodRequest,
-    GetSsoAuthUrlRequest,
-    SsoSignInRequest,
-    UserTransferKeyResponse,
-    AuthSelectRequest
+common_utils::impl_api_event_type!(
+    Miscellaneous,
+    (
+        SignUpRequest,
+        SignUpWithMerchantIdRequest,
+        ChangePasswordRequest,
+        GetMultipleMetaDataPayload,
+        GetMetaDataResponse,
+        GetMetaDataRequest,
+        SetMetaDataRequest,
+        SwitchMerchantIdRequest,
+        CreateInternalUserRequest,
+        UserMerchantCreate,
+        ListUsersResponse,
+        AuthorizeResponse,
+        ConnectAccountRequest,
+        ForgotPasswordRequest,
+        ResetPasswordRequest,
+        RotatePasswordRequest,
+        InviteUserRequest,
+        ReInviteUserRequest,
+        VerifyEmailRequest,
+        SendVerifyEmailRequest,
+        AcceptInviteFromEmailRequest,
+        SignInResponse,
+        UpdateUserAccountDetailsRequest,
+        GetUserDetailsResponse,
+        GetUserRoleDetailsRequest,
+        GetUserRoleDetailsResponse,
+        TokenResponse,
+        TwoFactorAuthStatusResponse,
+        UserFromEmailRequest,
+        BeginTotpResponse,
+        VerifyRecoveryCodeRequest,
+        VerifyTotpRequest,
+        RecoveryCodes,
+        GetUserAuthenticationMethodsRequest,
+        CreateUserAuthenticationMethodRequest,
+        UpdateUserAuthenticationMethodRequest,
+        GetSsoAuthUrlRequest,
+        SsoSignInRequest,
+        AuthSelectRequest
+    )
 );
 
 #[cfg(feature = "dummy_connector")]
-common_utils::impl_misc_api_event_type!(SampleDataRequest);
+common_utils::impl_api_event_type!(Miscellaneous, (SampleDataRequest));
diff --git a/crates/api_models/src/events/user_role.rs b/crates/api_models/src/events/user_role.rs
index 34375a22848..7e8eaf17e30 100644
--- a/crates/api_models/src/events/user_role.rs
+++ b/crates/api_models/src/events/user_role.rs
@@ -10,19 +10,22 @@ use crate::user_role::{
     MerchantSelectRequest, TransferOrgOwnershipRequest, UpdateUserRoleRequest,
 };
 
-common_utils::impl_misc_api_event_type!(
-    RoleInfoWithPermissionsResponse,
-    GetRoleRequest,
-    AuthorizationInfoResponse,
-    UpdateUserRoleRequest,
-    MerchantSelectRequest,
-    AcceptInvitationRequest,
-    DeleteUserRoleRequest,
-    TransferOrgOwnershipRequest,
-    CreateRoleRequest,
-    UpdateRoleRequest,
-    ListRolesResponse,
-    RoleInfoResponse,
-    GetRoleFromTokenResponse,
-    RoleInfoWithGroupsResponse
+common_utils::impl_api_event_type!(
+    Miscellaneous,
+    (
+        RoleInfoWithPermissionsResponse,
+        GetRoleRequest,
+        AuthorizationInfoResponse,
+        UpdateUserRoleRequest,
+        MerchantSelectRequest,
+        AcceptInvitationRequest,
+        DeleteUserRoleRequest,
+        TransferOrgOwnershipRequest,
+        CreateRoleRequest,
+        UpdateRoleRequest,
+        ListRolesResponse,
+        RoleInfoResponse,
+        GetRoleFromTokenResponse,
+        RoleInfoWithGroupsResponse
+    )
 );
diff --git a/crates/api_models/src/pm_auth.rs b/crates/api_models/src/pm_auth.rs
index 7044bd8d335..4a1c8eaa31b 100644
--- a/crates/api_models/src/pm_auth.rs
+++ b/crates/api_models/src/pm_auth.rs
@@ -1,7 +1,7 @@
 use common_enums::{PaymentMethod, PaymentMethodType};
 use common_utils::{
     events::{ApiEventMetric, ApiEventsType},
-    impl_misc_api_event_type,
+    impl_api_event_type,
 };
 
 #[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
@@ -49,9 +49,12 @@ pub struct PaymentMethodAuthConnectorChoice {
     pub mca_id: String,
 }
 
-impl_misc_api_event_type!(
-    LinkTokenCreateRequest,
-    LinkTokenCreateResponse,
-    ExchangeTokenCreateRequest,
-    ExchangeTokenCreateResponse
+impl_api_event_type!(
+    Miscellaneous,
+    (
+        LinkTokenCreateRequest,
+        LinkTokenCreateResponse,
+        ExchangeTokenCreateRequest,
+        ExchangeTokenCreateResponse
+    )
 );
diff --git a/crates/api_models/src/user.rs b/crates/api_models/src/user.rs
index 7c0b21f9679..2f3d29c0b89 100644
--- a/crates/api_models/src/user.rs
+++ b/crates/api_models/src/user.rs
@@ -380,6 +380,12 @@ pub struct AuthSelectRequest {
     pub id: Option<String>,
 }
 
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub struct UserKeyTransferRequest {
+    pub from: u32,
+    pub limit: u32,
+}
+
 #[derive(Debug, serde::Deserialize, serde::Serialize)]
 pub struct UserTransferKeyResponse {
     pub total_transferred: usize,
diff --git a/crates/api_models/src/verify_connector.rs b/crates/api_models/src/verify_connector.rs
index 1db5a19a030..0e415c35229 100644
--- a/crates/api_models/src/verify_connector.rs
+++ b/crates/api_models/src/verify_connector.rs
@@ -8,4 +8,4 @@ pub struct VerifyConnectorRequest {
     pub connector_account_details: admin::ConnectorAuthType,
 }
 
-common_utils::impl_misc_api_event_type!(VerifyConnectorRequest);
+common_utils::impl_api_event_type!(Miscellaneous, (VerifyConnectorRequest));
diff --git a/crates/common_utils/src/events.rs b/crates/common_utils/src/events.rs
index 3e3a0da4cab..b2420709a77 100644
--- a/crates/common_utils/src/events.rs
+++ b/crates/common_utils/src/events.rs
@@ -49,6 +49,7 @@ pub enum ApiEventsType {
     Gsm,
     // TODO: This has to be removed once the corresponding apiEventTypes are created
     Miscellaneous,
+    Keymanager,
     RustLocker,
     ApplePayCertificatesMigration,
     FraudCheck,
@@ -88,23 +89,26 @@ impl<T> ApiEventMetric for Vec<T> {
 }
 
 #[macro_export]
-macro_rules! impl_misc_api_event_type {
-    ($($type:ty),+) => {
+macro_rules! impl_api_event_type {
+    ($event: ident, ($($type:ty),+))=> {
         $(
             impl ApiEventMetric for $type {
                 fn get_api_event_type(&self) -> Option<ApiEventsType> {
-                    Some(ApiEventsType::Miscellaneous)
+                    Some(ApiEventsType::$event)
                 }
             }
         )+
      };
 }
 
-impl_misc_api_event_type!(
-    String,
-    (&String, &String),
-    (Option<i64>, Option<i64>, String),
-    bool
+impl_api_event_type!(
+    Miscellaneous,
+    (
+        String,
+        (&String, &String),
+        (Option<i64>, Option<i64>, String),
+        bool
+    )
 );
 
 impl<T: ApiEventMetric> ApiEventMetric for &T {
diff --git a/crates/diesel_models/src/query/merchant_key_store.rs b/crates/diesel_models/src/query/merchant_key_store.rs
index fc4d09c4989..4e8e4637678 100644
--- a/crates/diesel_models/src/query/merchant_key_store.rs
+++ b/crates/diesel_models/src/query/merchant_key_store.rs
@@ -55,7 +55,11 @@ impl MerchantKeyStore {
         .await
     }
 
-    pub async fn list_all_key_stores(conn: &PgPooledConn) -> StorageResult<Vec<Self>> {
+    pub async fn list_all_key_stores(
+        conn: &PgPooledConn,
+        from: u32,
+        limit: u32,
+    ) -> StorageResult<Vec<Self>> {
         generics::generic_filter::<
             <Self as HasTable>::Table,
             _,
@@ -64,8 +68,8 @@ impl MerchantKeyStore {
         >(
             conn,
             dsl::merchant_id.ne_all(vec!["".to_string()]),
-            None,
-            None,
+            Some(limit.into()),
+            Some(from.into()),
             None,
         )
         .await
diff --git a/crates/diesel_models/src/query/user_key_store.rs b/crates/diesel_models/src/query/user_key_store.rs
index 5aad28bb562..78d1648b470 100644
--- a/crates/diesel_models/src/query/user_key_store.rs
+++ b/crates/diesel_models/src/query/user_key_store.rs
@@ -14,7 +14,11 @@ impl UserKeyStoreNew {
 }
 
 impl UserKeyStore {
-    pub async fn get_all_user_key_stores(conn: &PgPooledConn) -> StorageResult<Vec<Self>> {
+    pub async fn get_all_user_key_stores(
+        conn: &PgPooledConn,
+        from: u32,
+        limit: u32,
+    ) -> StorageResult<Vec<Self>> {
         generics::generic_filter::<
             <Self as HasTable>::Table,
             _,
@@ -23,8 +27,8 @@ impl UserKeyStore {
         >(
             conn,
             dsl::user_id.ne_all(vec!["".to_string()]),
-            None,
-            None,
+            Some(limit.into()),
+            Some(from.into()),
             None,
         )
         .await
diff --git a/crates/hyperswitch_domain_models/src/api.rs b/crates/hyperswitch_domain_models/src/api.rs
index 07d9337e450..39f3f04c548 100644
--- a/crates/hyperswitch_domain_models/src/api.rs
+++ b/crates/hyperswitch_domain_models/src/api.rs
@@ -2,7 +2,7 @@ use std::{collections::HashSet, fmt::Display};
 
 use common_utils::{
     events::{ApiEventMetric, ApiEventsType},
-    impl_misc_api_event_type,
+    impl_api_event_type,
 };
 
 #[derive(Debug, Eq, PartialEq)]
@@ -28,7 +28,7 @@ impl<T: ApiEventMetric> ApiEventMetric for ApplicationResponse<T> {
     }
 }
 
-impl_misc_api_event_type!(PaymentLinkFormData, GenericLinkFormData);
+impl_api_event_type!(Miscellaneous, (PaymentLinkFormData, GenericLinkFormData));
 
 #[derive(Debug, Eq, PartialEq)]
 pub struct RedirectionFormData {
diff --git a/crates/hyperswitch_domain_models/src/type_encryption.rs b/crates/hyperswitch_domain_models/src/type_encryption.rs
index ed36ad59aec..ba739079169 100644
--- a/crates/hyperswitch_domain_models/src/type_encryption.rs
+++ b/crates/hyperswitch_domain_models/src/type_encryption.rs
@@ -7,792 +7,806 @@ use common_utils::{
     metrics::utils::record_operation_time,
     types::keymanager::{Identifier, KeyManagerState},
 };
-use error_stack::ResultExt;
-use masking::{PeekInterface, Secret};
-use router_env::{instrument, tracing};
+use encrypt::TypeEncryption;
+use masking::Secret;
 use rustc_hash::FxHashMap;
-#[cfg(feature = "encryption_service")]
-use {
-    common_utils::{
-        keymanager::call_encryption_service,
-        transformers::{ForeignFrom, ForeignTryFrom},
-        types::keymanager::{
-            BatchDecryptDataResponse, BatchEncryptDataRequest, BatchEncryptDataResponse,
-            DecryptDataResponse, EncryptDataRequest, EncryptDataResponse,
-            TransientBatchDecryptDataRequest, TransientDecryptDataRequest,
-        },
-    },
-    http::Method,
-    router_env::logger,
-};
 
-#[async_trait]
-pub trait TypeEncryption<
-    T,
-    V: crypto::EncodeMessage + crypto::DecodeMessage,
-    S: masking::Strategy<T>,
->: Sized
-{
-    async fn encrypt_via_api(
-        state: &KeyManagerState,
-        masked_data: Secret<T, S>,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError>;
-
-    async fn decrypt_via_api(
-        state: &KeyManagerState,
-        encrypted_data: Encryption,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError>;
-
-    async fn encrypt(
-        masked_data: Secret<T, S>,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError>;
-
-    async fn decrypt(
-        encrypted_data: Encryption,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError>;
-
-    async fn batch_encrypt_via_api(
-        state: &KeyManagerState,
-        masked_data: FxHashMap<String, Secret<T, S>>,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError>;
-
-    async fn batch_decrypt_via_api(
-        state: &KeyManagerState,
-        encrypted_data: FxHashMap<String, Encryption>,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError>;
-
-    async fn batch_encrypt(
-        masked_data: FxHashMap<String, Secret<T, S>>,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError>;
-
-    async fn batch_decrypt(
-        encrypted_data: FxHashMap<String, Encryption>,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError>;
-}
+mod encrypt {
+    use async_trait::async_trait;
+    use common_utils::{
+        crypto,
+        encryption::Encryption,
+        errors::{self, CustomResult},
+        types::keymanager::{Identifier, KeyManagerState},
+    };
+    use error_stack::ResultExt;
+    use masking::{PeekInterface, Secret};
+    use router_env::{instrument, tracing};
+    use rustc_hash::FxHashMap;
+    #[cfg(feature = "encryption_service")]
+    use {
+        common_utils::{
+            keymanager::call_encryption_service,
+            transformers::{ForeignFrom, ForeignTryFrom},
+            types::keymanager::{
+                BatchDecryptDataResponse, BatchEncryptDataRequest, BatchEncryptDataResponse,
+                DecryptDataResponse, EncryptDataRequest, EncryptDataResponse,
+                TransientBatchDecryptDataRequest, TransientDecryptDataRequest,
+            },
+        },
+        http::Method,
+        router_env::logger,
+    };
+
+    use super::metrics;
+
+    #[async_trait]
+    pub trait TypeEncryption<
+        T,
+        V: crypto::EncodeMessage + crypto::DecodeMessage,
+        S: masking::Strategy<T>,
+    >: Sized
+    {
+        async fn encrypt_via_api(
+            state: &KeyManagerState,
+            masked_data: Secret<T, S>,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError>;
+
+        async fn decrypt_via_api(
+            state: &KeyManagerState,
+            encrypted_data: Encryption,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError>;
+
+        async fn encrypt(
+            masked_data: Secret<T, S>,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError>;
+
+        async fn decrypt(
+            encrypted_data: Encryption,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError>;
+
+        async fn batch_encrypt_via_api(
+            state: &KeyManagerState,
+            masked_data: FxHashMap<String, Secret<T, S>>,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError>;
+
+        async fn batch_decrypt_via_api(
+            state: &KeyManagerState,
+            encrypted_data: FxHashMap<String, Encryption>,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError>;
+
+        async fn batch_encrypt(
+            masked_data: FxHashMap<String, Secret<T, S>>,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError>;
+
+        async fn batch_decrypt(
+            encrypted_data: FxHashMap<String, Encryption>,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError>;
+    }
 
-#[async_trait]
-impl<
-        V: crypto::DecodeMessage + crypto::EncodeMessage + Send + 'static,
-        S: masking::Strategy<String> + Send + Sync,
-    > TypeEncryption<String, V, S> for crypto::Encryptable<Secret<String, S>>
-{
-    #[instrument(skip_all)]
-    #[allow(unused_variables)]
-    async fn encrypt_via_api(
-        state: &KeyManagerState,
-        masked_data: Secret<String, S>,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError> {
-        #[cfg(not(feature = "encryption_service"))]
-        {
-            Self::encrypt(masked_data, key, crypt_algo).await
-        }
-        #[cfg(feature = "encryption_service")]
-        {
-            let result: Result<
-                EncryptDataResponse,
-                error_stack::Report<errors::KeyManagerClientError>,
-            > = call_encryption_service(
-                state,
-                Method::POST,
-                "data/encrypt",
-                EncryptDataRequest::from((masked_data.clone(), identifier)),
-            )
-            .await;
-            match result {
-                Ok(response) => Ok(ForeignFrom::foreign_from((masked_data.clone(), response))),
-                Err(err) => {
-                    logger::error!("Encryption error {:?}", err);
-                    metrics::ENCRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
-                    logger::info!("Fall back to Application Encryption");
-                    Self::encrypt(masked_data, key, crypt_algo).await
+    #[async_trait]
+    impl<
+            V: crypto::DecodeMessage + crypto::EncodeMessage + Send + 'static,
+            S: masking::Strategy<String> + Send + Sync,
+        > TypeEncryption<String, V, S> for crypto::Encryptable<Secret<String, S>>
+    {
+        #[instrument(skip_all)]
+        #[allow(unused_variables)]
+        async fn encrypt_via_api(
+            state: &KeyManagerState,
+            masked_data: Secret<String, S>,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError> {
+            #[cfg(not(feature = "encryption_service"))]
+            {
+                Self::encrypt(masked_data, key, crypt_algo).await
+            }
+            #[cfg(feature = "encryption_service")]
+            {
+                let result: Result<
+                    EncryptDataResponse,
+                    error_stack::Report<errors::KeyManagerClientError>,
+                > = call_encryption_service(
+                    state,
+                    Method::POST,
+                    "data/encrypt",
+                    EncryptDataRequest::from((masked_data.clone(), identifier)),
+                )
+                .await;
+                match result {
+                    Ok(response) => Ok(ForeignFrom::foreign_from((masked_data.clone(), response))),
+                    Err(err) => {
+                        logger::error!("Encryption error {:?}", err);
+                        metrics::ENCRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
+                        logger::info!("Fall back to Application Encryption");
+                        Self::encrypt(masked_data, key, crypt_algo).await
+                    }
                 }
             }
         }
-    }
 
-    #[instrument(skip_all)]
-    #[allow(unused_variables)]
-    async fn decrypt_via_api(
-        state: &KeyManagerState,
-        encrypted_data: Encryption,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError> {
-        #[cfg(not(feature = "encryption_service"))]
-        {
-            Self::decrypt(encrypted_data, key, crypt_algo).await
-        }
-        #[cfg(feature = "encryption_service")]
-        {
-            let result: Result<
-                DecryptDataResponse,
-                error_stack::Report<errors::KeyManagerClientError>,
-            > = call_encryption_service(
-                state,
-                Method::POST,
-                "data/decrypt",
-                TransientDecryptDataRequest::from((encrypted_data.clone(), identifier)),
-            )
-            .await;
-            let decrypted = match result {
-                Ok(decrypted_data) => {
-                    ForeignTryFrom::foreign_try_from((encrypted_data.clone(), decrypted_data))
-                }
-                Err(err) => {
-                    logger::error!("Decryption error {:?}", err);
-                    Err(err.change_context(errors::CryptoError::DecodingFailed))
-                }
-            };
-
-            match decrypted {
-                Ok(de) => Ok(de),
-                Err(_) => {
-                    metrics::DECRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
-                    logger::info!("Fall back to Application Decryption");
-                    Self::decrypt(encrypted_data, key, crypt_algo).await
+        #[instrument(skip_all)]
+        #[allow(unused_variables)]
+        async fn decrypt_via_api(
+            state: &KeyManagerState,
+            encrypted_data: Encryption,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError> {
+            #[cfg(not(feature = "encryption_service"))]
+            {
+                Self::decrypt(encrypted_data, key, crypt_algo).await
+            }
+            #[cfg(feature = "encryption_service")]
+            {
+                let result: Result<
+                    DecryptDataResponse,
+                    error_stack::Report<errors::KeyManagerClientError>,
+                > = call_encryption_service(
+                    state,
+                    Method::POST,
+                    "data/decrypt",
+                    TransientDecryptDataRequest::from((encrypted_data.clone(), identifier)),
+                )
+                .await;
+                let decrypted = match result {
+                    Ok(decrypted_data) => {
+                        ForeignTryFrom::foreign_try_from((encrypted_data.clone(), decrypted_data))
+                    }
+                    Err(err) => {
+                        logger::error!("Decryption error {:?}", err);
+                        Err(err.change_context(errors::CryptoError::DecodingFailed))
+                    }
+                };
+
+                match decrypted {
+                    Ok(de) => Ok(de),
+                    Err(_) => {
+                        metrics::DECRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
+                        logger::info!("Fall back to Application Decryption");
+                        Self::decrypt(encrypted_data, key, crypt_algo).await
+                    }
                 }
             }
         }
-    }
-
-    async fn encrypt(
-        masked_data: Secret<String, S>,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError> {
-        metrics::APPLICATION_ENCRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
-        let encrypted_data = crypt_algo.encode_message(key, masked_data.peek().as_bytes())?;
-        Ok(Self::new(masked_data, encrypted_data.into()))
-    }
 
-    async fn decrypt(
-        encrypted_data: Encryption,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError> {
-        metrics::APPLICATION_DECRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
-        let encrypted = encrypted_data.into_inner();
-        let data = crypt_algo.decode_message(key, encrypted.clone())?;
+        async fn encrypt(
+            masked_data: Secret<String, S>,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError> {
+            metrics::APPLICATION_ENCRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
+            let encrypted_data = crypt_algo.encode_message(key, masked_data.peek().as_bytes())?;
+            Ok(Self::new(masked_data, encrypted_data.into()))
+        }
 
-        let value: String = std::str::from_utf8(&data)
-            .change_context(errors::CryptoError::DecodingFailed)?
-            .to_string();
+        async fn decrypt(
+            encrypted_data: Encryption,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError> {
+            metrics::APPLICATION_DECRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
+            let encrypted = encrypted_data.into_inner();
+            let data = crypt_algo.decode_message(key, encrypted.clone())?;
 
-        Ok(Self::new(value.into(), encrypted))
-    }
+            let value: String = std::str::from_utf8(&data)
+                .change_context(errors::CryptoError::DecodingFailed)?
+                .to_string();
 
-    #[allow(unused_variables)]
-    async fn batch_encrypt_via_api(
-        state: &KeyManagerState,
-        masked_data: FxHashMap<String, Secret<String, S>>,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
-        #[cfg(not(feature = "encryption_service"))]
-        {
-            Self::batch_encrypt(masked_data, key, crypt_algo).await
+            Ok(Self::new(value.into(), encrypted))
         }
 
-        #[cfg(feature = "encryption_service")]
-        {
-            let result: Result<
-                BatchEncryptDataResponse,
-                error_stack::Report<errors::KeyManagerClientError>,
-            > = call_encryption_service(
-                state,
-                Method::POST,
-                "data/encrypt",
-                BatchEncryptDataRequest::from((masked_data.clone(), identifier)),
-            )
-            .await;
-            match result {
-                Ok(response) => Ok(ForeignFrom::foreign_from((masked_data, response))),
-                Err(err) => {
-                    metrics::ENCRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
-                    logger::error!("Encryption error {:?}", err);
-                    logger::info!("Fall back to Application Encryption");
-                    Self::batch_encrypt(masked_data, key, crypt_algo).await
+        #[allow(unused_variables)]
+        async fn batch_encrypt_via_api(
+            state: &KeyManagerState,
+            masked_data: FxHashMap<String, Secret<String, S>>,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
+            #[cfg(not(feature = "encryption_service"))]
+            {
+                Self::batch_encrypt(masked_data, key, crypt_algo).await
+            }
+
+            #[cfg(feature = "encryption_service")]
+            {
+                let result: Result<
+                    BatchEncryptDataResponse,
+                    error_stack::Report<errors::KeyManagerClientError>,
+                > = call_encryption_service(
+                    state,
+                    Method::POST,
+                    "data/encrypt",
+                    BatchEncryptDataRequest::from((masked_data.clone(), identifier)),
+                )
+                .await;
+                match result {
+                    Ok(response) => Ok(ForeignFrom::foreign_from((masked_data, response))),
+                    Err(err) => {
+                        metrics::ENCRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
+                        logger::error!("Encryption error {:?}", err);
+                        logger::info!("Fall back to Application Encryption");
+                        Self::batch_encrypt(masked_data, key, crypt_algo).await
+                    }
                 }
             }
         }
-    }
 
-    #[allow(unused_variables)]
-    async fn batch_decrypt_via_api(
-        state: &KeyManagerState,
-        encrypted_data: FxHashMap<String, Encryption>,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
-        #[cfg(not(feature = "encryption_service"))]
-        {
-            Self::batch_decrypt(encrypted_data, key, crypt_algo).await
-        }
+        #[allow(unused_variables)]
+        async fn batch_decrypt_via_api(
+            state: &KeyManagerState,
+            encrypted_data: FxHashMap<String, Encryption>,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
+            #[cfg(not(feature = "encryption_service"))]
+            {
+                Self::batch_decrypt(encrypted_data, key, crypt_algo).await
+            }
 
-        #[cfg(feature = "encryption_service")]
-        {
-            let result: Result<
-                BatchDecryptDataResponse,
-                error_stack::Report<errors::KeyManagerClientError>,
-            > = call_encryption_service(
-                state,
-                Method::POST,
-                "data/decrypt",
-                TransientBatchDecryptDataRequest::from((encrypted_data.clone(), identifier)),
-            )
-            .await;
-            let decrypted = match result {
-                Ok(decrypted_data) => {
-                    ForeignTryFrom::foreign_try_from((encrypted_data.clone(), decrypted_data))
-                }
-                Err(err) => {
-                    logger::error!("Decryption error {:?}", err);
-                    Err(err.change_context(errors::CryptoError::DecodingFailed))
-                }
-            };
-            match decrypted {
-                Ok(de) => Ok(de),
-                Err(_) => {
-                    metrics::DECRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
-                    logger::info!("Fall back to Application Decryption");
-                    Self::batch_decrypt(encrypted_data, key, crypt_algo).await
+            #[cfg(feature = "encryption_service")]
+            {
+                let result: Result<
+                    BatchDecryptDataResponse,
+                    error_stack::Report<errors::KeyManagerClientError>,
+                > = call_encryption_service(
+                    state,
+                    Method::POST,
+                    "data/decrypt",
+                    TransientBatchDecryptDataRequest::from((encrypted_data.clone(), identifier)),
+                )
+                .await;
+                let decrypted = match result {
+                    Ok(decrypted_data) => {
+                        ForeignTryFrom::foreign_try_from((encrypted_data.clone(), decrypted_data))
+                    }
+                    Err(err) => {
+                        logger::error!("Decryption error {:?}", err);
+                        Err(err.change_context(errors::CryptoError::DecodingFailed))
+                    }
+                };
+                match decrypted {
+                    Ok(de) => Ok(de),
+                    Err(_) => {
+                        metrics::DECRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
+                        logger::info!("Fall back to Application Decryption");
+                        Self::batch_decrypt(encrypted_data, key, crypt_algo).await
+                    }
                 }
             }
         }
-    }
 
-    async fn batch_encrypt(
-        masked_data: FxHashMap<String, Secret<String, S>>,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
-        metrics::APPLICATION_ENCRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
-        masked_data
-            .into_iter()
-            .map(|(k, v)| {
-                Ok((
-                    k,
-                    Self::new(
-                        v.clone(),
-                        crypt_algo.encode_message(key, v.peek().as_bytes())?.into(),
-                    ),
-                ))
-            })
-            .collect()
-    }
+        async fn batch_encrypt(
+            masked_data: FxHashMap<String, Secret<String, S>>,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
+            metrics::APPLICATION_ENCRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
+            masked_data
+                .into_iter()
+                .map(|(k, v)| {
+                    Ok((
+                        k,
+                        Self::new(
+                            v.clone(),
+                            crypt_algo.encode_message(key, v.peek().as_bytes())?.into(),
+                        ),
+                    ))
+                })
+                .collect()
+        }
 
-    async fn batch_decrypt(
-        encrypted_data: FxHashMap<String, Encryption>,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
-        metrics::APPLICATION_DECRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
-        encrypted_data
-            .into_iter()
-            .map(|(k, v)| {
-                let data = crypt_algo.decode_message(key, v.clone().into_inner())?;
-                let value: String = std::str::from_utf8(&data)
-                    .change_context(errors::CryptoError::DecodingFailed)?
-                    .to_string();
-                Ok((k, Self::new(value.into(), v.into_inner())))
-            })
-            .collect()
+        async fn batch_decrypt(
+            encrypted_data: FxHashMap<String, Encryption>,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
+            metrics::APPLICATION_DECRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
+            encrypted_data
+                .into_iter()
+                .map(|(k, v)| {
+                    let data = crypt_algo.decode_message(key, v.clone().into_inner())?;
+                    let value: String = std::str::from_utf8(&data)
+                        .change_context(errors::CryptoError::DecodingFailed)?
+                        .to_string();
+                    Ok((k, Self::new(value.into(), v.into_inner())))
+                })
+                .collect()
+        }
     }
-}
 
-#[async_trait]
-impl<
-        V: crypto::DecodeMessage + crypto::EncodeMessage + Send + 'static,
-        S: masking::Strategy<serde_json::Value> + Send + Sync,
-    > TypeEncryption<serde_json::Value, V, S>
-    for crypto::Encryptable<Secret<serde_json::Value, S>>
-{
-    #[instrument(skip_all)]
-    #[allow(unused_variables)]
-    async fn encrypt_via_api(
-        state: &KeyManagerState,
-        masked_data: Secret<serde_json::Value, S>,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError> {
-        #[cfg(not(feature = "encryption_service"))]
-        {
-            Self::encrypt(masked_data, key, crypt_algo).await
-        }
-        #[cfg(feature = "encryption_service")]
-        {
-            let result: Result<
-                EncryptDataResponse,
-                error_stack::Report<errors::KeyManagerClientError>,
-            > = call_encryption_service(
-                state,
-                Method::POST,
-                "data/encrypt",
-                EncryptDataRequest::from((masked_data.clone(), identifier)),
-            )
-            .await;
-            match result {
-                Ok(response) => Ok(ForeignFrom::foreign_from((masked_data.clone(), response))),
-                Err(err) => {
-                    logger::error!("Encryption error {:?}", err);
-                    metrics::ENCRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
-                    logger::info!("Fall back to Application Encryption");
-                    Self::encrypt(masked_data, key, crypt_algo).await
+    #[async_trait]
+    impl<
+            V: crypto::DecodeMessage + crypto::EncodeMessage + Send + 'static,
+            S: masking::Strategy<serde_json::Value> + Send + Sync,
+        > TypeEncryption<serde_json::Value, V, S>
+        for crypto::Encryptable<Secret<serde_json::Value, S>>
+    {
+        #[instrument(skip_all)]
+        #[allow(unused_variables)]
+        async fn encrypt_via_api(
+            state: &KeyManagerState,
+            masked_data: Secret<serde_json::Value, S>,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError> {
+            #[cfg(not(feature = "encryption_service"))]
+            {
+                Self::encrypt(masked_data, key, crypt_algo).await
+            }
+            #[cfg(feature = "encryption_service")]
+            {
+                let result: Result<
+                    EncryptDataResponse,
+                    error_stack::Report<errors::KeyManagerClientError>,
+                > = call_encryption_service(
+                    state,
+                    Method::POST,
+                    "data/encrypt",
+                    EncryptDataRequest::from((masked_data.clone(), identifier)),
+                )
+                .await;
+                match result {
+                    Ok(response) => Ok(ForeignFrom::foreign_from((masked_data.clone(), response))),
+                    Err(err) => {
+                        logger::error!("Encryption error {:?}", err);
+                        metrics::ENCRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
+                        logger::info!("Fall back to Application Encryption");
+                        Self::encrypt(masked_data, key, crypt_algo).await
+                    }
                 }
             }
         }
-    }
 
-    #[instrument(skip_all)]
-    #[allow(unused_variables)]
-    async fn decrypt_via_api(
-        state: &KeyManagerState,
-        encrypted_data: Encryption,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError> {
-        #[cfg(not(feature = "encryption_service"))]
-        {
-            Self::decrypt(encrypted_data, key, crypt_algo).await
-        }
-        #[cfg(feature = "encryption_service")]
-        {
-            let result: Result<
-                DecryptDataResponse,
-                error_stack::Report<errors::KeyManagerClientError>,
-            > = call_encryption_service(
-                state,
-                Method::POST,
-                "data/decrypt",
-                TransientDecryptDataRequest::from((encrypted_data.clone(), identifier)),
-            )
-            .await;
-            let decrypted = match result {
-                Ok(decrypted_data) => {
-                    ForeignTryFrom::foreign_try_from((encrypted_data.clone(), decrypted_data))
-                }
-                Err(err) => {
-                    logger::error!("Decryption error {:?}", err);
-                    Err(err.change_context(errors::CryptoError::EncodingFailed))
-                }
-            };
-            match decrypted {
-                Ok(de) => Ok(de),
-                Err(_) => {
-                    metrics::DECRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
-                    logger::info!("Fall back to Application Decryption");
-                    Self::decrypt(encrypted_data, key, crypt_algo).await
+        #[instrument(skip_all)]
+        #[allow(unused_variables)]
+        async fn decrypt_via_api(
+            state: &KeyManagerState,
+            encrypted_data: Encryption,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError> {
+            #[cfg(not(feature = "encryption_service"))]
+            {
+                Self::decrypt(encrypted_data, key, crypt_algo).await
+            }
+            #[cfg(feature = "encryption_service")]
+            {
+                let result: Result<
+                    DecryptDataResponse,
+                    error_stack::Report<errors::KeyManagerClientError>,
+                > = call_encryption_service(
+                    state,
+                    Method::POST,
+                    "data/decrypt",
+                    TransientDecryptDataRequest::from((encrypted_data.clone(), identifier)),
+                )
+                .await;
+                let decrypted = match result {
+                    Ok(decrypted_data) => {
+                        ForeignTryFrom::foreign_try_from((encrypted_data.clone(), decrypted_data))
+                    }
+                    Err(err) => {
+                        logger::error!("Decryption error {:?}", err);
+                        Err(err.change_context(errors::CryptoError::EncodingFailed))
+                    }
+                };
+                match decrypted {
+                    Ok(de) => Ok(de),
+                    Err(_) => {
+                        metrics::DECRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
+                        logger::info!("Fall back to Application Decryption");
+                        Self::decrypt(encrypted_data, key, crypt_algo).await
+                    }
                 }
             }
         }
-    }
-
-    #[instrument(skip_all)]
-    async fn encrypt(
-        masked_data: Secret<serde_json::Value, S>,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError> {
-        metrics::APPLICATION_ENCRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
-        let data = serde_json::to_vec(&masked_data.peek())
-            .change_context(errors::CryptoError::DecodingFailed)?;
-        let encrypted_data = crypt_algo.encode_message(key, &data)?;
-        Ok(Self::new(masked_data, encrypted_data.into()))
-    }
 
-    #[instrument(skip_all)]
-    async fn decrypt(
-        encrypted_data: Encryption,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError> {
-        metrics::APPLICATION_DECRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
-        let encrypted = encrypted_data.into_inner();
-        let data = crypt_algo.decode_message(key, encrypted.clone())?;
-
-        let value: serde_json::Value =
-            serde_json::from_slice(&data).change_context(errors::CryptoError::DecodingFailed)?;
-        Ok(Self::new(value.into(), encrypted))
-    }
+        #[instrument(skip_all)]
+        async fn encrypt(
+            masked_data: Secret<serde_json::Value, S>,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError> {
+            metrics::APPLICATION_ENCRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
+            let data = serde_json::to_vec(&masked_data.peek())
+                .change_context(errors::CryptoError::DecodingFailed)?;
+            let encrypted_data = crypt_algo.encode_message(key, &data)?;
+            Ok(Self::new(masked_data, encrypted_data.into()))
+        }
 
-    #[allow(unused_variables)]
-    async fn batch_encrypt_via_api(
-        state: &KeyManagerState,
-        masked_data: FxHashMap<String, Secret<serde_json::Value, S>>,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
-        #[cfg(not(feature = "encryption_service"))]
-        {
-            Self::batch_encrypt(masked_data, key, crypt_algo).await
+        #[instrument(skip_all)]
+        async fn decrypt(
+            encrypted_data: Encryption,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError> {
+            metrics::APPLICATION_DECRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
+            let encrypted = encrypted_data.into_inner();
+            let data = crypt_algo.decode_message(key, encrypted.clone())?;
+
+            let value: serde_json::Value = serde_json::from_slice(&data)
+                .change_context(errors::CryptoError::DecodingFailed)?;
+            Ok(Self::new(value.into(), encrypted))
         }
-        #[cfg(feature = "encryption_service")]
-        {
-            let result: Result<
-                BatchEncryptDataResponse,
-                error_stack::Report<errors::KeyManagerClientError>,
-            > = call_encryption_service(
-                state,
-                Method::POST,
-                "data/encrypt",
-                BatchEncryptDataRequest::from((masked_data.clone(), identifier)),
-            )
-            .await;
-            match result {
-                Ok(response) => Ok(ForeignFrom::foreign_from((masked_data, response))),
-                Err(err) => {
-                    metrics::ENCRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
-                    logger::error!("Encryption error {:?}", err);
-                    logger::info!("Fall back to Application Encryption");
-                    Self::batch_encrypt(masked_data, key, crypt_algo).await
+
+        #[allow(unused_variables)]
+        async fn batch_encrypt_via_api(
+            state: &KeyManagerState,
+            masked_data: FxHashMap<String, Secret<serde_json::Value, S>>,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
+            #[cfg(not(feature = "encryption_service"))]
+            {
+                Self::batch_encrypt(masked_data, key, crypt_algo).await
+            }
+            #[cfg(feature = "encryption_service")]
+            {
+                let result: Result<
+                    BatchEncryptDataResponse,
+                    error_stack::Report<errors::KeyManagerClientError>,
+                > = call_encryption_service(
+                    state,
+                    Method::POST,
+                    "data/encrypt",
+                    BatchEncryptDataRequest::from((masked_data.clone(), identifier)),
+                )
+                .await;
+                match result {
+                    Ok(response) => Ok(ForeignFrom::foreign_from((masked_data, response))),
+                    Err(err) => {
+                        metrics::ENCRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
+                        logger::error!("Encryption error {:?}", err);
+                        logger::info!("Fall back to Application Encryption");
+                        Self::batch_encrypt(masked_data, key, crypt_algo).await
+                    }
                 }
             }
         }
-    }
 
-    #[allow(unused_variables)]
-    async fn batch_decrypt_via_api(
-        state: &KeyManagerState,
-        encrypted_data: FxHashMap<String, Encryption>,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
-        #[cfg(not(feature = "encryption_service"))]
-        {
-            Self::batch_decrypt(encrypted_data, key, crypt_algo).await
-        }
-        #[cfg(feature = "encryption_service")]
-        {
-            let result: Result<
-                BatchDecryptDataResponse,
-                error_stack::Report<errors::KeyManagerClientError>,
-            > = call_encryption_service(
-                state,
-                Method::POST,
-                "data/decrypt",
-                TransientBatchDecryptDataRequest::from((encrypted_data.clone(), identifier)),
-            )
-            .await;
-            let decrypted = match result {
-                Ok(decrypted_data) => {
-                    ForeignTryFrom::foreign_try_from((encrypted_data.clone(), decrypted_data))
-                }
-                Err(err) => {
-                    logger::error!("Decryption error {:?}", err);
-                    Err(err.change_context(errors::CryptoError::DecodingFailed))
-                }
-            };
-            match decrypted {
-                Ok(de) => Ok(de),
-                Err(_) => {
-                    metrics::DECRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
-                    logger::info!("Fall back to Application Decryption");
-                    Self::batch_decrypt(encrypted_data, key, crypt_algo).await
+        #[allow(unused_variables)]
+        async fn batch_decrypt_via_api(
+            state: &KeyManagerState,
+            encrypted_data: FxHashMap<String, Encryption>,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
+            #[cfg(not(feature = "encryption_service"))]
+            {
+                Self::batch_decrypt(encrypted_data, key, crypt_algo).await
+            }
+            #[cfg(feature = "encryption_service")]
+            {
+                let result: Result<
+                    BatchDecryptDataResponse,
+                    error_stack::Report<errors::KeyManagerClientError>,
+                > = call_encryption_service(
+                    state,
+                    Method::POST,
+                    "data/decrypt",
+                    TransientBatchDecryptDataRequest::from((encrypted_data.clone(), identifier)),
+                )
+                .await;
+                let decrypted = match result {
+                    Ok(decrypted_data) => {
+                        ForeignTryFrom::foreign_try_from((encrypted_data.clone(), decrypted_data))
+                    }
+                    Err(err) => {
+                        logger::error!("Decryption error {:?}", err);
+                        Err(err.change_context(errors::CryptoError::DecodingFailed))
+                    }
+                };
+                match decrypted {
+                    Ok(de) => Ok(de),
+                    Err(_) => {
+                        metrics::DECRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
+                        logger::info!("Fall back to Application Decryption");
+                        Self::batch_decrypt(encrypted_data, key, crypt_algo).await
+                    }
                 }
             }
         }
-    }
 
-    async fn batch_encrypt(
-        masked_data: FxHashMap<String, Secret<serde_json::Value, S>>,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
-        metrics::APPLICATION_ENCRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
-        masked_data
-            .into_iter()
-            .map(|(k, v)| {
-                let data = serde_json::to_vec(v.peek())
-                    .change_context(errors::CryptoError::DecodingFailed)?;
-                Ok((
-                    k,
-                    Self::new(v, crypt_algo.encode_message(key, &data)?.into()),
-                ))
-            })
-            .collect()
-    }
+        async fn batch_encrypt(
+            masked_data: FxHashMap<String, Secret<serde_json::Value, S>>,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
+            metrics::APPLICATION_ENCRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
+            masked_data
+                .into_iter()
+                .map(|(k, v)| {
+                    let data = serde_json::to_vec(v.peek())
+                        .change_context(errors::CryptoError::DecodingFailed)?;
+                    Ok((
+                        k,
+                        Self::new(v, crypt_algo.encode_message(key, &data)?.into()),
+                    ))
+                })
+                .collect()
+        }
 
-    async fn batch_decrypt(
-        encrypted_data: FxHashMap<String, Encryption>,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
-        metrics::APPLICATION_DECRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
-        encrypted_data
-            .into_iter()
-            .map(|(k, v)| {
-                let data = crypt_algo.decode_message(key, v.clone().into_inner().clone())?;
-
-                let value: serde_json::Value = serde_json::from_slice(&data)
-                    .change_context(errors::CryptoError::DecodingFailed)?;
-                Ok((k, Self::new(value.into(), v.into_inner())))
-            })
-            .collect()
+        async fn batch_decrypt(
+            encrypted_data: FxHashMap<String, Encryption>,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
+            metrics::APPLICATION_DECRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
+            encrypted_data
+                .into_iter()
+                .map(|(k, v)| {
+                    let data = crypt_algo.decode_message(key, v.clone().into_inner().clone())?;
+
+                    let value: serde_json::Value = serde_json::from_slice(&data)
+                        .change_context(errors::CryptoError::DecodingFailed)?;
+                    Ok((k, Self::new(value.into(), v.into_inner())))
+                })
+                .collect()
+        }
     }
-}
 
-#[async_trait]
-impl<
-        V: crypto::DecodeMessage + crypto::EncodeMessage + Send + 'static,
-        S: masking::Strategy<Vec<u8>> + Send + Sync,
-    > TypeEncryption<Vec<u8>, V, S> for crypto::Encryptable<Secret<Vec<u8>, S>>
-{
-    #[instrument(skip_all)]
-    #[allow(unused_variables)]
-    async fn encrypt_via_api(
-        state: &KeyManagerState,
-        masked_data: Secret<Vec<u8>, S>,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError> {
-        #[cfg(not(feature = "encryption_service"))]
-        {
-            Self::encrypt(masked_data, key, crypt_algo).await
-        }
-        #[cfg(feature = "encryption_service")]
-        {
-            let result: Result<
-                EncryptDataResponse,
-                error_stack::Report<errors::KeyManagerClientError>,
-            > = call_encryption_service(
-                state,
-                Method::POST,
-                "data/encrypt",
-                EncryptDataRequest::from((masked_data.clone(), identifier)),
-            )
-            .await;
-            match result {
-                Ok(response) => Ok(ForeignFrom::foreign_from((masked_data.clone(), response))),
-                Err(err) => {
-                    logger::error!("Encryption error {:?}", err);
-                    metrics::ENCRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
-                    logger::info!("Fall back to Application Encryption");
-                    Self::encrypt(masked_data, key, crypt_algo).await
+    #[async_trait]
+    impl<
+            V: crypto::DecodeMessage + crypto::EncodeMessage + Send + 'static,
+            S: masking::Strategy<Vec<u8>> + Send + Sync,
+        > TypeEncryption<Vec<u8>, V, S> for crypto::Encryptable<Secret<Vec<u8>, S>>
+    {
+        #[instrument(skip_all)]
+        #[allow(unused_variables)]
+        async fn encrypt_via_api(
+            state: &KeyManagerState,
+            masked_data: Secret<Vec<u8>, S>,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError> {
+            #[cfg(not(feature = "encryption_service"))]
+            {
+                Self::encrypt(masked_data, key, crypt_algo).await
+            }
+            #[cfg(feature = "encryption_service")]
+            {
+                let result: Result<
+                    EncryptDataResponse,
+                    error_stack::Report<errors::KeyManagerClientError>,
+                > = call_encryption_service(
+                    state,
+                    Method::POST,
+                    "data/encrypt",
+                    EncryptDataRequest::from((masked_data.clone(), identifier)),
+                )
+                .await;
+                match result {
+                    Ok(response) => Ok(ForeignFrom::foreign_from((masked_data.clone(), response))),
+                    Err(err) => {
+                        logger::error!("Encryption error {:?}", err);
+                        metrics::ENCRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
+                        logger::info!("Fall back to Application Encryption");
+                        Self::encrypt(masked_data, key, crypt_algo).await
+                    }
                 }
             }
         }
-    }
 
-    #[instrument(skip_all)]
-    #[allow(unused_variables)]
-    async fn decrypt_via_api(
-        state: &KeyManagerState,
-        encrypted_data: Encryption,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError> {
-        #[cfg(not(feature = "encryption_service"))]
-        {
-            Self::decrypt(encrypted_data, key, crypt_algo).await
-        }
-        #[cfg(feature = "encryption_service")]
-        {
-            let result: Result<
-                DecryptDataResponse,
-                error_stack::Report<errors::KeyManagerClientError>,
-            > = call_encryption_service(
-                state,
-                Method::POST,
-                "data/decrypt",
-                TransientDecryptDataRequest::from((encrypted_data.clone(), identifier)),
-            )
-            .await;
-            let decrypted = match result {
-                Ok(decrypted_data) => {
-                    ForeignTryFrom::foreign_try_from((encrypted_data.clone(), decrypted_data))
-                }
-                Err(err) => {
-                    logger::error!("Decryption error {:?}", err);
-                    Err(err.change_context(errors::CryptoError::DecodingFailed))
-                }
-            };
-            match decrypted {
-                Ok(de) => Ok(de),
-                Err(_) => {
-                    metrics::DECRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
-                    logger::info!("Fall back to Application Decryption");
-                    Self::decrypt(encrypted_data, key, crypt_algo).await
+        #[instrument(skip_all)]
+        #[allow(unused_variables)]
+        async fn decrypt_via_api(
+            state: &KeyManagerState,
+            encrypted_data: Encryption,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError> {
+            #[cfg(not(feature = "encryption_service"))]
+            {
+                Self::decrypt(encrypted_data, key, crypt_algo).await
+            }
+            #[cfg(feature = "encryption_service")]
+            {
+                let result: Result<
+                    DecryptDataResponse,
+                    error_stack::Report<errors::KeyManagerClientError>,
+                > = call_encryption_service(
+                    state,
+                    Method::POST,
+                    "data/decrypt",
+                    TransientDecryptDataRequest::from((encrypted_data.clone(), identifier)),
+                )
+                .await;
+                let decrypted = match result {
+                    Ok(decrypted_data) => {
+                        ForeignTryFrom::foreign_try_from((encrypted_data.clone(), decrypted_data))
+                    }
+                    Err(err) => {
+                        logger::error!("Decryption error {:?}", err);
+                        Err(err.change_context(errors::CryptoError::DecodingFailed))
+                    }
+                };
+                match decrypted {
+                    Ok(de) => Ok(de),
+                    Err(_) => {
+                        metrics::DECRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
+                        logger::info!("Fall back to Application Decryption");
+                        Self::decrypt(encrypted_data, key, crypt_algo).await
+                    }
                 }
             }
         }
-    }
 
-    #[instrument(skip_all)]
-    async fn encrypt(
-        masked_data: Secret<Vec<u8>, S>,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError> {
-        metrics::APPLICATION_ENCRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
-        let encrypted_data = crypt_algo.encode_message(key, masked_data.peek())?;
-        Ok(Self::new(masked_data, encrypted_data.into()))
-    }
-
-    #[instrument(skip_all)]
-    async fn decrypt(
-        encrypted_data: Encryption,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<Self, errors::CryptoError> {
-        metrics::APPLICATION_DECRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
-        let encrypted = encrypted_data.into_inner();
-        let data = crypt_algo.decode_message(key, encrypted.clone())?;
-        Ok(Self::new(data.into(), encrypted))
-    }
+        #[instrument(skip_all)]
+        async fn encrypt(
+            masked_data: Secret<Vec<u8>, S>,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError> {
+            metrics::APPLICATION_ENCRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
+            let encrypted_data = crypt_algo.encode_message(key, masked_data.peek())?;
+            Ok(Self::new(masked_data, encrypted_data.into()))
+        }
 
-    #[allow(unused_variables)]
-    async fn batch_encrypt_via_api(
-        state: &KeyManagerState,
-        masked_data: FxHashMap<String, Secret<Vec<u8>, S>>,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
-        #[cfg(not(feature = "encryption_service"))]
-        {
-            Self::batch_encrypt(masked_data, key, crypt_algo).await
+        #[instrument(skip_all)]
+        async fn decrypt(
+            encrypted_data: Encryption,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<Self, errors::CryptoError> {
+            metrics::APPLICATION_DECRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
+            let encrypted = encrypted_data.into_inner();
+            let data = crypt_algo.decode_message(key, encrypted.clone())?;
+            Ok(Self::new(data.into(), encrypted))
         }
 
-        #[cfg(feature = "encryption_service")]
-        {
-            let result: Result<
-                BatchEncryptDataResponse,
-                error_stack::Report<errors::KeyManagerClientError>,
-            > = call_encryption_service(
-                state,
-                Method::POST,
-                "data/encrypt",
-                BatchEncryptDataRequest::from((masked_data.clone(), identifier)),
-            )
-            .await;
-            match result {
-                Ok(response) => Ok(ForeignFrom::foreign_from((masked_data, response))),
-                Err(err) => {
-                    metrics::ENCRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
-                    logger::error!("Encryption error {:?}", err);
-                    logger::info!("Fall back to Application Encryption");
-                    Self::batch_encrypt(masked_data, key, crypt_algo).await
+        #[allow(unused_variables)]
+        async fn batch_encrypt_via_api(
+            state: &KeyManagerState,
+            masked_data: FxHashMap<String, Secret<Vec<u8>, S>>,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
+            #[cfg(not(feature = "encryption_service"))]
+            {
+                Self::batch_encrypt(masked_data, key, crypt_algo).await
+            }
+
+            #[cfg(feature = "encryption_service")]
+            {
+                let result: Result<
+                    BatchEncryptDataResponse,
+                    error_stack::Report<errors::KeyManagerClientError>,
+                > = call_encryption_service(
+                    state,
+                    Method::POST,
+                    "data/encrypt",
+                    BatchEncryptDataRequest::from((masked_data.clone(), identifier)),
+                )
+                .await;
+                match result {
+                    Ok(response) => Ok(ForeignFrom::foreign_from((masked_data, response))),
+                    Err(err) => {
+                        metrics::ENCRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
+                        logger::error!("Encryption error {:?}", err);
+                        logger::info!("Fall back to Application Encryption");
+                        Self::batch_encrypt(masked_data, key, crypt_algo).await
+                    }
                 }
             }
         }
-    }
 
-    #[allow(unused_variables)]
-    async fn batch_decrypt_via_api(
-        state: &KeyManagerState,
-        encrypted_data: FxHashMap<String, Encryption>,
-        identifier: Identifier,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
-        #[cfg(not(feature = "encryption_service"))]
-        {
-            Self::batch_decrypt(encrypted_data, key, crypt_algo).await
-        }
-        #[cfg(feature = "encryption_service")]
-        {
-            let result: Result<
-                BatchDecryptDataResponse,
-                error_stack::Report<errors::KeyManagerClientError>,
-            > = call_encryption_service(
-                state,
-                Method::POST,
-                "data/decrypt",
-                TransientBatchDecryptDataRequest::from((encrypted_data.clone(), identifier)),
-            )
-            .await;
-            let decrypted = match result {
-                Ok(response) => {
-                    ForeignTryFrom::foreign_try_from((encrypted_data.clone(), response))
-                }
-                Err(err) => {
-                    logger::error!("Decryption error {:?}", err);
-                    Err(err.change_context(errors::CryptoError::DecodingFailed))
-                }
-            };
-            match decrypted {
-                Ok(de) => Ok(de),
-                Err(_) => {
-                    metrics::DECRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
-                    logger::info!("Fall back to Application Decryption");
-                    Self::batch_decrypt(encrypted_data, key, crypt_algo).await
+        #[allow(unused_variables)]
+        async fn batch_decrypt_via_api(
+            state: &KeyManagerState,
+            encrypted_data: FxHashMap<String, Encryption>,
+            identifier: Identifier,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
+            #[cfg(not(feature = "encryption_service"))]
+            {
+                Self::batch_decrypt(encrypted_data, key, crypt_algo).await
+            }
+            #[cfg(feature = "encryption_service")]
+            {
+                let result: Result<
+                    BatchDecryptDataResponse,
+                    error_stack::Report<errors::KeyManagerClientError>,
+                > = call_encryption_service(
+                    state,
+                    Method::POST,
+                    "data/decrypt",
+                    TransientBatchDecryptDataRequest::from((encrypted_data.clone(), identifier)),
+                )
+                .await;
+                let decrypted = match result {
+                    Ok(response) => {
+                        ForeignTryFrom::foreign_try_from((encrypted_data.clone(), response))
+                    }
+                    Err(err) => {
+                        logger::error!("Decryption error {:?}", err);
+                        Err(err.change_context(errors::CryptoError::DecodingFailed))
+                    }
+                };
+                match decrypted {
+                    Ok(de) => Ok(de),
+                    Err(_) => {
+                        metrics::DECRYPTION_API_FAILURES.add(&metrics::CONTEXT, 1, &[]);
+                        logger::info!("Fall back to Application Decryption");
+                        Self::batch_decrypt(encrypted_data, key, crypt_algo).await
+                    }
                 }
             }
         }
-    }
 
-    async fn batch_encrypt(
-        masked_data: FxHashMap<String, Secret<Vec<u8>, S>>,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
-        metrics::APPLICATION_ENCRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
-        masked_data
-            .into_iter()
-            .map(|(k, v)| {
-                Ok((
-                    k,
-                    Self::new(v.clone(), crypt_algo.encode_message(key, v.peek())?.into()),
-                ))
-            })
-            .collect()
-    }
+        async fn batch_encrypt(
+            masked_data: FxHashMap<String, Secret<Vec<u8>, S>>,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
+            metrics::APPLICATION_ENCRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
+            masked_data
+                .into_iter()
+                .map(|(k, v)| {
+                    Ok((
+                        k,
+                        Self::new(v.clone(), crypt_algo.encode_message(key, v.peek())?.into()),
+                    ))
+                })
+                .collect()
+        }
 
-    async fn batch_decrypt(
-        encrypted_data: FxHashMap<String, Encryption>,
-        key: &[u8],
-        crypt_algo: V,
-    ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
-        metrics::APPLICATION_DECRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
-        encrypted_data
-            .into_iter()
-            .map(|(k, v)| {
-                Ok((
-                    k,
-                    Self::new(
-                        crypt_algo
-                            .decode_message(key, v.clone().into_inner().clone())?
-                            .into(),
-                        v.into_inner(),
-                    ),
-                ))
-            })
-            .collect()
+        async fn batch_decrypt(
+            encrypted_data: FxHashMap<String, Encryption>,
+            key: &[u8],
+            crypt_algo: V,
+        ) -> CustomResult<FxHashMap<String, Self>, errors::CryptoError> {
+            metrics::APPLICATION_DECRYPTION_COUNT.add(&metrics::CONTEXT, 1, &[]);
+            encrypted_data
+                .into_iter()
+                .map(|(k, v)| {
+                    Ok((
+                        k,
+                        Self::new(
+                            crypt_algo
+                                .decode_message(key, v.clone().into_inner().clone())?
+                                .into(),
+                            v.into_inner(),
+                        ),
+                    ))
+                })
+                .collect()
+        }
     }
 }
-
 pub trait Lift<U> {
     type SelfWrapper<T>;
     type OtherWrapper<T, E>;
diff --git a/crates/router/src/core/admin.rs b/crates/router/src/core/admin.rs
index 5f3a820447d..78fc3760a47 100644
--- a/crates/router/src/core/admin.rs
+++ b/crates/router/src/core/admin.rs
@@ -2754,8 +2754,9 @@ pub(crate) fn validate_connector_auth_type(
 
 pub async fn transfer_key_store_to_key_manager(
     state: SessionState,
+    req: admin_types::MerchantKeyTransferRequest,
 ) -> RouterResponse<admin_types::TransferKeyResponse> {
-    let resp = transfer_encryption_key(&state).await?;
+    let resp = transfer_encryption_key(&state, req).await?;
 
     Ok(service_api::ApplicationResponse::Json(
         admin_types::TransferKeyResponse {
diff --git a/crates/router/src/core/encryption.rs b/crates/router/src/core/encryption.rs
index efd7d3bfeaa..fb690b56cb2 100644
--- a/crates/router/src/core/encryption.rs
+++ b/crates/router/src/core/encryption.rs
@@ -1,3 +1,4 @@
+use api_models::admin::MerchantKeyTransferRequest;
 use base64::Engine;
 use common_utils::{
     keymanager::transfer_key_to_key_manager,
@@ -11,10 +12,16 @@ use crate::{consts::BASE64_ENGINE, errors, types::domain::UserKeyStore, SessionS
 
 pub async fn transfer_encryption_key(
     state: &SessionState,
+    req: MerchantKeyTransferRequest,
 ) -> errors::CustomResult<usize, errors::ApiErrorResponse> {
     let db = &*state.store;
     let key_stores = db
-        .get_all_key_stores(&state.into(), &db.get_master_key().to_vec().into())
+        .get_all_key_stores(
+            &state.into(),
+            &db.get_master_key().to_vec().into(),
+            req.from,
+            req.limit,
+        )
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)?;
     send_request_to_key_service_for_merchant(state, key_stores).await
@@ -24,17 +31,18 @@ pub async fn send_request_to_key_service_for_merchant(
     state: &SessionState,
     keys: Vec<MerchantKeyStore>,
 ) -> errors::CustomResult<usize, errors::ApiErrorResponse> {
-    futures::future::try_join_all(keys.into_iter().map(|key| async move {
+    let total = keys.len();
+    for key in keys {
         let key_encoded = BASE64_ENGINE.encode(key.key.clone().into_inner().expose());
         let req = EncryptionTransferRequest {
             identifier: Identifier::Merchant(key.merchant_id.clone()),
             key: key_encoded,
         };
-        transfer_key_to_key_manager(&state.into(), req).await
-    }))
-    .await
-    .change_context(errors::ApiErrorResponse::InternalServerError)
-    .map(|v| v.len())
+        transfer_key_to_key_manager(&state.into(), req)
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)?;
+    }
+    Ok(total)
 }
 
 pub async fn send_request_to_key_service_for_user(
diff --git a/crates/router/src/core/user.rs b/crates/router/src/core/user.rs
index be33ccc69e1..e44493cff39 100644
--- a/crates/router/src/core/user.rs
+++ b/crates/router/src/core/user.rs
@@ -1926,6 +1926,7 @@ pub async fn generate_recovery_codes(
 
 pub async fn transfer_user_key_store_keymanager(
     state: SessionState,
+    req: user_api::UserKeyTransferRequest,
 ) -> UserResponse<user_api::UserTransferKeyResponse> {
     let db = &state.global_store;
 
@@ -1933,6 +1934,8 @@ pub async fn transfer_user_key_store_keymanager(
         .get_all_user_key_store(
             &(&state).into(),
             &state.store.get_master_key().to_vec().into(),
+            req.from,
+            req.limit,
         )
         .await
         .change_context(UserErrors::InternalServerError)?;
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index ed2357c1a1c..7062ea141e0 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -2193,8 +2193,12 @@ impl MerchantKeyStoreInterface for KafkaStore {
         &self,
         state: &KeyManagerState,
         key: &Secret<Vec<u8>>,
+        from: u32,
+        to: u32,
     ) -> CustomResult<Vec<domain::MerchantKeyStore>, errors::StorageError> {
-        self.diesel_store.get_all_key_stores(state, key).await
+        self.diesel_store
+            .get_all_key_stores(state, key, from, to)
+            .await
     }
 }
 
@@ -3067,8 +3071,12 @@ impl UserKeyStoreInterface for KafkaStore {
         &self,
         state: &KeyManagerState,
         key: &Secret<Vec<u8>>,
+        from: u32,
+        limit: u32,
     ) -> CustomResult<Vec<domain::UserKeyStore>, errors::StorageError> {
-        self.diesel_store.get_all_user_key_store(state, key).await
+        self.diesel_store
+            .get_all_user_key_store(state, key, from, limit)
+            .await
     }
 }
 
diff --git a/crates/router/src/db/merchant_key_store.rs b/crates/router/src/db/merchant_key_store.rs
index 1cc12796bc6..21c1a184425 100644
--- a/crates/router/src/db/merchant_key_store.rs
+++ b/crates/router/src/db/merchant_key_store.rs
@@ -49,6 +49,8 @@ pub trait MerchantKeyStoreInterface {
         &self,
         state: &KeyManagerState,
         key: &Secret<Vec<u8>>,
+        from: u32,
+        to: u32,
     ) -> CustomResult<Vec<domain::MerchantKeyStore>, errors::StorageError>;
 }
 
@@ -183,15 +185,17 @@ impl MerchantKeyStoreInterface for Store {
         &self,
         state: &KeyManagerState,
         key: &Secret<Vec<u8>>,
+        from: u32,
+        to: u32,
     ) -> CustomResult<Vec<domain::MerchantKeyStore>, errors::StorageError> {
         let conn = connection::pg_connection_read(self).await?;
-        let fetch_func = || async {
-            diesel_models::merchant_key_store::MerchantKeyStore::list_all_key_stores(&conn)
-                .await
-                .map_err(|err| report!(errors::StorageError::from(err)))
-        };
+        let stores = diesel_models::merchant_key_store::MerchantKeyStore::list_all_key_stores(
+            &conn, from, to,
+        )
+        .await
+        .map_err(|err| report!(errors::StorageError::from(err)))?;
 
-        futures::future::try_join_all(fetch_func().await?.into_iter().map(|key_store| async {
+        futures::future::try_join_all(stores.into_iter().map(|key_store| async {
             let merchant_id = key_store.merchant_id.clone();
             key_store
                 .convert(state, key, merchant_id)
@@ -295,6 +299,8 @@ impl MerchantKeyStoreInterface for MockDb {
         &self,
         state: &KeyManagerState,
         key: &Secret<Vec<u8>>,
+        _from: u32,
+        _to: u32,
     ) -> CustomResult<Vec<domain::MerchantKeyStore>, errors::StorageError> {
         let merchant_key_stores = self.merchant_key_store.lock().await;
 
diff --git a/crates/router/src/db/user_key_store.rs b/crates/router/src/db/user_key_store.rs
index e7dbe531c65..111989a293e 100644
--- a/crates/router/src/db/user_key_store.rs
+++ b/crates/router/src/db/user_key_store.rs
@@ -34,6 +34,8 @@ pub trait UserKeyStoreInterface {
         &self,
         state: &KeyManagerState,
         key: &Secret<Vec<u8>>,
+        from: u32,
+        limit: u32,
     ) -> CustomResult<Vec<domain::UserKeyStore>, errors::StorageError>;
 }
 
@@ -81,16 +83,17 @@ impl UserKeyStoreInterface for Store {
         &self,
         state: &KeyManagerState,
         key: &Secret<Vec<u8>>,
+        from: u32,
+        limit: u32,
     ) -> CustomResult<Vec<domain::UserKeyStore>, errors::StorageError> {
         let conn = connection::pg_connection_read(self).await?;
 
-        let fetch_func = || async {
-            diesel_models::user_key_store::UserKeyStore::get_all_user_key_stores(&conn)
-                .await
-                .map_err(|err| report!(errors::StorageError::from(err)))
-        };
-
-        futures::future::try_join_all(fetch_func().await?.into_iter().map(|key_store| async {
+        let key_stores = diesel_models::user_key_store::UserKeyStore::get_all_user_key_stores(
+            &conn, from, limit,
+        )
+        .await
+        .map_err(|err| report!(errors::StorageError::from(err)))?;
+        futures::future::try_join_all(key_stores.into_iter().map(|key_store| async {
             let user_id = key_store.user_id.clone();
             key_store
                 .convert(state, key, user_id)
@@ -137,6 +140,8 @@ impl UserKeyStoreInterface for MockDb {
         &self,
         state: &KeyManagerState,
         key: &Secret<Vec<u8>>,
+        _from: u32,
+        _limit: u32,
     ) -> CustomResult<Vec<domain::UserKeyStore>, errors::StorageError> {
         let user_key_store = self.user_key_store.lock().await;
 
diff --git a/crates/router/src/events/api_logs.rs b/crates/router/src/events/api_logs.rs
index 7bf66dd0951..9eb8f1021aa 100644
--- a/crates/router/src/events/api_logs.rs
+++ b/crates/router/src/events/api_logs.rs
@@ -1,6 +1,6 @@
 use actix_web::HttpRequest;
 pub use common_utils::events::{ApiEventMetric, ApiEventsType};
-use common_utils::impl_misc_api_event_type;
+use common_utils::impl_api_event_type;
 use router_env::{tracing_actix_web::RequestId, types::FlowMetric};
 use serde::Serialize;
 use time::OffsetDateTime;
@@ -98,24 +98,30 @@ impl KafkaMessage for ApiEvent {
     }
 }
 
-impl_misc_api_event_type!(
-    Config,
-    CreateFileRequest,
-    FileId,
-    AttachEvidenceRequest,
-    ConfigUpdate
+impl_api_event_type!(
+    Miscellaneous,
+    (
+        Config,
+        CreateFileRequest,
+        FileId,
+        AttachEvidenceRequest,
+        ConfigUpdate
+    )
 );
 
 #[cfg(feature = "dummy_connector")]
-impl_misc_api_event_type!(
-    DummyConnectorPaymentCompleteRequest,
-    DummyConnectorPaymentRequest,
-    DummyConnectorPaymentResponse,
-    DummyConnectorPaymentRetrieveRequest,
-    DummyConnectorPaymentConfirmRequest,
-    DummyConnectorRefundRetrieveRequest,
-    DummyConnectorRefundResponse,
-    DummyConnectorRefundRequest
+impl_api_event_type!(
+    Miscellaneous,
+    (
+        DummyConnectorPaymentCompleteRequest,
+        DummyConnectorPaymentRequest,
+        DummyConnectorPaymentResponse,
+        DummyConnectorPaymentRetrieveRequest,
+        DummyConnectorPaymentConfirmRequest,
+        DummyConnectorRefundRetrieveRequest,
+        DummyConnectorRefundResponse,
+        DummyConnectorRefundRequest
+    )
 );
 
 impl ApiEventMetric for PaymentsRedirectResponseData {
diff --git a/crates/router/src/routes/admin.rs b/crates/router/src/routes/admin.rs
index 74bb775d9c0..fe791bfaea9 100644
--- a/crates/router/src/routes/admin.rs
+++ b/crates/router/src/routes/admin.rs
@@ -658,14 +658,15 @@ pub async fn merchant_account_kv_status(
 pub async fn merchant_account_transfer_keys(
     state: web::Data<AppState>,
     req: HttpRequest,
+    payload: web::Json<api_models::admin::MerchantKeyTransferRequest>,
 ) -> HttpResponse {
     let flow = Flow::ConfigKeyFetch;
     api::server_wrap(
         flow,
         state,
         &req,
-        (),
-        |state, _, _, _| transfer_key_store_to_key_manager(state),
+        payload.into_inner(),
+        |state, _, req, _| transfer_key_store_to_key_manager(state, req),
         &auth::AdminApiAuth,
         api_locking::LockAction::NotApplicable,
     )
diff --git a/crates/router/src/routes/user.rs b/crates/router/src/routes/user.rs
index 29935bc8ab7..39839e423ea 100644
--- a/crates/router/src/routes/user.rs
+++ b/crates/router/src/routes/user.rs
@@ -896,15 +896,19 @@ pub async fn terminate_auth_select(
     .await
 }
 
-pub async fn transfer_user_key(state: web::Data<AppState>, req: HttpRequest) -> HttpResponse {
+pub async fn transfer_user_key(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    payload: web::Json<user_api::UserKeyTransferRequest>,
+) -> HttpResponse {
     let flow = Flow::UserTransferKey;
 
     Box::pin(api::server_wrap(
         flow,
         state.clone(),
         &req,
-        (),
-        |state, _, _, _| user_core::transfer_user_key_store_keymanager(state),
+        payload.into_inner(),
+        |state, _, req, _| user_core::transfer_user_key_store_keymanager(state, req),
         &auth::AdminApiAuth,
         api_locking::LockAction::NotApplicable,
     ))