Issue:
Bug: [FEAT] add key in keymanager



[FEAT] add key in keymanager

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->
- [x] New feature

## Description
<!-- Describe your changes in detail -->
This PR adds support to

- Create Key for every User and Merchant during User and Merchant Account Create.
- Add an API to transfer Keys to KeyManager
- Add MTLS for KeyManager Client

### Additional Changes
- [x] This PR modifies application configuration/environment variables

## 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).
-->
 - Create Key in Key Manager for every merchant

## 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)?
-->
### Tests for  Create a Merchant
-
```bash
curl --location 'http://localhost:8080/accounts' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: test_admin' \
--data-raw '{
  "merchant_id": "merchant_1720435957",
  "locker_id": "m0010",
  "merchant_name": "NewAge Retailer",
  "merchant_details": {
    "primary_contact_person": "John Test",
    "primary_email": "JohnTest@test.com",
    "primary_phone": "sunt laborum",
    "secondary_contact_person": "John Test2",
    "secondary_email": "JohnTest2@test.com",
    "secondary_phone": "cillum do dolor id",
    "website": "www.example.com",
    "about_business": "Online Retail with a wide selection of organic products for North America",
    "address": {
      "line1": "1467",
      "line2": "Harrison Street",
      "line3": "Harrison Street",
      "city": "San Fransico",
      "state": "California",
      "zip": "94122",
      "country": "US"
    }
  },
  "return_url": "https://google.com/success",
  "webhook_details": {
    "webhook_version": "1.0.1",
    "webhook_username": "ekart_retail",
    "webhook_password": "password_ekart@123",
    "payment_created_enabled": true,
    "payment_succeeded_enabled": true,
    "payment_failed_enabled": true
  },
  "sub_merchants_enabled": false,
  "metadata": {
    "city": "NY",
    "unit": "245"
  },
  "primary_business_details": [
    {
      "country": "US",
      "business": "default"
    }
  ]
}'
```
 - Verify `/key/create` to Keymanager was succeeded.
 The response status of `keymanager_response` should be 200

<img width="1334" alt="Screenshot 2024-07-11 at 5 01 27 PM" src="https://github.com/juspay/hyperswitch/assets/43412619/95dcf641-0908-4f4d-8803-53e435906220">

### Create Key for User

- Signin with the account and password
```
curl --location 'http://localhost:8080/user/v2/signin?token_only=true' \
--header 'Content-Type: application/json' \
--data-raw '{
 "email": "gagagagag@gagagagagag.com",
    "password": "gagagagag@1234"
}'

```
- Begin TOTP flow
```
curl --location 'http://localhost:8080/user/2fa/totp/begin' \
--header 'Authorization: Bearer spt'
```
- Verify that call to `/key/create` is successful in the logs

## 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:

### Tests for  Create a Merchant
-
```bash
curl --location 'http://localhost:8080/accounts' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: test_admin' \
--data-raw '{
  "merchant_id": "merchant_1720435957",
  "locker_id": "m0010",
  "merchant_name": "NewAge Retailer",
  "merchant_details": {
    "primary_contact_person": "John Test",
    "primary_email": "JohnTest@test.com",
    "primary_phone": "sunt laborum",
    "secondary_contact_person": "John Test2",
    "secondary_email": "JohnTest2@test.com",
    "secondary_phone": "cillum do dolor id",
    "website": "www.example.com",
    "about_business": "Online Retail with a wide selection of organic products for North America",
    "address": {
      "line1": "1467",
      "line2": "Harrison Street",
      "line3": "Harrison Street",
      "city": "San Fransico",
      "state": "California",
      "zip": "94122",
      "country": "US"
    }
  },
  "return_url": "https://google.com/success",
  "webhook_details": {
    "webhook_version": "1.0.1",
    "webhook_username": "ekart_retail",
    "webhook_password": "password_ekart@123",
    "payment_created_enabled": true,
    "payment_succeeded_enabled": true,
    "payment_failed_enabled": true
  },
  "sub_merchants_enabled": false,
  "metadata": {
    "city": "NY",
    "unit": "245"
  },
  "primary_business_details": [
    {
      "country": "US",
      "business": "default"
    }
  ]
}'
```
 - Verify `/key/create` to Keymanager was succeeded.
 The response status of `keymanager_response` should be 200

<img width="1334" alt="Screenshot 2024-07-11 at 5 01 27 PM" src="https://github.com/juspay/hyperswitch/assets/43412619/95dcf641-0908-4f4d-8803-53e435906220">

### Create Key for User

- Signin with the account and password
```
curl --location 'http://localhost:8080/user/v2/signin?token_only=true' \
--header 'Content-Type: application/json' \
--data-raw '{
 "email": "gagagagag@gagagagagag.com",
    "password": "gagagagag@1234"
}'

```
- Begin TOTP flow
```
curl --location 'http://localhost:8080/user/2fa/totp/begin' \
--header 'Authorization: Bearer spt'
```
- Verify that call to `/key/create` is successful in the logs

Patch:
diff --git a/config/config.example.toml b/config/config.example.toml
index e4a15e7fb12..5a734237513 100644
--- a/config/config.example.toml
+++ b/config/config.example.toml
@@ -28,6 +28,10 @@ certificate = "/path/to/certificate.pem"
 idle_pool_connection_timeout = 90 # Timeout for idle pool connections (defaults to 90s)
 
 
+# Configuration for the Key Manager Service
+[key_manager]
+url = "http://localhost:5000" # URL of the encryption service
+
 # Main SQL data store credentials
 [master_database]
 username = "db_user"      # DB Username
diff --git a/config/deployments/env_specific.toml b/config/deployments/env_specific.toml
index 67f32396b74..3f7fb52f502 100644
--- a/config/deployments/env_specific.toml
+++ b/config/deployments/env_specific.toml
@@ -248,6 +248,10 @@ payment_intents = "hyperswitch-payment-intent-events"
 refunds = "hyperswitch-refund-events"
 disputes = "hyperswitch-dispute-events"
 
+# Configuration for the Key Manager Service
+[key_manager]
+url = "http://localhost:5000" # URL of the encryption service
+
 # This section provides some secret values.
 [secrets]
 master_enc_key = "sample_key"            # Master Encryption key used to encrypt merchant wise encryption key. Should be 32-byte long.
diff --git a/config/development.toml b/config/development.toml
index b3548b35d06..23682ab1542 100644
--- a/config/development.toml
+++ b/config/development.toml
@@ -12,6 +12,9 @@ metrics_enabled = false
 use_xray_generator = false
 bg_metrics_collection_interval_in_secs = 15
 
+[key_manager]
+url = "http://localhost:5000"
+
 # TODO: Update database credentials before running application
 [master_database]
 username = "db_user"
diff --git a/config/docker_compose.toml b/config/docker_compose.toml
index b11ca005741..10e5545c1bd 100644
--- a/config/docker_compose.toml
+++ b/config/docker_compose.toml
@@ -90,6 +90,9 @@ default_command_timeout = 30
 unresponsive_timeout = 10
 max_feed_count = 200
 
+[key_manager]
+url = "http://localhost:5000"
+
 [cors]
 max_age = 30
 # origins = "http://localhost:8080,http://localhost:9000"
diff --git a/crates/api_models/src/admin.rs b/crates/api_models/src/admin.rs
index e9b84285847..ec9172b7fdf 100644
--- a/crates/api_models/src/admin.rs
+++ b/crates/api_models/src/admin.rs
@@ -994,6 +994,12 @@ pub struct ToggleKVResponse {
     pub kv_enabled: bool,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct TransferKeyResponse {
+    /// The identifier for the Merchant Account
+    #[schema(example = 32)]
+    pub total_transferred: usize,
+}
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct ToggleKVRequest {
     #[serde(skip_deserializing)]
diff --git a/crates/api_models/src/events.rs b/crates/api_models/src/events.rs
index 4dcf1a3896a..3757dd98d78 100644
--- a/crates/api_models/src/events.rs
+++ b/crates/api_models/src/events.rs
@@ -80,6 +80,7 @@ impl_misc_api_event_type!(
     ToggleKVRequest,
     ToggleAllKVRequest,
     ToggleAllKVResponse,
+    TransferKeyResponse,
     MerchantAccountDeleteResponse,
     MerchantAccountUpdate,
     CardInfoResponse,
diff --git a/crates/api_models/src/events/user.rs b/crates/api_models/src/events/user.rs
index b357a3389d9..b8c893ae25d 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,
-    VerifyEmailRequest, VerifyRecoveryCodeRequest, VerifyTotpRequest,
+    UserTransferKeyResponse, VerifyEmailRequest, VerifyRecoveryCodeRequest, VerifyTotpRequest,
 };
 
 impl ApiEventMetric for DashboardEntryResponse {
@@ -85,6 +85,7 @@ common_utils::impl_misc_api_event_type!(
     UpdateUserAuthenticationMethodRequest,
     GetSsoAuthUrlRequest,
     SsoSignInRequest,
+    UserTransferKeyResponse,
     AuthSelectRequest
 );
 
diff --git a/crates/api_models/src/user.rs b/crates/api_models/src/user.rs
index 43b476c9f67..7c0b21f9679 100644
--- a/crates/api_models/src/user.rs
+++ b/crates/api_models/src/user.rs
@@ -379,3 +379,8 @@ pub struct AuthIdQueryParam {
 pub struct AuthSelectRequest {
     pub id: Option<String>,
 }
+
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
+pub struct UserTransferKeyResponse {
+    pub total_transferred: usize,
+}
diff --git a/crates/common_utils/Cargo.toml b/crates/common_utils/Cargo.toml
index 45953210701..f6c3d8b6d0d 100644
--- a/crates/common_utils/Cargo.toml
+++ b/crates/common_utils/Cargo.toml
@@ -8,6 +8,8 @@ readme = "README.md"
 license.workspace = true
 
 [features]
+keymanager = ["dep:router_env"]
+keymanager_mtls = ["reqwest/rustls-tls"]
 signals = ["dep:signal-hook-tokio", "dep:signal-hook", "dep:tokio", "dep:router_env", "dep:futures"]
 async_ext = ["dep:async-trait", "dep:futures"]
 logs = ["dep:router_env"]
diff --git a/crates/common_utils/src/errors.rs b/crates/common_utils/src/errors.rs
index d5e86985041..23b4f2be4fe 100644
--- a/crates/common_utils/src/errors.rs
+++ b/crates/common_utils/src/errors.rs
@@ -145,6 +145,42 @@ where
     }
 }
 
+#[allow(missing_docs)]
+#[derive(Debug, thiserror::Error)]
+pub enum KeyManagerClientError {
+    #[error("Failed to construct header from the given value")]
+    FailedtoConstructHeader,
+    #[error("Failed to send request to Keymanager")]
+    RequestNotSent(String),
+    #[error("URL encoding of request failed")]
+    UrlEncodingFailed,
+    #[error("Failed to build the reqwest client ")]
+    ClientConstructionFailed,
+    #[error("Failed to send the request to Keymanager")]
+    RequestSendFailed,
+    #[error("Internal Server Error Received {0:?}")]
+    InternalServerError(bytes::Bytes),
+    #[error("Bad request received {0:?}")]
+    BadRequest(bytes::Bytes),
+    #[error("Unexpected Error occurred while calling the KeyManager")]
+    Unexpected(bytes::Bytes),
+    #[error("Response Decoding failed")]
+    ResponseDecodingFailed,
+}
+
+#[allow(missing_docs)]
+#[derive(Debug, thiserror::Error)]
+pub enum KeyManagerError {
+    #[error("Failed to add key to the KeyManager")]
+    KeyAddFailed,
+    #[error("Failed to transfer the key to the KeyManager")]
+    KeyTransferFailed,
+    #[error("Failed to Encrypt the data in the KeyManager")]
+    EncryptionFailed,
+    #[error("Failed to Decrypt the data in the KeyManager")]
+    DecryptionFailed,
+}
+
 /// Allow [error_stack::Report] to convert between error types
 /// This auto-implements [ReportSwitchExt] for the corresponding errors
 pub trait ErrorSwitch<T> {
diff --git a/crates/common_utils/src/keymanager.rs b/crates/common_utils/src/keymanager.rs
new file mode 100644
index 00000000000..ecdb9b63ad5
--- /dev/null
+++ b/crates/common_utils/src/keymanager.rs
@@ -0,0 +1,175 @@
+//! Consists of all the common functions to use the Keymanager.
+
+use core::fmt::Debug;
+use std::str::FromStr;
+
+use error_stack::ResultExt;
+use http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode};
+#[cfg(feature = "keymanager_mtls")]
+use masking::PeekInterface;
+use once_cell::sync::OnceCell;
+use router_env::{instrument, logger, tracing};
+
+use crate::{
+    errors,
+    types::keymanager::{
+        DataKeyCreateResponse, EncryptionCreateRequest, EncryptionTransferRequest, KeyManagerState,
+    },
+};
+
+const CONTENT_TYPE: &str = "Content-Type";
+static ENCRYPTION_API_CLIENT: OnceCell<reqwest::Client> = OnceCell::new();
+
+/// Get keymanager client constructed from the url and state
+#[instrument(skip_all)]
+#[allow(unused_mut)]
+fn get_api_encryption_client(
+    state: &KeyManagerState,
+) -> errors::CustomResult<reqwest::Client, errors::KeyManagerClientError> {
+    let get_client = || {
+        let mut client = reqwest::Client::builder()
+            .redirect(reqwest::redirect::Policy::none())
+            .pool_idle_timeout(std::time::Duration::from_secs(
+                state.client_idle_timeout.unwrap_or_default(),
+            ));
+
+        #[cfg(feature = "keymanager_mtls")]
+        {
+            let cert = state.cert.clone();
+            let ca = state.ca.clone();
+
+            let identity = reqwest::Identity::from_pem(cert.peek().as_ref())
+                .change_context(errors::KeyManagerClientError::ClientConstructionFailed)?;
+            let ca_cert = reqwest::Certificate::from_pem(ca.peek().as_ref())
+                .change_context(errors::KeyManagerClientError::ClientConstructionFailed)?;
+
+            client = client
+                .use_rustls_tls()
+                .identity(identity)
+                .add_root_certificate(ca_cert)
+                .https_only(true);
+        }
+
+        client
+            .build()
+            .change_context(errors::KeyManagerClientError::ClientConstructionFailed)
+    };
+
+    Ok(ENCRYPTION_API_CLIENT.get_or_try_init(get_client)?.clone())
+}
+
+/// Generic function to send the request to keymanager
+#[instrument(skip_all)]
+pub async fn send_encryption_request<T>(
+    state: &KeyManagerState,
+    headers: HeaderMap,
+    url: String,
+    method: Method,
+    request_body: T,
+) -> errors::CustomResult<reqwest::Response, errors::KeyManagerClientError>
+where
+    T: serde::Serialize,
+{
+    let client = get_api_encryption_client(state)?;
+    let url = reqwest::Url::parse(&url)
+        .change_context(errors::KeyManagerClientError::UrlEncodingFailed)?;
+
+    client
+        .request(method, url)
+        .json(&request_body)
+        .headers(headers)
+        .send()
+        .await
+        .change_context(errors::KeyManagerClientError::RequestNotSent(
+            "Unable to send request to encryption service".to_string(),
+        ))
+}
+
+/// Generic function to call the Keymanager and parse the response back
+#[instrument(skip_all)]
+pub async fn call_encryption_service<T, R>(
+    state: &KeyManagerState,
+    method: Method,
+    endpoint: &str,
+    request_body: T,
+) -> errors::CustomResult<R, errors::KeyManagerClientError>
+where
+    T: serde::Serialize + Send + Sync + 'static + Debug,
+    R: serde::de::DeserializeOwned,
+{
+    let url = format!("{}/{endpoint}", &state.url);
+
+    logger::info!(key_manager_request=?request_body);
+
+    let response = send_encryption_request(
+        state,
+        HeaderMap::from_iter(
+            vec![(
+                HeaderName::from_str(CONTENT_TYPE)
+                    .change_context(errors::KeyManagerClientError::FailedtoConstructHeader)?,
+                HeaderValue::from_str("application/json")
+                    .change_context(errors::KeyManagerClientError::FailedtoConstructHeader)?,
+            )]
+            .into_iter(),
+        ),
+        url,
+        method,
+        request_body,
+    )
+    .await
+    .map_err(|err| err.change_context(errors::KeyManagerClientError::RequestSendFailed))?;
+
+    logger::info!(key_manager_response=?response);
+
+    match response.status() {
+        StatusCode::OK => response
+            .json::<R>()
+            .await
+            .change_context(errors::KeyManagerClientError::ResponseDecodingFailed),
+        StatusCode::INTERNAL_SERVER_ERROR => {
+            Err(errors::KeyManagerClientError::InternalServerError(
+                response
+                    .bytes()
+                    .await
+                    .change_context(errors::KeyManagerClientError::ResponseDecodingFailed)?,
+            )
+            .into())
+        }
+        StatusCode::BAD_REQUEST => Err(errors::KeyManagerClientError::BadRequest(
+            response
+                .bytes()
+                .await
+                .change_context(errors::KeyManagerClientError::ResponseDecodingFailed)?,
+        )
+        .into()),
+        _ => Err(errors::KeyManagerClientError::Unexpected(
+            response
+                .bytes()
+                .await
+                .change_context(errors::KeyManagerClientError::ResponseDecodingFailed)?,
+        )
+        .into()),
+    }
+}
+
+/// A function to create the key in keymanager
+#[instrument(skip_all)]
+pub async fn create_key_in_key_manager(
+    state: &KeyManagerState,
+    request_body: EncryptionCreateRequest,
+) -> errors::CustomResult<DataKeyCreateResponse, errors::KeyManagerError> {
+    call_encryption_service(state, Method::POST, "key/create", request_body)
+        .await
+        .change_context(errors::KeyManagerError::KeyAddFailed)
+}
+
+/// A function to transfer the key in keymanager
+#[instrument(skip_all)]
+pub async fn transfer_key_to_key_manager(
+    state: &KeyManagerState,
+    request_body: EncryptionTransferRequest,
+) -> errors::CustomResult<DataKeyCreateResponse, errors::KeyManagerError> {
+    call_encryption_service(state, Method::POST, "key/transfer", request_body)
+        .await
+        .change_context(errors::KeyManagerError::KeyTransferFailed)
+}
diff --git a/crates/common_utils/src/lib.rs b/crates/common_utils/src/lib.rs
index 39826944a1b..9b339fb956e 100644
--- a/crates/common_utils/src/lib.rs
+++ b/crates/common_utils/src/lib.rs
@@ -19,6 +19,8 @@ pub mod events;
 pub mod ext_traits;
 pub mod fp_utils;
 pub mod id_type;
+#[cfg(feature = "keymanager")]
+pub mod keymanager;
 pub mod link_utils;
 pub mod macros;
 pub mod new_type;
diff --git a/crates/common_utils/src/types.rs b/crates/common_utils/src/types.rs
index f469a0299ff..02b49f52730 100644
--- a/crates/common_utils/src/types.rs
+++ b/crates/common_utils/src/types.rs
@@ -1,4 +1,6 @@
 //! Types that can be used in other crates
+pub mod keymanager;
+
 use std::{
     fmt::Display,
     ops::{Add, Sub},
diff --git a/crates/common_utils/src/types/keymanager.rs b/crates/common_utils/src/types/keymanager.rs
new file mode 100644
index 00000000000..356ccdd4832
--- /dev/null
+++ b/crates/common_utils/src/types/keymanager.rs
@@ -0,0 +1,41 @@
+#![allow(missing_docs)]
+
+#[cfg(feature = "keymanager_mtls")]
+use masking::Secret;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug)]
+pub struct KeyManagerState {
+    pub url: String,
+    pub client_idle_timeout: Option<u64>,
+    #[cfg(feature = "keymanager_mtls")]
+    pub ca: Secret<String>,
+    #[cfg(feature = "keymanager_mtls")]
+    pub cert: Secret<String>,
+}
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq, Clone)]
+#[serde(tag = "data_identifier", content = "key_identifier")]
+pub enum Identifier {
+    User(String),
+    Merchant(String),
+}
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq, Clone)]
+pub struct EncryptionCreateRequest {
+    #[serde(flatten)]
+    pub identifier: Identifier,
+}
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq, Clone)]
+pub struct EncryptionTransferRequest {
+    #[serde(flatten)]
+    pub identifier: Identifier,
+    pub key: String,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct DataKeyCreateResponse {
+    #[serde(flatten)]
+    pub identifier: Identifier,
+    pub key_version: String,
+}
diff --git a/crates/diesel_models/src/query/merchant_key_store.rs b/crates/diesel_models/src/query/merchant_key_store.rs
index 1be82cb2c18..fc4d09c4989 100644
--- a/crates/diesel_models/src/query/merchant_key_store.rs
+++ b/crates/diesel_models/src/query/merchant_key_store.rs
@@ -54,4 +54,20 @@ impl MerchantKeyStore {
         )
         .await
     }
+
+    pub async fn list_all_key_stores(conn: &PgPooledConn) -> StorageResult<Vec<Self>> {
+        generics::generic_filter::<
+            <Self as HasTable>::Table,
+            _,
+            <<Self as HasTable>::Table as diesel::Table>::PrimaryKey,
+            _,
+        >(
+            conn,
+            dsl::merchant_id.ne_all(vec!["".to_string()]),
+            None,
+            None,
+            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 42dfe223b1a..5aad28bb562 100644
--- a/crates/diesel_models/src/query/user_key_store.rs
+++ b/crates/diesel_models/src/query/user_key_store.rs
@@ -14,6 +14,22 @@ impl UserKeyStoreNew {
 }
 
 impl UserKeyStore {
+    pub async fn get_all_user_key_stores(conn: &PgPooledConn) -> StorageResult<Vec<Self>> {
+        generics::generic_filter::<
+            <Self as HasTable>::Table,
+            _,
+            <<Self as HasTable>::Table as diesel::Table>::PrimaryKey,
+            _,
+        >(
+            conn,
+            dsl::user_id.ne_all(vec!["".to_string()]),
+            None,
+            None,
+            None,
+        )
+        .await
+    }
+
     pub async fn find_by_user_id(conn: &PgPooledConn, user_id: &str) -> StorageResult<Self> {
         generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
             conn,
diff --git a/crates/router/Cargo.toml b/crates/router/Cargo.toml
index f23029e36d6..4373c7fb169 100644
--- a/crates/router/Cargo.toml
+++ b/crates/router/Cargo.toml
@@ -11,10 +11,12 @@ license.workspace = true
 [features]
 default = ["kv_store", "stripe", "oltp", "olap", "backwards_compatibility", "accounts_cache", "dummy_connector", "payouts", "payout_retry", "business_profile_routing", "connector_choice_mca_id", "profile_specific_fallback_routing", "retry", "frm", "tls"]
 tls = ["actix-web/rustls-0_22"]
+keymanager_mtls = ["reqwest/rustls-tls","common_utils/keymanager_mtls"]
 email = ["external_services/email", "scheduler/email", "olap"]
+keymanager_create = []
 frm = ["api_models/frm", "hyperswitch_domain_models/frm"]
 stripe = ["dep:serde_qs"]
-release = ["stripe", "email", "backwards_compatibility", "business_profile_routing", "accounts_cache", "kv_store", "connector_choice_mca_id", "profile_specific_fallback_routing", "vergen", "recon", "external_services/aws_kms", "external_services/aws_s3"]
+release = ["stripe", "email", "backwards_compatibility", "business_profile_routing", "accounts_cache", "kv_store", "connector_choice_mca_id", "profile_specific_fallback_routing", "vergen", "recon", "external_services/aws_kms", "external_services/aws_s3","keymanager_mtls","keymanager_create"]
 olap = ["hyperswitch_domain_models/olap", "storage_impl/olap", "scheduler/olap", "api_models/olap", "dep:analytics"]
 oltp = ["storage_impl/oltp"]
 kv_store = ["scheduler/kv_store"]
@@ -107,7 +109,7 @@ api_models = { version = "0.1.0", path = "../api_models", features = ["errors"]
 analytics = { version = "0.1.0", path = "../analytics", optional = true }
 cards = { version = "0.1.0", path = "../cards" }
 common_enums = { version = "0.1.0", path = "../common_enums" }
-common_utils = { version = "0.1.0", path = "../common_utils", features = ["signals", "async_ext", "logs", "metrics"] }
+common_utils = { version = "0.1.0", path = "../common_utils", features = ["signals", "async_ext", "logs", "metrics","keymanager"] }
 hyperswitch_constraint_graph = { version = "0.1.0", path = "../hyperswitch_constraint_graph" }
 currency_conversion = { version = "0.1.0", path = "../currency_conversion" }
 hyperswitch_domain_models = { version = "0.1.0", path = "../hyperswitch_domain_models", default-features = false }
diff --git a/crates/router/src/configs/defaults.rs b/crates/router/src/configs/defaults.rs
index bab567899f5..819a5a06cb4 100644
--- a/crates/router/src/configs/defaults.rs
+++ b/crates/router/src/configs/defaults.rs
@@ -8810,3 +8810,15 @@ impl Default for super::settings::ApiKeys {
         }
     }
 }
+
+impl Default for super::settings::KeyManagerConfig {
+    fn default() -> Self {
+        Self {
+            url: String::from("localhost:5000"),
+            #[cfg(feature = "keymanager_mtls")]
+            ca: String::default().into(),
+            #[cfg(feature = "keymanager_mtls")]
+            cert: String::default().into(),
+        }
+    }
+}
diff --git a/crates/router/src/configs/secrets_transformers.rs b/crates/router/src/configs/secrets_transformers.rs
index 1ab2e519b7a..884feee9124 100644
--- a/crates/router/src/configs/secrets_transformers.rs
+++ b/crates/router/src/configs/secrets_transformers.rs
@@ -197,6 +197,35 @@ impl SecretsHandler for settings::PaymentMethodAuth {
     }
 }
 
+#[async_trait::async_trait]
+impl SecretsHandler for settings::KeyManagerConfig {
+    async fn convert_to_raw_secret(
+        value: SecretStateContainer<Self, SecuredSecret>,
+        _secret_management_client: &dyn SecretManagementInterface,
+    ) -> CustomResult<SecretStateContainer<Self, RawSecret>, SecretsManagementError> {
+        #[cfg(feature = "keymanager_mtls")]
+        let keyconfig = value.get_inner();
+
+        #[cfg(feature = "keymanager_mtls")]
+        let ca = _secret_management_client
+            .get_secret(keyconfig.ca.clone())
+            .await?;
+
+        #[cfg(feature = "keymanager_mtls")]
+        let cert = _secret_management_client
+            .get_secret(keyconfig.cert.clone())
+            .await?;
+
+        Ok(value.transition_state(|keyconfig| Self {
+            #[cfg(feature = "keymanager_mtls")]
+            ca,
+            #[cfg(feature = "keymanager_mtls")]
+            cert,
+            ..keyconfig
+        }))
+    }
+}
+
 #[async_trait::async_trait]
 impl SecretsHandler for settings::Secrets {
     async fn convert_to_raw_secret(
@@ -318,6 +347,14 @@ pub(crate) async fn fetch_raw_secrets(
     .await
     .expect("Failed to decrypt payment method auth configs");
 
+    #[allow(clippy::expect_used)]
+    let key_manager = settings::KeyManagerConfig::convert_to_raw_secret(
+        conf.key_manager,
+        secret_management_client,
+    )
+    .await
+    .expect("Failed to decrypt keymanager configs");
+
     #[allow(clippy::expect_used)]
     let user_auth_methods = settings::UserAuthMethodSettings::convert_to_raw_secret(
         conf.user_auth_methods,
@@ -337,6 +374,7 @@ pub(crate) async fn fetch_raw_secrets(
         secrets_management: conf.secrets_management,
         proxy: conf.proxy,
         env: conf.env,
+        key_manager,
         #[cfg(feature = "olap")]
         replica_database,
         secrets,
diff --git a/crates/router/src/configs/settings.rs b/crates/router/src/configs/settings.rs
index a29d7e1502d..33024a6e112 100644
--- a/crates/router/src/configs/settings.rs
+++ b/crates/router/src/configs/settings.rs
@@ -69,6 +69,7 @@ pub struct Settings<S: SecretState> {
     pub log: Log,
     pub secrets: SecretStateContainer<Secrets, S>,
     pub locker: Locker,
+    pub key_manager: SecretStateContainer<KeyManagerConfig, S>,
     pub connectors: Connectors,
     pub forex_api: SecretStateContainer<ForexApi, S>,
     pub refund: Refund,
@@ -213,6 +214,15 @@ pub struct KvConfig {
     pub soft_kill: Option<bool>,
 }
 
+#[derive(Debug, Deserialize, Clone)]
+pub struct KeyManagerConfig {
+    pub url: String,
+    #[cfg(feature = "keymanager_mtls")]
+    pub cert: Secret<String>,
+    #[cfg(feature = "keymanager_mtls")]
+    pub ca: Secret<String>,
+}
+
 #[derive(Debug, Deserialize, Clone, Default)]
 pub struct GenericLink {
     pub payment_method_collect: GenericLinkEnvConfig,
diff --git a/crates/router/src/core.rs b/crates/router/src/core.rs
index 7402d9730fd..5dc4e295919 100644
--- a/crates/router/src/core.rs
+++ b/crates/router/src/core.rs
@@ -14,6 +14,7 @@ pub mod connector_onboarding;
 pub mod currency;
 pub mod customers;
 pub mod disputes;
+pub mod encryption;
 pub mod errors;
 pub mod files;
 #[cfg(feature = "frm")]
diff --git a/crates/router/src/core/admin.rs b/crates/router/src/core/admin.rs
index dc72ec2ba89..67b3a05adb7 100644
--- a/crates/router/src/core/admin.rs
+++ b/crates/router/src/core/admin.rs
@@ -9,6 +9,8 @@ use common_utils::{
     ext_traits::{AsyncExt, ConfigExt, Encode, ValueExt},
     pii,
 };
+#[cfg(all(feature = "keymanager_create", feature = "olap"))]
+use common_utils::{keymanager, types::keymanager as km_types};
 use diesel_models::configs;
 use error_stack::{report, FutureExt, ResultExt};
 use futures::future::try_join_all;
@@ -22,6 +24,7 @@ use crate::types::transformers::ForeignFrom;
 use crate::{
     consts,
     core::{
+        encryption::transfer_encryption_key,
         errors::{self, RouterResponse, RouterResult, StorageErrorExt},
         payments::helpers,
         routing::helpers as routing_helpers,
@@ -125,6 +128,19 @@ pub async fn create_merchant_account(
         .create_domain_model_from_request(db, key_store.clone())
         .await?;
 
+    #[cfg(feature = "keymanager_create")]
+    {
+        keymanager::create_key_in_key_manager(
+            &(&state).into(),
+            km_types::EncryptionCreateRequest {
+                identifier: km_types::Identifier::Merchant(merchant_id.clone()),
+            },
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::DuplicateMerchantAccount)
+        .attach_printable("Failed to insert key to KeyManager")?;
+    }
+
     db.insert_merchant_key_store(key_store.clone(), &master_key.to_vec().into())
         .await
         .to_duplicate_response(errors::ApiErrorResponse::DuplicateMerchantAccount)?;
@@ -2491,6 +2507,18 @@ pub(crate) fn validate_connector_auth_type(
     }
 }
 
+pub async fn transfer_key_store_to_key_manager(
+    state: SessionState,
+) -> RouterResponse<admin_types::TransferKeyResponse> {
+    let resp = transfer_encryption_key(&state).await?;
+
+    Ok(service_api::ApplicationResponse::Json(
+        admin_types::TransferKeyResponse {
+            total_transferred: resp,
+        },
+    ))
+}
+
 #[cfg(feature = "dummy_connector")]
 pub async fn validate_dummy_connector_enabled(
     state: &SessionState,
diff --git a/crates/router/src/core/encryption.rs b/crates/router/src/core/encryption.rs
new file mode 100644
index 00000000000..7f7945bfd1c
--- /dev/null
+++ b/crates/router/src/core/encryption.rs
@@ -0,0 +1,55 @@
+use base64::Engine;
+use common_utils::{
+    keymanager::transfer_key_to_key_manager,
+    types::keymanager::{EncryptionTransferRequest, Identifier},
+};
+use error_stack::ResultExt;
+use hyperswitch_domain_models::merchant_key_store::MerchantKeyStore;
+use masking::ExposeInterface;
+
+use crate::{consts::BASE64_ENGINE, errors, types::domain::UserKeyStore, SessionState};
+
+pub async fn transfer_encryption_key(
+    state: &SessionState,
+) -> errors::CustomResult<usize, errors::ApiErrorResponse> {
+    let db = &*state.store;
+    let key_stores = db
+        .get_all_key_stores(&db.get_master_key().to_vec().into())
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)?;
+    send_request_to_key_service_for_merchant(state, key_stores).await
+}
+
+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 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())
+}
+
+pub async fn send_request_to_key_service_for_user(
+    state: &SessionState,
+    keys: Vec<UserKeyStore>,
+) -> errors::CustomResult<usize, errors::ApiErrorResponse> {
+    futures::future::try_join_all(keys.into_iter().map(|key| async move {
+        let key_encoded = BASE64_ENGINE.encode(key.key.clone().into_inner().expose());
+        let req = EncryptionTransferRequest {
+            identifier: Identifier::User(key.user_id.clone()),
+            key: key_encoded,
+        };
+        transfer_key_to_key_manager(&state.into(), req).await
+    }))
+    .await
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .map(|v| v.len())
+}
diff --git a/crates/router/src/core/user.rs b/crates/router/src/core/user.rs
index 2abcb5ea78b..bbc0336abcf 100644
--- a/crates/router/src/core/user.rs
+++ b/crates/router/src/core/user.rs
@@ -27,6 +27,7 @@ use super::errors::{StorageErrorExt, UserErrors, UserResponse, UserResult};
 use crate::services::email::types as email_types;
 use crate::{
     consts,
+    core::encryption::send_request_to_key_service_for_user,
     db::domain::user_authentication_method::DEFAULT_USER_AUTH_METHOD,
     routes::{app::ReqState, SessionState},
     services::{authentication as auth, authorization::roles, openidconnect, ApplicationResponse},
@@ -1911,6 +1912,25 @@ pub async fn generate_recovery_codes(
     }))
 }
 
+pub async fn transfer_user_key_store_keymanager(
+    state: SessionState,
+) -> UserResponse<user_api::UserTransferKeyResponse> {
+    let db = &state.global_store;
+
+    let key_stores = db
+        .get_all_user_key_store(&state.store.get_master_key().to_vec().into())
+        .await
+        .change_context(UserErrors::InternalServerError)?;
+
+    Ok(ApplicationResponse::Json(
+        user_api::UserTransferKeyResponse {
+            total_transferred: send_request_to_key_service_for_user(&state, key_stores)
+                .await
+                .change_context(UserErrors::InternalServerError)?,
+        },
+    ))
+}
+
 pub async fn verify_recovery_code(
     state: SessionState,
     user_token: auth::UserIdFromAuth,
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index edf8a53346c..ce0821c6322 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -2096,6 +2096,12 @@ impl MerchantKeyStoreInterface for KafkaStore {
             .list_multiple_key_stores(merchant_ids, key)
             .await
     }
+    async fn get_all_key_stores(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> CustomResult<Vec<domain::MerchantKeyStore>, errors::StorageError> {
+        self.diesel_store.get_all_key_stores(key).await
+    }
 }
 
 #[async_trait::async_trait]
@@ -2958,6 +2964,13 @@ impl UserKeyStoreInterface for KafkaStore {
             .get_user_key_store_by_user_id(user_id, key)
             .await
     }
+
+    async fn get_all_user_key_store(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> CustomResult<Vec<domain::UserKeyStore>, errors::StorageError> {
+        self.diesel_store.get_all_user_key_store(key).await
+    }
 }
 
 #[async_trait::async_trait]
diff --git a/crates/router/src/db/merchant_key_store.rs b/crates/router/src/db/merchant_key_store.rs
index 77eb2609835..42a862e2d3d 100644
--- a/crates/router/src/db/merchant_key_store.rs
+++ b/crates/router/src/db/merchant_key_store.rs
@@ -40,6 +40,11 @@ pub trait MerchantKeyStoreInterface {
         merchant_ids: Vec<String>,
         key: &Secret<Vec<u8>>,
     ) -> CustomResult<Vec<domain::MerchantKeyStore>, errors::StorageError>;
+
+    async fn get_all_key_stores(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> CustomResult<Vec<domain::MerchantKeyStore>, errors::StorageError>;
 }
 
 #[async_trait::async_trait]
@@ -163,6 +168,26 @@ impl MerchantKeyStoreInterface for Store {
         }))
         .await
     }
+
+    async fn get_all_key_stores(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> 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)))
+        };
+
+        futures::future::try_join_all(fetch_func().await?.into_iter().map(|key_store| async {
+            key_store
+                .convert(key)
+                .await
+                .change_context(errors::StorageError::DecryptionError)
+        }))
+        .await
+    }
 }
 
 #[async_trait::async_trait]
@@ -251,6 +276,21 @@ impl MerchantKeyStoreInterface for MockDb {
         )
         .await
     }
+    async fn get_all_key_stores(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> CustomResult<Vec<domain::MerchantKeyStore>, errors::StorageError> {
+        let merchant_key_stores = self.merchant_key_store.lock().await;
+
+        futures::future::try_join_all(merchant_key_stores.iter().map(|merchant_key| async {
+            merchant_key
+                .to_owned()
+                .convert(key)
+                .await
+                .change_context(errors::StorageError::DecryptionError)
+        }))
+        .await
+    }
 }
 
 #[cfg(test)]
diff --git a/crates/router/src/db/user_key_store.rs b/crates/router/src/db/user_key_store.rs
index e08d17d280e..c7e48037e96 100644
--- a/crates/router/src/db/user_key_store.rs
+++ b/crates/router/src/db/user_key_store.rs
@@ -27,6 +27,11 @@ pub trait UserKeyStoreInterface {
         user_id: &str,
         key: &Secret<Vec<u8>>,
     ) -> CustomResult<domain::UserKeyStore, errors::StorageError>;
+
+    async fn get_all_user_key_store(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> CustomResult<Vec<domain::UserKeyStore>, errors::StorageError>;
 }
 
 #[async_trait::async_trait]
@@ -65,6 +70,27 @@ impl UserKeyStoreInterface for Store {
             .await
             .change_context(errors::StorageError::DecryptionError)
     }
+
+    async fn get_all_user_key_store(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> 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 {
+            key_store
+                .convert(key)
+                .await
+                .change_context(errors::StorageError::DecryptionError)
+        }))
+        .await
+    }
 }
 
 #[async_trait::async_trait]
@@ -98,6 +124,22 @@ impl UserKeyStoreInterface for MockDb {
             .change_context(errors::StorageError::DecryptionError)
     }
 
+    async fn get_all_user_key_store(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> CustomResult<Vec<domain::UserKeyStore>, errors::StorageError> {
+        let user_key_store = self.user_key_store.lock().await;
+
+        futures::future::try_join_all(user_key_store.iter().map(|user_key| async {
+            user_key
+                .to_owned()
+                .convert(key)
+                .await
+                .change_context(errors::StorageError::DecryptionError)
+        }))
+        .await
+    }
+
     #[instrument(skip_all)]
     async fn get_user_key_store_by_user_id(
         &self,
diff --git a/crates/router/src/routes/admin.rs b/crates/router/src/routes/admin.rs
index 3f08834e441..da12c98c77e 100644
--- a/crates/router/src/routes/admin.rs
+++ b/crates/router/src/routes/admin.rs
@@ -446,16 +446,16 @@ pub async fn merchant_account_toggle_kv(
     .await
 }
 
-/// Merchant Account - Toggle KV
+/// Merchant Account - Transfer Keys
 ///
-/// Toggle KV mode for all Merchant Accounts
+/// Transfer Merchant Encryption key to keymanager
 #[instrument(skip_all)]
 pub async fn merchant_account_toggle_all_kv(
     state: web::Data<AppState>,
     req: HttpRequest,
     json_payload: web::Json<admin::ToggleAllKVRequest>,
 ) -> HttpResponse {
-    let flow = Flow::ConfigKeyUpdate;
+    let flow = Flow::MerchantTransferKey;
     let payload = json_payload.into_inner();
 
     api::server_wrap(
@@ -651,6 +651,27 @@ pub async fn merchant_account_kv_status(
     .await
 }
 
+/// Merchant Account - KV Status
+///
+/// Toggle KV mode for the Merchant Account
+#[instrument(skip_all)]
+pub async fn merchant_account_transfer_keys(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+) -> HttpResponse {
+    let flow = Flow::ConfigKeyFetch;
+    api::server_wrap(
+        flow,
+        state,
+        &req,
+        (),
+        |state, _, _, _| transfer_key_store_to_key_manager(state),
+        &auth::AdminApiAuth,
+        api_locking::LockAction::NotApplicable,
+    )
+    .await
+}
+
 #[instrument(skip_all, fields(flow = ?Flow::ToggleExtendedCardInfo))]
 pub async fn toggle_extended_card_info(
     state: web::Data<AppState>,
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 797de75dfcf..d82f98f295d 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -1028,6 +1028,9 @@ impl MerchantAccount {
                     .route(web::post().to(merchant_account_toggle_kv))
                     .route(web::get().to(merchant_account_kv_status)),
             )
+            .service(
+                web::resource("/transfer").route(web::post().to(merchant_account_transfer_keys)),
+            )
             .service(web::resource("/kv").route(web::post().to(merchant_account_toggle_all_kv)))
             .service(
                 web::resource("/{id}")
@@ -1398,6 +1401,11 @@ impl User {
                     .route(web::post().to(set_dashboard_metadata)),
             );
 
+        route = route.service(
+            web::scope("/key")
+                .service(web::resource("/transfer").route(web::post().to(transfer_user_key))),
+        );
+
         // Two factor auth routes
         route = route.service(
             web::scope("/2fa")
diff --git a/crates/router/src/routes/lock_utils.rs b/crates/router/src/routes/lock_utils.rs
index 56e8751a979..8f0c1aa804c 100644
--- a/crates/router/src/routes/lock_utils.rs
+++ b/crates/router/src/routes/lock_utils.rs
@@ -45,6 +45,7 @@ impl From<Flow> for ApiIdentifier {
             | Flow::MerchantsAccountRetrieve
             | Flow::MerchantsAccountUpdate
             | Flow::MerchantsAccountDelete
+            | Flow::MerchantTransferKey
             | Flow::MerchantAccountList => Self::MerchantAccount,
 
             Flow::RoutingCreateConfig
@@ -233,6 +234,7 @@ impl From<Flow> for ApiIdentifier {
             | Flow::CreateUserAuthenticationMethod
             | Flow::UpdateUserAuthenticationMethod
             | Flow::ListUserAuthenticationMethods
+            | Flow::UserTransferKey
             | Flow::GetSsoAuthUrl
             | Flow::SignInWithSso
             | Flow::AuthSelect => Self::User,
diff --git a/crates/router/src/routes/user.rs b/crates/router/src/routes/user.rs
index 7e55393cc57..29935bc8ab7 100644
--- a/crates/router/src/routes/user.rs
+++ b/crates/router/src/routes/user.rs
@@ -895,3 +895,18 @@ pub async fn terminate_auth_select(
     ))
     .await
 }
+
+pub async fn transfer_user_key(state: web::Data<AppState>, req: HttpRequest) -> HttpResponse {
+    let flow = Flow::UserTransferKey;
+
+    Box::pin(api::server_wrap(
+        flow,
+        state.clone(),
+        &req,
+        (),
+        |state, _, _, _| user_core::transfer_user_key_store_keymanager(state),
+        &auth::AdminApiAuth,
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
diff --git a/crates/router/src/types/domain/types.rs b/crates/router/src/types/domain/types.rs
index 176b88b3f5b..151ae61f5fe 100644
--- a/crates/router/src/types/domain/types.rs
+++ b/crates/router/src/types/domain/types.rs
@@ -1,3 +1,18 @@
+use common_utils::types::keymanager::KeyManagerState;
 pub use hyperswitch_domain_models::type_encryption::{
     decrypt, encrypt, encrypt_optional, AsyncLift, Lift, TypeEncryption,
 };
+
+impl From<&crate::SessionState> for KeyManagerState {
+    fn from(state: &crate::SessionState) -> Self {
+        let conf = state.conf.key_manager.get_inner();
+        Self {
+            url: conf.url.clone(),
+            client_idle_timeout: state.conf.proxy.idle_pool_connection_timeout,
+            #[cfg(feature = "keymanager_mtls")]
+            cert: conf.cert.clone(),
+            #[cfg(feature = "keymanager_mtls")]
+            ca: conf.ca.clone(),
+        }
+    }
+}
diff --git a/crates/router/src/types/domain/user.rs b/crates/router/src/types/domain/user.rs
index d5100c89c19..c4c67796d46 100644
--- a/crates/router/src/types/domain/user.rs
+++ b/crates/router/src/types/domain/user.rs
@@ -6,6 +6,8 @@ use api_models::{
 use common_enums::TokenPurpose;
 #[cfg(not(feature = "v2"))]
 use common_utils::id_type;
+#[cfg(feature = "keymanager_create")]
+use common_utils::types::keymanager::{EncryptionCreateRequest, Identifier};
 use common_utils::{crypto::Encryptable, errors::CustomResult, new_type::MerchantName, pii};
 use diesel_models::{
     enums::{TotpStatus, UserStatus},
@@ -971,6 +973,19 @@ impl UserFromStorage {
                     .change_context(UserErrors::InternalServerError)?,
                 created_at: common_utils::date_time::now(),
             };
+
+            #[cfg(feature = "keymanager_create")]
+            {
+                common_utils::keymanager::create_key_in_key_manager(
+                    &state.into(),
+                    EncryptionCreateRequest {
+                        identifier: Identifier::User(key_store.user_id.clone()),
+                    },
+                )
+                .await
+                .change_context(UserErrors::InternalServerError)?;
+            }
+
             state
                 .global_store
                 .insert_user_key_store(key_store, &master_key.to_vec().into())
diff --git a/crates/router_env/src/logger/types.rs b/crates/router_env/src/logger/types.rs
index 2fef21d420c..8ff019f9b5f 100644
--- a/crates/router_env/src/logger/types.rs
+++ b/crates/router_env/src/logger/types.rs
@@ -81,6 +81,8 @@ pub enum Flow {
     MerchantConnectorsDelete,
     /// Merchant Connectors list flow.
     MerchantConnectorsList,
+    /// Merchant Transfer Keys
+    MerchantTransferKey,
     /// ConfigKey create flow.
     ConfigKeyCreate,
     /// ConfigKey fetch flow.
@@ -316,6 +318,8 @@ pub enum Flow {
     UserSignUpWithMerchantId,
     /// User Sign In
     UserSignIn,
+    /// User transfer key
+    UserTransferKey,
     /// User connect account
     UserConnectAccount,
     /// Upsert Decision Manager Config
diff --git a/loadtest/config/development.toml b/loadtest/config/development.toml
index 2f1aa1dbbd8..f38cd05e7c4 100644
--- a/loadtest/config/development.toml
+++ b/loadtest/config/development.toml
@@ -9,6 +9,9 @@ traces_enabled = true
 metrics_enabled = true
 ignore_errors = false
 
+[key_manager]
+url = "http://localhost:5000"
+
 [master_database]
 username = "postgres"
 password = "postgres"
@@ -339,4 +342,4 @@ enabled = false
 global_tenant = { schema = "public", redis_key_prefix = "" }
 
 [multitenancy.tenants]
-public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = "", clickhouse_database = "default"}
\ No newline at end of file
+public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = "", clickhouse_database = "default"}