Bug:  feat: add connector zsl and local bank transfer payment method  to dashboard



### Feature Description

Add ZSL to dashboard connectors and add local bank transfer payment method type to dashboard


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

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

### Have you read the Contributing Guidelines?

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

### Are you willing to submit a PR?

None
diff --git a/config/config.example.toml b/config/config.example.toml
index 56d4500cba2..6f880acbba4 100644
--- a/config/config.example.toml
+++ b/config/config.example.toml
@@ -79,6 +79,7 @@ sampling_rate = 0.1 # decimal rate between 0.0 - 1.0
 [secrets]
 admin_api_key = "test_admin" # admin API key for admin authentication
 jwt_secret = "secret"        # JWT secret used for user authentication
+master_enc_key = "sample_key" # Master Encryption key used to encrypt merchant wise encryption key. Should be 32-byte long.
 
 # Locker settings contain details for accessing a card locker, a
 # PCI Compliant storage entity which stores payment method information
diff --git a/crates/router/src/configs/defaults.rs b/crates/router/src/configs/defaults.rs
index 4b48d4292e0..50034a87604 100644
--- a/crates/router/src/configs/defaults.rs
+++ b/crates/router/src/configs/defaults.rs
@@ -33,6 +33,8 @@ impl Default for super::settings::Secrets {
         Self {
             jwt_secret: "secret".into(),
             admin_api_key: "test_admin".into(),
+            master_enc_key: "73ad7bbbbc640c845a150f67d058b279849370cd2c1f3c67c4dd6c869213e13a"
+                .into(),
         }
     }
 }
diff --git a/crates/router/src/configs/settings.rs b/crates/router/src/configs/settings.rs
index a8719e221a3..2a3bcc7355e 100644
--- a/crates/router/src/configs/settings.rs
+++ b/crates/router/src/configs/settings.rs
@@ -158,6 +158,7 @@ where
 pub struct Secrets {
     pub jwt_secret: String,
     pub admin_api_key: String,
+    pub master_enc_key: String,
 }
 
 #[derive(Debug, Deserialize, Clone)]
diff --git a/crates/router/src/core/admin.rs b/crates/router/src/core/admin.rs
index 0d6e4273e83..fa42f9dee28 100644
--- a/crates/router/src/core/admin.rs
+++ b/crates/router/src/core/admin.rs
@@ -1,5 +1,6 @@
 use common_utils::{
-    crypto::{self, GcmAes256, OptionalSecretValue},
+    crypto::OptionalSecretValue,
+    date_time,
     ext_traits::{AsyncExt, ValueExt},
 };
 use error_stack::{report, FutureExt, IntoReport, ResultExt};
@@ -15,12 +16,12 @@ use crate::{
     },
     db::StorageInterface,
     routes::AppState,
-    services::api as service_api,
+    services::{self, api as service_api},
     types::{
         self, api,
         domain::{
-            self, merchant_account as merchant_domain,
-            types::{get_key_and_algo, AsyncLift, TypeEncryption},
+            self, merchant_account as merchant_domain, merchant_key_store,
+            types::{self as domain_types, AsyncLift},
         },
         storage,
         transformers::ForeignInto,
@@ -42,6 +43,12 @@ pub async fn create_merchant_account(
     req: api::MerchantAccountCreate,
 ) -> RouterResponse<api::MerchantAccountResponse> {
     let db = &*state.store;
+    let master_key = db.get_master_key();
+
+    let key = services::generate_aes256_key()
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Unable to generate aes 256 key")?;
+
     let publishable_key = Some(create_merchant_publishable_key());
 
     let api_key_request = api::CreateApiKeyRequest {
@@ -53,10 +60,6 @@ pub async fn create_merchant_account(
         expiration: api::ApiKeyExpiration::Never,
     };
 
-    let key = get_key_and_algo(db, req.merchant_id.clone())
-        .await
-        .change_context(errors::ApiErrorResponse::InternalServerError)?;
-
     let api_key = match api_keys::create_api_key(
         db,
         &state.conf.api_keys,
@@ -100,16 +103,31 @@ pub async fn create_merchant_account(
             .attach_printable("Invalid routing algorithm given")?;
     }
 
+    let key_store = merchant_key_store::MerchantKeyStore {
+        merchant_id: req.merchant_id.clone(),
+        key: domain_types::encrypt(key.to_vec().into(), master_key)
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed to decrypt data from key store")?,
+        created_at: date_time::now(),
+    };
+
+    db.insert_merchant_key_store(key_store)
+        .await
+        .map_err(|error| {
+            error.to_duplicate_response(errors::ApiErrorResponse::DuplicateMerchantAccount)
+        })?;
+
     let encrypt_string = |inner: Option<masking::Secret<String>>| async {
         inner
-            .async_map(|value| crypto::Encryptable::encrypt(value, &key, GcmAes256 {}))
+            .async_map(|value| domain_types::encrypt(value, &key))
             .await
             .transpose()
     };
 
     let encrypt_value = |inner: OptionalSecretValue| async {
         inner
-            .async_map(|value| crypto::Encryptable::encrypt(value, &key, GcmAes256 {}))
+            .async_map(|value| domain_types::encrypt(value, &key))
             .await
             .transpose()
     };
@@ -121,10 +139,7 @@ pub async fn create_merchant_account(
         Ok(merchant_domain::MerchantAccount {
             merchant_id: req.merchant_id,
             merchant_name: req.merchant_name.async_lift(encrypt_string).await?,
-            api_key: Some(
-                crypto::Encryptable::encrypt(api_key.peek().clone().into(), &key, GcmAes256 {})
-                    .await?,
-            ),
+            api_key: Some(domain_types::encrypt(api_key.peek().clone().into(), &key).await?),
             merchant_details: merchant_details.async_lift(encrypt_value).await?,
             return_url: req.return_url.map(|a| a.to_string()),
             webhook_details,
@@ -152,7 +167,6 @@ pub async fn create_merchant_account(
         .map_err(|error| {
             error.to_duplicate_response(errors::ApiErrorResponse::DuplicateMerchantAccount)
         })?;
-
     Ok(service_api::ApplicationResponse::Json(
         merchant_account.into(),
     ))
@@ -179,9 +193,10 @@ pub async fn merchant_account_update(
     merchant_id: &String,
     req: api::MerchantAccountUpdate,
 ) -> RouterResponse<api::MerchantAccountResponse> {
-    let key = get_key_and_algo(db, merchant_id.clone())
+    let key = domain_types::get_merchant_enc_key(db, merchant_id.clone())
         .await
-        .change_context(errors::ApiErrorResponse::InternalServerError)?;
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Unable to get data from merchant key store")?;
 
     if &req.merchant_id != merchant_id {
         Err(report!(errors::ValidationError::IncorrectValueProvided {
@@ -205,24 +220,41 @@ pub async fn merchant_account_update(
             .attach_printable("Invalid routing algorithm given")?;
     }
 
+    let encrypt_string = |inner: Option<masking::Secret<String>>| async {
+        inner
+            .async_map(|value| domain_types::encrypt(value, &key))
+            .await
+            .transpose()
+    };
+
+    let encrypt_value = |inner: OptionalSecretValue| async {
+        inner
+            .async_map(|value| domain_types::encrypt(value, &key))
+            .await
+            .transpose()
+    };
+
     let updated_merchant_account = storage::MerchantAccountUpdate::Update {
         merchant_name: req
             .merchant_name
-            .async_map(|inner| crypto::Encryptable::encrypt(inner.into(), &key, GcmAes256 {}))
+            .map(masking::Secret::new)
+            .async_lift(encrypt_string)
             .await
-            .transpose()
-            .change_context(errors::ApiErrorResponse::InternalServerError)?,
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Unable to encrypt merchant name")?,
 
         merchant_details: req
             .merchant_details
             .as_ref()
             .map(utils::Encode::<api::MerchantDetails>::encode_to_value)
             .transpose()
-            .change_context(errors::ApiErrorResponse::InternalServerError)?
-            .async_map(|inner| crypto::Encryptable::encrypt(inner.into(), &key, GcmAes256 {}))
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Unable to convert merchant_details to a value")?
+            .map(masking::Secret::new)
+            .async_lift(encrypt_value)
             .await
-            .transpose()
-            .change_context(errors::ApiErrorResponse::InternalServerError)?,
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Unable to encrypt merchant details")?,
 
         return_url: req.return_url.map(|a| a.to_string()),
 
@@ -323,6 +355,11 @@ pub async fn create_payment_connector(
     req: api::MerchantConnector,
     merchant_id: &String,
 ) -> RouterResponse<api::MerchantConnector> {
+    let key = domain_types::get_merchant_enc_key(store, merchant_id.clone())
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Unable to get key from merchant key store")?;
+
     let _merchant_account = store
         .find_merchant_account_by_merchant_id(merchant_id)
         .await
@@ -362,11 +399,17 @@ pub async fn create_payment_connector(
         connector_type: req.connector_type.foreign_into(),
         connector_name: req.connector_name,
         merchant_connector_id: utils::generate_id(consts::ID_LENGTH, "mca"),
-        connector_account_details: req.connector_account_details.ok_or(
-            errors::ApiErrorResponse::MissingRequiredField {
-                field_name: "connector_account_details",
-            },
-        )?,
+        connector_account_details: domain_types::encrypt(
+            req.connector_account_details.ok_or(
+                errors::ApiErrorResponse::MissingRequiredField {
+                    field_name: "connector_account_details",
+                },
+            )?,
+            &key,
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Unable to encrypt connector account details")?,
         payment_methods_enabled,
         test_mode: req.test_mode,
         disabled: req.disabled,
@@ -444,6 +487,11 @@ pub async fn update_payment_connector(
     merchant_connector_id: &str,
     req: api::MerchantConnector,
 ) -> RouterResponse<api::MerchantConnector> {
+    let key = domain_types::get_merchant_enc_key(db, merchant_id)
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Unable to get key from merchant key store")?;
+
     let _merchant_account = db
         .find_merchant_account_by_merchant_id(merchant_id)
         .await
@@ -470,12 +518,24 @@ pub async fn update_payment_connector(
             .collect::<Vec<serde_json::Value>>()
     });
 
+    let encrypt = |inner: Option<masking::Secret<serde_json::Value>>| async {
+        inner
+            .async_map(|inner| domain_types::encrypt(inner, &key))
+            .await
+            .transpose()
+    };
+
     let payment_connector = storage::MerchantConnectorAccountUpdate::Update {
         merchant_id: Some(merchant_id.to_string()),
         connector_type: Some(req.connector_type.foreign_into()),
         connector_name: Some(req.connector_name),
         merchant_connector_id: Some(merchant_connector_id.to_string()),
-        connector_account_details: req.connector_account_details,
+        connector_account_details: req
+            .connector_account_details
+            .async_lift(encrypt)
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Unable to encrypt connector_account_details")?,
         payment_methods_enabled,
         test_mode: req.test_mode,
         disabled: req.disabled,
@@ -483,7 +543,7 @@ pub async fn update_payment_connector(
     };
 
     let updated_mca = db
-        .update_merchant_connector_account(mca, payment_connector)
+        .update_merchant_connector_account(mca, payment_connector.into())
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable_lazy(|| {
@@ -505,7 +565,7 @@ pub async fn update_payment_connector(
         connector_type: updated_mca.connector_type.foreign_into(),
         connector_name: updated_mca.connector_name,
         merchant_connector_id: Some(updated_mca.merchant_connector_id),
-        connector_account_details: Some(updated_mca.connector_account_details),
+        connector_account_details: Some(updated_mca.connector_account_details.into_inner()),
         test_mode: updated_mca.test_mode,
         disabled: updated_mca.disabled,
         payment_methods_enabled: updated_pm_enabled,
diff --git a/crates/router/src/core/customers.rs b/crates/router/src/core/customers.rs
index 3ae20a34cf3..6be36d971dd 100644
--- a/crates/router/src/core/customers.rs
+++ b/crates/router/src/core/customers.rs
@@ -1,5 +1,5 @@
 use common_utils::{
-    crypto::{self, Encryptable, GcmAes256},
+    crypto::{Encryptable, GcmAes256},
     ext_traits::{AsyncExt, ValueExt},
 };
 use error_stack::ResultExt;
@@ -20,7 +20,7 @@ use crate::{
         api::customers::{self, CustomerRequestExt},
         domain::{
             self, customer, merchant_account,
-            types::{get_key_and_algo, AsyncLift, TypeEncryption},
+            types::{self, AsyncLift, TypeEncryption},
         },
         storage::{self, enums},
     },
@@ -40,21 +40,21 @@ pub async fn create_customer(
     let merchant_id = &merchant_account.merchant_id;
     customer_data.merchant_id = merchant_id.to_owned();
 
-    let key = get_key_and_algo(db, merchant_id.clone())
+    let key = types::get_merchant_enc_key(db, merchant_id.clone())
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Failed while getting encryption key")?;
 
     let encrypt = |inner: Option<masking::Secret<String>>| async {
         inner
-            .async_map(|value| crypto::Encryptable::encrypt(value, &key, GcmAes256 {}))
+            .async_map(|value| types::encrypt(value, &key))
             .await
             .transpose()
     };
 
     let encrypt_email = |inner: Option<masking::Secret<String, pii::Email>>| async {
         inner
-            .async_map(|value| crypto::Encryptable::encrypt(value, &key, GcmAes256 {}))
+            .async_map(|value| types::encrypt(value, &key))
             .await
             .transpose()
     };
@@ -214,7 +214,7 @@ pub async fn delete_customer(
         }?,
     };
 
-    let key = get_key_and_algo(&**db, merchant_account.merchant_id.clone())
+    let key = types::get_merchant_enc_key(&**db, merchant_account.merchant_id.clone())
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Failed while getting key for encryption")?;
@@ -300,21 +300,21 @@ pub async fn update_customer(
     .await
     .map_err(|err| err.to_not_found_response(errors::ApiErrorResponse::CustomerNotFound))?;
 
-    let key = get_key_and_algo(db, merchant_account.merchant_id.clone())
+    let key = types::get_merchant_enc_key(db, merchant_account.merchant_id.clone())
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Failed while getting key for encryption")?;
 
     let encrypt = |inner: Option<masking::Secret<String>>| async {
         inner
-            .async_map(|value| crypto::Encryptable::encrypt(value, &key, GcmAes256 {}))
+            .async_map(|value| types::encrypt(value, &key))
             .await
             .transpose()
     };
 
     let encrypt_email = |inner: Option<masking::Secret<String, pii::Email>>| async {
         inner
-            .async_map(|value| crypto::Encryptable::encrypt(value, &key, GcmAes256 {}))
+            .async_map(|value| types::encrypt(value, &key))
             .await
             .transpose()
     };
diff --git a/crates/router/src/core/errors.rs b/crates/router/src/core/errors.rs
index bd918b899a8..585afd21486 100644
--- a/crates/router/src/core/errors.rs
+++ b/crates/router/src/core/errors.rs
@@ -71,6 +71,10 @@ pub enum StorageError {
     CustomerRedacted,
     #[error("Deserialization failure")]
     DeserializationFailed,
+    #[error("Error while encrypting data")]
+    EncryptionError,
+    #[error("Error while decrypting data from database")]
+    DecryptionError,
     #[error("RedisError: {0:?}")]
     RedisError(error_stack::Report<RedisError>),
 }
diff --git a/crates/router/src/core/payment_methods/cards.rs b/crates/router/src/core/payment_methods/cards.rs
index 79f4dd46ab0..9bc55fd7bc1 100644
--- a/crates/router/src/core/payment_methods/cards.rs
+++ b/crates/router/src/core/payment_methods/cards.rs
@@ -29,7 +29,10 @@ use crate::{
     },
     db, logger,
     pii::prelude::*,
-    routes::{self, metrics},
+    routes::{
+        self,
+        metrics::{self, request},
+    },
     services,
     types::{
         api::{self, PaymentMethodCreateExt},
@@ -162,7 +165,7 @@ pub async fn add_card_to_locker(
     merchant_account: &merchant_account::MerchantAccount,
 ) -> errors::CustomResult<api::PaymentMethodResponse, errors::VaultError> {
     metrics::STORED_TO_LOCKER.add(&metrics::CONTEXT, 1, &[]);
-    metrics::request::record_card_operation_time(
+    request::record_operation_time(
         async {
             match state.conf.locker.locker_setup {
                 settings::LockerSetup::BasiliskLocker => {
@@ -191,7 +194,7 @@ pub async fn get_card_from_locker(
 ) -> errors::RouterResult<payment_methods::Card> {
     metrics::GET_FROM_LOCKER.add(&metrics::CONTEXT, 1, &[]);
 
-    metrics::request::record_card_operation_time(
+    request::record_operation_time(
         async {
             match state.conf.locker.locker_setup {
                 settings::LockerSetup::LegacyLocker => {
@@ -227,7 +230,7 @@ pub async fn delete_card_from_locker(
 ) -> errors::RouterResult<payment_methods::DeleteCardResp> {
     metrics::DELETE_FROM_LOCKER.add(&metrics::CONTEXT, 1, &[]);
 
-    metrics::request::record_card_operation_time(
+    request::record_operation_time(
         async {
             match state.conf.locker.locker_setup {
                 settings::LockerSetup::LegacyLocker => {
diff --git a/crates/router/src/core/payments/helpers.rs b/crates/router/src/core/payments/helpers.rs
index 658dcb182b2..29ae6330a83 100644
--- a/crates/router/src/core/payments/helpers.rs
+++ b/crates/router/src/core/payments/helpers.rs
@@ -32,7 +32,7 @@ use crate::{
         api::{self, admin, enums as api_enums, CustomerAcceptanceExt, MandateValidationFieldsExt},
         domain::{
             self, customer, merchant_account,
-            types::{get_key_and_algo, AsyncLift, TypeEncryption},
+            types::{self, AsyncLift},
         },
         storage::{self, enums as storage_enums, ephemeral_key},
         transformers::ForeignInto,
@@ -51,14 +51,14 @@ pub async fn get_address_for_payment_request(
     merchant_id: &str,
     customer_id: &Option<String>,
 ) -> CustomResult<Option<domain::address::Address>, errors::ApiErrorResponse> {
-    let key = get_key_and_algo(db, merchant_id.to_string())
+    let key = types::get_merchant_enc_key(db, merchant_id.to_string())
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Failed while getting key for encryption")?;
 
     let encrypt = |inner: Option<masking::Secret<String>>| async {
         inner
-            .async_map(|value| crypto::Encryptable::encrypt(value, &key, crypto::GcmAes256 {}))
+            .async_map(|value| types::encrypt(value, &key))
             .await
             .transpose()
     };
@@ -783,22 +783,18 @@ pub async fn create_customer_if_not_exist<'a, F: Clone, R>(
             Some(match customer_data {
                 Some(c) => Ok(c),
                 None => {
-                    let key = get_key_and_algo(db, merchant_id.to_string()).await?;
+                    let key = types::get_merchant_enc_key(db, merchant_id.to_string()).await?;
 
                     let encrypt = |inner: Option<masking::Secret<String>>| async {
                         inner
-                            .async_map(|value| {
-                                crypto::Encryptable::encrypt(value, &key, crypto::GcmAes256 {})
-                            })
+                            .async_map(|value| types::encrypt(value, &key))
                             .await
                             .transpose()
                     };
 
                     let encrypt_email = |inner: Option<masking::Secret<String, pii::Email>>| async {
                         inner
-                            .async_map(|value| {
-                                crypto::Encryptable::encrypt(value, &key, crypto::GcmAes256 {})
-                            })
+                            .async_map(|value| types::encrypt(value, &key))
                             .await
                             .transpose()
                     };
diff --git a/crates/router/src/db.rs b/crates/router/src/db.rs
index 9e8c08f1d2e..262b9b31f04 100644
--- a/crates/router/src/db.rs
+++ b/crates/router/src/db.rs
@@ -12,6 +12,7 @@ pub mod locker_mock_up;
 pub mod mandate;
 pub mod merchant_account;
 pub mod merchant_connector_account;
+pub mod merchant_key_store;
 pub mod payment_attempt;
 pub mod payment_intent;
 pub mod payment_method;
@@ -59,11 +60,33 @@ pub trait StorageInterface:
     + refund::RefundInterface
     + reverse_lookup::ReverseLookupInterface
     + cards_info::CardsInfoInterface
+    + merchant_key_store::MerchantKeyStoreInterface
+    + MasterKeyInterface
     + 'static
 {
     async fn close(&mut self) {}
 }
 
+pub trait MasterKeyInterface {
+    fn get_master_key(&self) -> &[u8];
+}
+
+impl MasterKeyInterface for Store {
+    fn get_master_key(&self) -> &[u8] {
+        &self.master_key
+    }
+}
+
+/// Default dummy key for MockDb
+impl MasterKeyInterface for MockDb {
+    fn get_master_key(&self) -> &[u8] {
+        &[
+            1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24,
+            25, 26, 27, 28, 29, 30, 31, 32,
+        ]
+    }
+}
+
 #[async_trait::async_trait]
 impl StorageInterface for Store {
     #[allow(clippy::expect_used)]
diff --git a/crates/router/src/db/address.rs b/crates/router/src/db/address.rs
index 29da891d052..eb6ee3d77c9 100644
--- a/crates/router/src/db/address.rs
+++ b/crates/router/src/db/address.rs
@@ -56,10 +56,11 @@ impl AddressInterface for Store {
             .map_err(Into::into)
             .into_report()
             .async_and_then(|address| async {
+                let merchant_id = address.merchant_id.clone();
                 address
-                    .convert()
+                    .convert(self, &merchant_id)
                     .await
-                    .change_context(errors::StorageError::DeserializationFailed)
+                    .change_context(errors::StorageError::DecryptionError)
             })
             .await
     }
@@ -75,10 +76,11 @@ impl AddressInterface for Store {
             .map_err(Into::into)
             .into_report()
             .async_and_then(|address| async {
+                let merchant_id = address.merchant_id.clone();
                 address
-                    .convert()
+                    .convert(self, &merchant_id)
                     .await
-                    .change_context(errors::StorageError::DeserializationFailed)
+                    .change_context(errors::StorageError::DecryptionError)
             })
             .await
     }
@@ -91,16 +93,17 @@ impl AddressInterface for Store {
         address
             .construct_new()
             .await
-            .change_context(errors::StorageError::DeserializationFailed)?
+            .change_context(errors::StorageError::EncryptionError)?
             .insert(&conn)
             .await
             .map_err(Into::into)
             .into_report()
             .async_and_then(|address| async {
+                let merchant_id = address.merchant_id.clone();
                 address
-                    .convert()
+                    .convert(self, &merchant_id)
                     .await
-                    .change_context(errors::StorageError::DeserializationFailed)
+                    .change_context(errors::StorageError::DecryptionError)
             })
             .await
     }
@@ -124,11 +127,12 @@ impl AddressInterface for Store {
         .async_and_then(|addresses| async {
             let mut output = Vec::with_capacity(addresses.len());
             for address in addresses.into_iter() {
+                let merchant_id = address.merchant_id.clone();
                 output.push(
                     address
-                        .convert()
+                        .convert(self, &merchant_id)
                         .await
-                        .change_context(errors::StorageError::DeserializationFailed)?,
+                        .change_context(errors::StorageError::DecryptionError)?,
                 )
             }
             Ok(output)
diff --git a/crates/router/src/db/customers.rs b/crates/router/src/db/customers.rs
index 9c54e34ba35..77e7f8fb099 100644
--- a/crates/router/src/db/customers.rs
+++ b/crates/router/src/db/customers.rs
@@ -72,9 +72,9 @@ impl CustomerInterface for Store {
             .map_err(Into::into)
             .into_report()?
             .async_map(|c| async {
-                c.convert()
+                c.convert(self, merchant_id)
                     .await
-                    .change_context(errors::StorageError::DeserializationFailed)
+                    .change_context(errors::StorageError::DecryptionError)
             })
             .await
             .transpose()?;
@@ -107,9 +107,10 @@ impl CustomerInterface for Store {
         .map_err(Into::into)
         .into_report()
         .async_and_then(|c| async {
-            c.convert()
+            let merchant_id = c.merchant_id.clone();
+            c.convert(self, &merchant_id)
                 .await
-                .change_context(errors::StorageError::DeserializationFailed)
+                .change_context(errors::StorageError::DecryptionError)
         })
         .await
     }
@@ -126,9 +127,10 @@ impl CustomerInterface for Store {
                 .map_err(Into::into)
                 .into_report()
                 .async_and_then(|c| async {
-                    c.convert()
+                    let merchant_id = c.merchant_id.clone();
+                    c.convert(self, &merchant_id)
                         .await
-                        .change_context(errors::StorageError::DeserializationFailed)
+                        .change_context(errors::StorageError::DecryptionError)
                 })
                 .await?;
         match customer.name {
@@ -147,15 +149,16 @@ impl CustomerInterface for Store {
         customer_data
             .construct_new()
             .await
-            .change_context(errors::StorageError::DeserializationFailed)?
+            .change_context(errors::StorageError::EncryptionError)?
             .insert(&conn)
             .await
             .map_err(Into::into)
             .into_report()
             .async_and_then(|c| async {
-                c.convert()
+                let merchant_id = c.merchant_id.clone();
+                c.convert(self, &merchant_id)
                     .await
-                    .change_context(errors::StorageError::DeserializationFailed)
+                    .change_context(errors::StorageError::DecryptionError)
             })
             .await
     }
@@ -190,9 +193,10 @@ impl CustomerInterface for MockDb {
             .cloned();
         customer
             .async_map(|c| async {
-                c.convert()
+                let merchant_id = c.merchant_id.clone();
+                c.convert(self, &merchant_id)
                     .await
-                    .change_context(errors::StorageError::DeserializationFailed)
+                    .change_context(errors::StorageError::DecryptionError)
             })
             .await
             .transpose()
@@ -226,13 +230,15 @@ impl CustomerInterface for MockDb {
 
         let customer = Conversion::convert(customer_data)
             .await
-            .change_context(errors::StorageError::SerializationFailed)?;
+            .change_context(errors::StorageError::EncryptionError)?;
+
+        let merchant_id = customer.merchant_id.clone();
         customers.push(customer.clone());
 
         customer
-            .convert()
+            .convert(self, &merchant_id)
             .await
-            .change_context(errors::StorageError::DeserializationFailed)
+            .change_context(errors::StorageError::DecryptionError)
     }
 
     async fn delete_customer_by_customer_id_merchant_id(
diff --git a/crates/router/src/db/merchant_account.rs b/crates/router/src/db/merchant_account.rs
index bb79863a811..b88db0d1947 100644
--- a/crates/router/src/db/merchant_account.rs
+++ b/crates/router/src/db/merchant_account.rs
@@ -1,6 +1,8 @@
 use common_utils::ext_traits::AsyncExt;
 use error_stack::{IntoReport, ResultExt};
 
+#[cfg(feature = "accounts_cache")]
+use super::cache;
 use super::{MockDb, Store};
 use crate::{
     connection,
@@ -10,7 +12,7 @@ use crate::{
             behaviour::{Conversion, ReverseConversion},
             merchant_account,
         },
-        storage::{self},
+        storage,
     },
 };
 
@@ -60,17 +62,18 @@ impl MerchantAccountInterface for Store {
         merchant_account: merchant_account::MerchantAccount,
     ) -> CustomResult<merchant_account::MerchantAccount, errors::StorageError> {
         let conn = connection::pg_connection_write(self).await?;
+        let merchant_id = merchant_account.merchant_id.clone();
         merchant_account
             .construct_new()
             .await
-            .change_context(errors::StorageError::DeserializationFailed)?
+            .change_context(errors::StorageError::EncryptionError)?
             .insert(&conn)
             .await
             .map_err(Into::into)
             .into_report()?
-            .convert()
+            .convert(self, &merchant_id)
             .await
-            .change_context(errors::StorageError::DeserializationFailed)
+            .change_context(errors::StorageError::DecryptionError)
     }
 
     async fn find_merchant_account_by_merchant_id(
@@ -89,18 +92,18 @@ impl MerchantAccountInterface for Store {
         {
             fetch_func()
                 .await?
-                .convert()
+                .convert(self, merchant_id)
                 .await
-                .change_context(errors::StorageError::DeserializationFailed)
+                .change_context(errors::StorageError::DecryptionError)
         }
 
         #[cfg(feature = "accounts_cache")]
         {
-            super::cache::get_or_populate_redis(self, merchant_id, fetch_func)
+            cache::get_or_populate_redis(self, merchant_id, fetch_func)
                 .await?
-                .convert()
+                .convert(self, merchant_id)
                 .await
-                .change_context(errors::StorageError::DeserializationFailed)
+                .change_context(errors::StorageError::DecryptionError)
         }
     }
 
@@ -114,15 +117,15 @@ impl MerchantAccountInterface for Store {
             let conn = connection::pg_connection_write(self).await?;
             Conversion::convert(this)
                 .await
-                .change_context(errors::StorageError::DeserializationFailed)?
+                .change_context(errors::StorageError::EncryptionError)?
                 .update(&conn, merchant_account.into())
                 .await
                 .map_err(Into::into)
                 .into_report()
                 .async_and_then(|item| async {
-                    item.convert()
+                    item.convert(self, &_merchant_id)
                         .await
-                        .change_context(errors::StorageError::DeserializationFailed)
+                        .change_context(errors::StorageError::DecryptionError)
                 })
                 .await
         };
@@ -134,7 +137,7 @@ impl MerchantAccountInterface for Store {
 
         #[cfg(feature = "accounts_cache")]
         {
-            super::cache::redact_cache(self, &_merchant_id, update_func, None).await
+            cache::redact_cache(self, &_merchant_id, update_func, None).await
         }
     }
 
@@ -154,9 +157,9 @@ impl MerchantAccountInterface for Store {
             .map_err(Into::into)
             .into_report()
             .async_and_then(|item| async {
-                item.convert()
+                item.convert(self, merchant_id)
                     .await
-                    .change_context(errors::StorageError::DeserializationFailed)
+                    .change_context(errors::StorageError::DecryptionError)
             })
             .await
         };
@@ -168,7 +171,7 @@ impl MerchantAccountInterface for Store {
 
         #[cfg(feature = "accounts_cache")]
         {
-            super::cache::redact_cache(self, merchant_id, update_func, None).await
+            cache::redact_cache(self, merchant_id, update_func, None).await
         }
     }
 
@@ -182,9 +185,10 @@ impl MerchantAccountInterface for Store {
             .map_err(Into::into)
             .into_report()
             .async_and_then(|item| async {
-                item.convert()
+                let merchant_id = item.merchant_id.clone();
+                item.convert(self, &merchant_id)
                     .await
-                    .change_context(errors::StorageError::DeserializationFailed)
+                    .change_context(errors::StorageError::DecryptionError)
             })
             .await
     }
@@ -208,7 +212,7 @@ impl MerchantAccountInterface for Store {
 
         #[cfg(feature = "accounts_cache")]
         {
-            super::cache::redact_cache(self, merchant_id, delete_func, None).await
+            cache::redact_cache(self, merchant_id, delete_func, None).await
         }
     }
 }
@@ -223,13 +227,14 @@ impl MerchantAccountInterface for MockDb {
         let mut accounts = self.merchant_accounts.lock().await;
         let account = Conversion::convert(merchant_account)
             .await
-            .change_context(errors::StorageError::SerializationFailed)?;
+            .change_context(errors::StorageError::EncryptionError)?;
+        let merchant_id = account.merchant_id.clone();
         accounts.push(account.clone());
 
         account
-            .convert()
+            .convert(self, &merchant_id)
             .await
-            .change_context(errors::StorageError::SerializationFailed)
+            .change_context(errors::StorageError::DecryptionError)
     }
 
     #[allow(clippy::panic)]
@@ -243,9 +248,9 @@ impl MerchantAccountInterface for MockDb {
             .find(|account| account.merchant_id == merchant_id)
             .cloned()
             .async_map(|a| async {
-                a.convert()
+                a.convert(self, merchant_id)
                     .await
-                    .change_context(errors::StorageError::DeserializationFailed)
+                    .change_context(errors::StorageError::DecryptionError)
             })
             .await
             .transpose()?;
diff --git a/crates/router/src/db/merchant_connector_account.rs b/crates/router/src/db/merchant_connector_account.rs
index d84908193a7..33e5592350d 100644
--- a/crates/router/src/db/merchant_connector_account.rs
+++ b/crates/router/src/db/merchant_connector_account.rs
@@ -1,7 +1,8 @@
 use common_utils::ext_traits::{AsyncExt, ByteSliceExt, Encode};
 use error_stack::{IntoReport, ResultExt};
-use masking::ExposeInterface;
 
+#[cfg(feature = "accounts_cache")]
+use super::cache;
 use super::{MockDb, Store};
 use crate::{
     connection,
@@ -149,7 +150,7 @@ where
     async fn update_merchant_connector_account(
         &self,
         this: domain::merchant_connector_account::MerchantConnectorAccount,
-        merchant_connector_account: storage::MerchantConnectorAccountUpdate,
+        merchant_connector_account: storage::MerchantConnectorAccountUpdateInternal,
     ) -> CustomResult<
         domain::merchant_connector_account::MerchantConnectorAccount,
         errors::StorageError,
@@ -182,9 +183,9 @@ impl MerchantConnectorAccountInterface for Store {
         .map_err(Into::into)
         .into_report()
         .async_and_then(|item| async {
-            item.convert()
+            item.convert(self, merchant_id)
                 .await
-                .change_context(errors::StorageError::DeserializationFailed)
+                .change_context(errors::StorageError::DecryptionError)
         })
         .await
     }
@@ -207,21 +208,24 @@ impl MerchantConnectorAccountInterface for Store {
             .await
             .map_err(Into::into)
             .into_report()
-            .async_and_then(|item| async {
-                item.convert()
-                    .await
-                    .change_context(errors::StorageError::DeserializationFailed)
-            })
-            .await
         };
+
         #[cfg(not(feature = "accounts_cache"))]
         {
-            find_call().await
+            find_call()
+                .await?
+                .convert(self, merchant_id)
+                .await
+                .change_context(errors::StorageError::DeserializationFailed)
         }
 
         #[cfg(feature = "accounts_cache")]
         {
-            super::cache::get_or_populate_redis(self, merchant_connector_id, find_call).await
+            cache::get_or_populate_redis(self, merchant_connector_id, find_call)
+                .await?
+                .convert(self, merchant_id)
+                .await
+                .change_context(errors::StorageError::DeserializationFailed)
         }
     }
 
@@ -235,15 +239,16 @@ impl MerchantConnectorAccountInterface for Store {
         let conn = connection::pg_connection_write(self).await?;
         t.construct_new()
             .await
-            .change_context(errors::StorageError::DeserializationFailed)?
+            .change_context(errors::StorageError::EncryptionError)?
             .insert(&conn)
             .await
             .map_err(Into::into)
             .into_report()
             .async_and_then(|item| async {
-                item.convert()
+                let merchant_id = item.merchant_id.clone();
+                item.convert(self, &merchant_id)
                     .await
-                    .change_context(errors::StorageError::DeserializationFailed)
+                    .change_context(errors::StorageError::DecryptionError)
             })
             .await
     }
@@ -265,9 +270,9 @@ impl MerchantConnectorAccountInterface for Store {
                 let mut output = Vec::with_capacity(items.len());
                 for item in items.into_iter() {
                     output.push(
-                        item.convert()
+                        item.convert(self, merchant_id)
                             .await
-                            .change_context(errors::StorageError::DeserializationFailed)?,
+                            .change_context(errors::StorageError::DecryptionError)?,
                     )
                 }
                 Ok(output)
@@ -278,7 +283,7 @@ impl MerchantConnectorAccountInterface for Store {
     async fn update_merchant_connector_account(
         &self,
         this: domain::merchant_connector_account::MerchantConnectorAccount,
-        merchant_connector_account: storage::MerchantConnectorAccountUpdate,
+        merchant_connector_account: storage::MerchantConnectorAccountUpdateInternal,
     ) -> CustomResult<
         domain::merchant_connector_account::MerchantConnectorAccount,
         errors::StorageError,
@@ -288,22 +293,23 @@ impl MerchantConnectorAccountInterface for Store {
             let conn = connection::pg_connection_write(self).await?;
             Conversion::convert(this)
                 .await
-                .change_context(errors::StorageError::DeserializationFailed)?
+                .change_context(errors::StorageError::EncryptionError)?
                 .update(&conn, merchant_connector_account)
                 .await
                 .map_err(Into::into)
                 .into_report()
                 .async_and_then(|item| async {
-                    item.convert()
+                    let merchant_id = item.merchant_id.clone();
+                    item.convert(self, &merchant_id)
                         .await
-                        .change_context(errors::StorageError::DeserializationFailed)
+                        .change_context(errors::StorageError::DecryptionError)
                 })
                 .await
         };
 
         #[cfg(feature = "accounts_cache")]
         {
-            super::cache::redact_cache(self, &_merchant_connector_id, update_call, None).await
+            cache::redact_cache(self, &_merchant_connector_id, update_call, None).await
         }
 
         #[cfg(not(feature = "accounts_cache"))]
@@ -350,9 +356,9 @@ impl MerchantConnectorAccountInterface for MockDb {
             .cloned()
             .unwrap();
         account
-            .convert()
+            .convert(self, merchant_id)
             .await
-            .change_context(errors::StorageError::DeserializationFailed)
+            .change_context(errors::StorageError::DecryptionError)
     }
 
     async fn find_by_merchant_connector_account_merchant_id_merchant_connector_id(
@@ -376,12 +382,13 @@ impl MerchantConnectorAccountInterface for MockDb {
         errors::StorageError,
     > {
         let mut accounts = self.merchant_connector_accounts.lock().await;
+        let merchant_id = t.merchant_id.clone();
         let account = storage::MerchantConnectorAccount {
             #[allow(clippy::as_conversions)]
             id: accounts.len() as i32,
             merchant_id: t.merchant_id,
             connector_name: t.connector_name,
-            connector_account_details: t.connector_account_details.expose(),
+            connector_account_details: t.connector_account_details.into(),
             test_mode: t.test_mode,
             disabled: t.disabled,
             merchant_connector_id: t.merchant_connector_id,
@@ -391,9 +398,9 @@ impl MerchantConnectorAccountInterface for MockDb {
         };
         accounts.push(account.clone());
         account
-            .convert()
+            .convert(self, &merchant_id)
             .await
-            .change_context(errors::StorageError::DeserializationFailed)
+            .change_context(errors::StorageError::DecryptionError)
     }
 
     async fn find_merchant_connector_account_by_merchant_id_and_disabled_list(
@@ -411,7 +418,7 @@ impl MerchantConnectorAccountInterface for MockDb {
     async fn update_merchant_connector_account(
         &self,
         _this: domain::merchant_connector_account::MerchantConnectorAccount,
-        _merchant_connector_account: storage::MerchantConnectorAccountUpdate,
+        _merchant_connector_account: storage::MerchantConnectorAccountUpdateInternal,
     ) -> CustomResult<
         domain::merchant_connector_account::MerchantConnectorAccount,
         errors::StorageError,
diff --git a/crates/router/src/db/merchant_key_store.rs b/crates/router/src/db/merchant_key_store.rs
new file mode 100644
index 00000000000..b4637af8c0c
--- /dev/null
+++ b/crates/router/src/db/merchant_key_store.rs
@@ -0,0 +1,81 @@
+use error_stack::{IntoReport, ResultExt};
+
+use crate::{
+    connection,
+    core::errors::{self, CustomResult},
+    db::MockDb,
+    services::Store,
+    types::domain::{
+        behaviour::{Conversion, ReverseConversion},
+        merchant_key_store,
+    },
+};
+
+#[async_trait::async_trait]
+pub trait MerchantKeyStoreInterface {
+    async fn insert_merchant_key_store(
+        &self,
+        merchant_key_store: merchant_key_store::MerchantKeyStore,
+    ) -> CustomResult<merchant_key_store::MerchantKeyStore, errors::StorageError>;
+
+    async fn get_merchant_key_store_by_merchant_id(
+        &self,
+        merchant_id: &str,
+    ) -> CustomResult<merchant_key_store::MerchantKeyStore, errors::StorageError>;
+}
+
+#[async_trait::async_trait]
+impl MerchantKeyStoreInterface for Store {
+    async fn insert_merchant_key_store(
+        &self,
+        merchant_key_store: merchant_key_store::MerchantKeyStore,
+    ) -> CustomResult<merchant_key_store::MerchantKeyStore, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        let merchant_id = merchant_key_store.merchant_id.clone();
+        merchant_key_store
+            .construct_new()
+            .await
+            .change_context(errors::StorageError::EncryptionError)?
+            .insert(&conn)
+            .await
+            .map_err(Into::into)
+            .into_report()?
+            .convert(self, &merchant_id)
+            .await
+            .change_context(errors::StorageError::DecryptionError)
+    }
+    async fn get_merchant_key_store_by_merchant_id(
+        &self,
+        merchant_id: &str,
+    ) -> CustomResult<merchant_key_store::MerchantKeyStore, errors::StorageError> {
+        let conn = connection::pg_connection_read(self).await?;
+        storage_models::merchant_key_store::MerchantKeyStore::find_by_merchant_id(
+            &conn,
+            merchant_id,
+        )
+        .await
+        .map_err(Into::into)
+        .into_report()?
+        .convert(self, merchant_id)
+        .await
+        .change_context(errors::StorageError::DecryptionError)
+    }
+}
+
+#[async_trait::async_trait]
+impl MerchantKeyStoreInterface for MockDb {
+    async fn insert_merchant_key_store(
+        &self,
+        _merchant_key_store: merchant_key_store::MerchantKeyStore,
+    ) -> CustomResult<merchant_key_store::MerchantKeyStore, errors::StorageError> {
+        // [#172]: Implement function for `MockDb`
+        Err(errors::StorageError::MockDbError.into())
+    }
+    async fn get_merchant_key_store_by_merchant_id(
+        &self,
+        _merchant_id: &str,
+    ) -> CustomResult<merchant_key_store::MerchantKeyStore, errors::StorageError> {
+        // [#172]: Implement function for `MockDb`
+        Err(errors::StorageError::MockDbError.into())
+    }
+}
diff --git a/crates/router/src/routes/metrics.rs b/crates/router/src/routes/metrics.rs
index 133a01bb5cd..305cf5204ff 100644
--- a/crates/router/src/routes/metrics.rs
+++ b/crates/router/src/routes/metrics.rs
@@ -68,5 +68,9 @@ histogram_metric!(CARD_ADD_TIME, GLOBAL_METER);
 histogram_metric!(CARD_GET_TIME, GLOBAL_METER);
 histogram_metric!(CARD_DELETE_TIME, GLOBAL_METER);
 
+// Encryption and Decryption metrics
+histogram_metric!(ENCRYPTION_TIME, GLOBAL_METER);
+histogram_metric!(DECRYPTION_TIME, GLOBAL_METER);
+
 pub mod request;
 pub mod utils;
diff --git a/crates/router/src/routes/metrics/request.rs b/crates/router/src/routes/metrics/request.rs
index 92f6d823784..a38ba471ebf 100644
--- a/crates/router/src/routes/metrics/request.rs
+++ b/crates/router/src/routes/metrics/request.rs
@@ -19,7 +19,7 @@ where
 }
 
 #[inline]
-pub async fn record_card_operation_time<F, R>(
+pub async fn record_operation_time<F, R>(
     future: F,
     metric: &once_cell::sync::Lazy<router_env::opentelemetry::metrics::Histogram<f64>>,
 ) -> R
diff --git a/crates/router/src/services.rs b/crates/router/src/services.rs
index 74c4a71c6dc..587cce9c0d5 100644
--- a/crates/router/src/services.rs
+++ b/crates/router/src/services.rs
@@ -6,6 +6,8 @@ pub mod logger;
 use std::sync::{atomic, Arc};
 
 use error_stack::{IntoReport, ResultExt};
+#[cfg(feature = "kms")]
+use external_services::kms;
 use futures::StreamExt;
 use redis_interface::{errors as redis_errors, PubsubInterface};
 
@@ -80,6 +82,7 @@ pub struct Store {
     pub redis_conn: Arc<redis_interface::RedisConnectionPool>,
     #[cfg(feature = "kv_store")]
     pub(crate) config: StoreConfig,
+    pub master_key: Vec<u8>,
 }
 
 #[cfg(feature = "kv_store")]
@@ -108,6 +111,13 @@ impl Store {
             redis_clone.on_error().await;
         });
 
+        let master_enc_key = get_master_enc_key(
+            config,
+            #[cfg(feature = "kms")]
+            &config.kms,
+        )
+        .await;
+
         Self {
             master_pool: diesel_make_pg_pool(
                 &config.master_database,
@@ -130,6 +140,7 @@ impl Store {
                 drainer_stream_name: config.drainer.stream_name.clone(),
                 drainer_num_partitions: config.drainer.num_partitions,
             },
+            master_key: master_enc_key,
         }
     }
 
@@ -177,3 +188,37 @@ impl Store {
             .change_context(crate::core::errors::StorageError::KVError)
     }
 }
+
+#[allow(clippy::expect_used)]
+async fn get_master_enc_key(
+    conf: &crate::configs::settings::Settings,
+    #[cfg(feature = "kms")] kms_config: &kms::KmsConfig,
+) -> Vec<u8> {
+    #[cfg(feature = "kms")]
+    let master_enc_key = hex::decode(
+        kms::get_kms_client(kms_config)
+            .await
+            .decrypt(&conf.secrets.master_enc_key)
+            .await
+            .expect("Failed to decrypt master enc key"),
+    )
+    .expect("Failed to decode from hex");
+
+    #[cfg(not(feature = "kms"))]
+    let master_enc_key =
+        hex::decode(&conf.secrets.master_enc_key).expect("Failed to decode from hex");
+
+    master_enc_key
+}
+
+#[inline]
+pub fn generate_aes256_key() -> errors::CustomResult<[u8; 32], common_utils::errors::CryptoError> {
+    use ring::rand::SecureRandom;
+
+    let rng = ring::rand::SystemRandom::new();
+    let mut key: [u8; 256 / 8] = [0_u8; 256 / 8];
+    rng.fill(&mut key)
+        .into_report()
+        .change_context(common_utils::errors::CryptoError::EncodingFailed)?;
+    Ok(key)
+}
diff --git a/crates/router/src/types/domain.rs b/crates/router/src/types/domain.rs
index 252cd63d6d0..0eb5b6118a0 100644
--- a/crates/router/src/types/domain.rs
+++ b/crates/router/src/types/domain.rs
@@ -3,4 +3,5 @@ pub mod behaviour;
 pub mod customer;
 pub mod merchant_account;
 pub mod merchant_connector_account;
+pub mod merchant_key_store;
 pub mod types;
diff --git a/crates/router/src/types/domain/address.rs b/crates/router/src/types/domain/address.rs
index eebbff2b4f6..d69343cc798 100644
--- a/crates/router/src/types/domain/address.rs
+++ b/crates/router/src/types/domain/address.rs
@@ -12,6 +12,7 @@ use super::{
     behaviour,
     types::{self, AsyncLift},
 };
+use crate::db::StorageInterface;
 
 #[derive(Clone, Debug, serde::Serialize)]
 pub struct Address {
@@ -69,11 +70,19 @@ impl behaviour::Conversion for Address {
         })
     }
 
-    async fn convert_back(other: Self::DstType) -> CustomResult<Self, ValidationError> {
-        let key = &[0];
+    async fn convert_back(
+        other: Self::DstType,
+        db: &dyn StorageInterface,
+        merchant_id: &str,
+    ) -> CustomResult<Self, ValidationError> {
+        let key = types::get_merchant_enc_key(db, merchant_id)
+            .await
+            .change_context(ValidationError::InvalidValue {
+                message: "Failed while getting key from key store".to_string(),
+            })?;
 
         async {
-            let inner_decrypt = |inner| types::decrypt(inner, key);
+            let inner_decrypt = |inner| types::decrypt(inner, &key);
             Ok(Self {
                 id: Some(other.id),
                 address_id: other.address_id,
diff --git a/crates/router/src/types/domain/behaviour.rs b/crates/router/src/types/domain/behaviour.rs
index 298c7c83e66..2f95fe1611c 100644
--- a/crates/router/src/types/domain/behaviour.rs
+++ b/crates/router/src/types/domain/behaviour.rs
@@ -1,5 +1,7 @@
 use common_utils::errors::{CustomResult, ValidationError};
 
+use crate::db::StorageInterface;
+
 /// Trait for converting domain types to storage models
 #[async_trait::async_trait]
 pub trait Conversion {
@@ -7,7 +9,11 @@ pub trait Conversion {
     type NewDstType;
     async fn convert(self) -> CustomResult<Self::DstType, ValidationError>;
 
-    async fn convert_back(item: Self::DstType) -> CustomResult<Self, ValidationError>
+    async fn convert_back(
+        item: Self::DstType,
+        db: &dyn StorageInterface,
+        merchant_id: &str,
+    ) -> CustomResult<Self, ValidationError>
     where
         Self: Sized;
 
@@ -16,12 +22,20 @@ pub trait Conversion {
 
 #[async_trait::async_trait]
 pub trait ReverseConversion<SrcType: Conversion> {
-    async fn convert(self) -> CustomResult<SrcType, ValidationError>;
+    async fn convert(
+        self,
+        db: &dyn StorageInterface,
+        merchant_id: &str,
+    ) -> CustomResult<SrcType, ValidationError>;
 }
 
 #[async_trait::async_trait]
 impl<T: Send, U: Conversion<DstType = T>> ReverseConversion<U> for T {
-    async fn convert(self) -> CustomResult<U, ValidationError> {
-        U::convert_back(self).await
+    async fn convert(
+        self,
+        db: &dyn StorageInterface,
+        merchant_id: &str,
+    ) -> CustomResult<U, ValidationError> {
+        U::convert_back(self, db, merchant_id).await
     }
 }
diff --git a/crates/router/src/types/domain/customer.rs b/crates/router/src/types/domain/customer.rs
index d4e4d2b26e4..ca812bd8869 100644
--- a/crates/router/src/types/domain/customer.rs
+++ b/crates/router/src/types/domain/customer.rs
@@ -7,7 +7,10 @@ use storage_models::{customers::CustomerUpdateInternal, encryption::Encryption};
 use time::PrimitiveDateTime;
 
 use super::types::{self, AsyncLift};
-use crate::errors::{CustomResult, ValidationError};
+use crate::{
+    db::StorageInterface,
+    errors::{CustomResult, ValidationError},
+};
 
 #[derive(Clone, Debug)]
 pub struct Customer {
@@ -44,14 +47,22 @@ impl super::behaviour::Conversion for Customer {
         })
     }
 
-    async fn convert_back(item: Self::DstType) -> CustomResult<Self, ValidationError>
+    async fn convert_back(
+        item: Self::DstType,
+        db: &dyn StorageInterface,
+        merchant_id: &str,
+    ) -> CustomResult<Self, ValidationError>
     where
         Self: Sized,
     {
-        let key = &[0]; // To be replaced by key fetched from Store
+        let key = types::get_merchant_enc_key(db, merchant_id)
+            .await
+            .change_context(ValidationError::InvalidValue {
+                message: "Failed while getting key from key store".to_string(),
+            })?;
         async {
-            let inner_decrypt = |inner| types::decrypt(inner, key);
-            let inner_decrypt_email = |inner| types::decrypt(inner, key);
+            let inner_decrypt = |inner| types::decrypt(inner, &key);
+            let inner_decrypt_email = |inner| types::decrypt(inner, &key);
             Ok(Self {
                 id: Some(item.id),
                 customer_id: item.customer_id,
diff --git a/crates/router/src/types/domain/merchant_account.rs b/crates/router/src/types/domain/merchant_account.rs
index bd6f15d686b..76ab589889e 100644
--- a/crates/router/src/types/domain/merchant_account.rs
+++ b/crates/router/src/types/domain/merchant_account.rs
@@ -10,6 +10,7 @@ use storage_models::{
 };
 
 use crate::{
+    db::StorageInterface,
     errors::{CustomResult, ValidationError},
     types::domain::types::{self, AsyncLift, TypeEncryption},
 };
@@ -126,11 +127,19 @@ impl super::behaviour::Conversion for MerchantAccount {
         })
     }
 
-    async fn convert_back(item: Self::DstType) -> CustomResult<Self, ValidationError>
+    async fn convert_back(
+        item: Self::DstType,
+        db: &dyn StorageInterface,
+        merchant_id: &str,
+    ) -> CustomResult<Self, ValidationError>
     where
         Self: Sized,
     {
-        let key = &[0];
+        let key = types::get_merchant_enc_key(db, merchant_id.to_owned())
+            .await
+            .change_context(ValidationError::InvalidValue {
+                message: "Failed while getting key from key store".to_string(),
+            })?;
         async {
             Ok(Self {
                 id: Some(item.id),
@@ -141,11 +150,11 @@ impl super::behaviour::Conversion for MerchantAccount {
                 redirect_to_merchant_with_http_post: item.redirect_to_merchant_with_http_post,
                 merchant_name: item
                     .merchant_name
-                    .async_lift(|inner| types::decrypt(inner, key))
+                    .async_lift(|inner| types::decrypt(inner, &key))
                     .await?,
                 merchant_details: item
                     .merchant_details
-                    .async_lift(|inner| types::decrypt(inner, key))
+                    .async_lift(|inner| types::decrypt(inner, &key))
                     .await?,
                 webhook_details: item.webhook_details,
                 sub_merchants_enabled: item.sub_merchants_enabled,
@@ -157,7 +166,7 @@ impl super::behaviour::Conversion for MerchantAccount {
                 routing_algorithm: item.routing_algorithm,
                 api_key: item
                     .api_key
-                    .async_map(|value| Encryptable::decrypt(value, key, GcmAes256 {}))
+                    .async_map(|value| Encryptable::decrypt(value, &key, GcmAes256 {}))
                     .await
                     .transpose()?,
             })
diff --git a/crates/router/src/types/domain/merchant_connector_account.rs b/crates/router/src/types/domain/merchant_connector_account.rs
index 84db026c43a..52ad4db926a 100644
--- a/crates/router/src/types/domain/merchant_connector_account.rs
+++ b/crates/router/src/types/domain/merchant_connector_account.rs
@@ -1,18 +1,27 @@
 use common_utils::{
+    crypto::{Encryptable, GcmAes256},
     errors::{CustomResult, ValidationError},
     pii,
 };
-use masking::{ExposeInterface, Secret};
-use storage_models::enums;
+use error_stack::ResultExt;
+use masking::Secret;
+use storage_models::{
+    encryption::Encryption, enums,
+    merchant_connector_account::MerchantConnectorAccountUpdateInternal,
+};
 
-use super::behaviour;
+use super::{
+    behaviour,
+    types::{self, TypeEncryption},
+};
+use crate::db::StorageInterface;
 
-#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+#[derive(Clone, Debug)]
 pub struct MerchantConnectorAccount {
     pub id: Option<i32>,
     pub merchant_id: String,
     pub connector_name: String,
-    pub connector_account_details: Secret<serde_json::Value>,
+    pub connector_account_details: Encryptable<Secret<serde_json::Value>>,
     pub test_mode: Option<bool>,
     pub disabled: Option<bool>,
     pub merchant_connector_id: String,
@@ -21,6 +30,21 @@ pub struct MerchantConnectorAccount {
     pub metadata: Option<pii::SecretSerdeValue>,
 }
 
+#[derive(Debug)]
+pub enum MerchantConnectorAccountUpdate {
+    Update {
+        merchant_id: Option<String>,
+        connector_type: Option<enums::ConnectorType>,
+        connector_name: Option<String>,
+        connector_account_details: Option<Encryptable<Secret<serde_json::Value>>>,
+        test_mode: Option<bool>,
+        disabled: Option<bool>,
+        merchant_connector_id: Option<String>,
+        payment_methods_enabled: Option<Vec<serde_json::Value>>,
+        metadata: Option<pii::SecretSerdeValue>,
+    },
+}
+
 #[async_trait::async_trait]
 impl behaviour::Conversion for MerchantConnectorAccount {
     type DstType = storage_models::merchant_connector_account::MerchantConnectorAccount;
@@ -34,7 +58,7 @@ impl behaviour::Conversion for MerchantConnectorAccount {
                 })?,
                 merchant_id: self.merchant_id,
                 connector_name: self.connector_name,
-                connector_account_details: self.connector_account_details.expose(),
+                connector_account_details: self.connector_account_details.into(),
                 test_mode: self.test_mode,
                 disabled: self.disabled,
                 merchant_connector_id: self.merchant_connector_id,
@@ -45,12 +69,29 @@ impl behaviour::Conversion for MerchantConnectorAccount {
         )
     }
 
-    async fn convert_back(other: Self::DstType) -> CustomResult<Self, ValidationError> {
+    async fn convert_back(
+        other: Self::DstType,
+        db: &dyn StorageInterface,
+        merchant_id: &str,
+    ) -> CustomResult<Self, ValidationError> {
+        let key = types::get_merchant_enc_key(db, merchant_id)
+            .await
+            .change_context(ValidationError::InvalidValue {
+                message: "Error while getting key from keystore".to_string(),
+            })?;
         Ok(Self {
             id: Some(other.id),
             merchant_id: other.merchant_id,
             connector_name: other.connector_name,
-            connector_account_details: other.connector_account_details.into(),
+            connector_account_details: Encryptable::decrypt(
+                other.connector_account_details,
+                &key,
+                GcmAes256 {},
+            )
+            .await
+            .change_context(ValidationError::InvalidValue {
+                message: "Failed while decrypting connector account details".to_string(),
+            })?,
             test_mode: other.test_mode,
             disabled: other.disabled,
             merchant_connector_id: other.merchant_connector_id,
@@ -64,7 +105,7 @@ impl behaviour::Conversion for MerchantConnectorAccount {
         Ok(Self::NewDstType {
             merchant_id: Some(self.merchant_id),
             connector_name: Some(self.connector_name),
-            connector_account_details: Some(self.connector_account_details),
+            connector_account_details: Some(self.connector_account_details.into()),
             test_mode: self.test_mode,
             disabled: self.disabled,
             merchant_connector_id: self.merchant_connector_id,
@@ -74,3 +115,31 @@ impl behaviour::Conversion for MerchantConnectorAccount {
         })
     }
 }
+
+impl From<MerchantConnectorAccountUpdate> for MerchantConnectorAccountUpdateInternal {
+    fn from(merchant_connector_account_update: MerchantConnectorAccountUpdate) -> Self {
+        match merchant_connector_account_update {
+            MerchantConnectorAccountUpdate::Update {
+                merchant_id,
+                connector_type,
+                connector_name,
+                connector_account_details,
+                test_mode,
+                disabled,
+                merchant_connector_id,
+                payment_methods_enabled,
+                metadata,
+            } => Self {
+                merchant_id,
+                connector_type,
+                connector_name,
+                connector_account_details: connector_account_details.map(Encryption::from),
+                test_mode,
+                disabled,
+                merchant_connector_id,
+                payment_methods_enabled,
+                metadata,
+            },
+        }
+    }
+}
diff --git a/crates/router/src/types/domain/merchant_key_store.rs b/crates/router/src/types/domain/merchant_key_store.rs
new file mode 100644
index 00000000000..c4c0f4d830d
--- /dev/null
+++ b/crates/router/src/types/domain/merchant_key_store.rs
@@ -0,0 +1,61 @@
+use common_utils::{
+    crypto::{Encryptable, GcmAes256},
+    custom_serde,
+};
+use error_stack::ResultExt;
+use masking::Secret;
+use time::PrimitiveDateTime;
+
+use crate::{
+    db::StorageInterface,
+    errors::{CustomResult, ValidationError},
+    types::domain::types::TypeEncryption,
+};
+
+#[derive(Clone, Debug, serde::Serialize)]
+pub struct MerchantKeyStore {
+    pub merchant_id: String,
+    pub key: Encryptable<Secret<Vec<u8>>>,
+    #[serde(with = "custom_serde::iso8601")]
+    pub created_at: PrimitiveDateTime,
+}
+
+#[async_trait::async_trait]
+impl super::behaviour::Conversion for MerchantKeyStore {
+    type DstType = storage_models::merchant_key_store::MerchantKeyStore;
+    type NewDstType = storage_models::merchant_key_store::MerchantKeyStoreNew;
+    async fn convert(self) -> CustomResult<Self::DstType, ValidationError> {
+        Ok(storage_models::merchant_key_store::MerchantKeyStore {
+            key: self.key.into(),
+            merchant_id: self.merchant_id,
+            created_at: self.created_at,
+        })
+    }
+
+    async fn convert_back(
+        item: Self::DstType,
+        db: &dyn StorageInterface,
+        _merchant_id: &str,
+    ) -> CustomResult<Self, ValidationError>
+    where
+        Self: Sized,
+    {
+        let key = &db.get_master_key();
+        Ok(Self {
+            key: Encryptable::decrypt(item.key, key, GcmAes256 {})
+                .await
+                .change_context(ValidationError::InvalidValue {
+                    message: "Failed while decrypting customer data".to_string(),
+                })?,
+            merchant_id: item.merchant_id,
+            created_at: item.created_at,
+        })
+    }
+
+    async fn construct_new(self) -> CustomResult<Self::NewDstType, ValidationError> {
+        Ok(storage_models::merchant_key_store::MerchantKeyStoreNew {
+            merchant_id: self.merchant_id,
+            key: self.key.into(),
+        })
+    }
+}
diff --git a/crates/router/src/types/domain/types.rs b/crates/router/src/types/domain/types.rs
index ade0ecd9d96..80dab9d1b0f 100644
--- a/crates/router/src/types/domain/types.rs
+++ b/crates/router/src/types/domain/types.rs
@@ -5,10 +5,11 @@ use common_utils::{
     ext_traits::AsyncExt,
 };
 use error_stack::{IntoReport, ResultExt};
-use masking::{PeekInterface, Secret};
+use masking::{ExposeInterface, PeekInterface, Secret};
+use router_env::{instrument, tracing};
 use storage_models::encryption::Encryption;
 
-use crate::core::errors as core_errors;
+use crate::routes::metrics::{request, DECRYPTION_TIME, ENCRYPTION_TIME};
 
 #[async_trait]
 pub trait TypeEncryption<
@@ -35,6 +36,7 @@ impl<
         S: masking::Strategy<String> + Send,
     > TypeEncryption<String, V, S> for crypto::Encryptable<Secret<String, S>>
 {
+    #[instrument(skip_all)]
     async fn encrypt(
         masked_data: Secret<String, S>,
         key: &[u8],
@@ -45,6 +47,7 @@ impl<
         Ok(Self::new(masked_data, encrypted_data))
     }
 
+    #[instrument(skip_all)]
     async fn decrypt(
         encrypted_data: Encryption,
         key: &[u8],
@@ -68,6 +71,7 @@ impl<
     > TypeEncryption<serde_json::Value, V, S>
     for crypto::Encryptable<Secret<serde_json::Value, S>>
 {
+    #[instrument(skip_all)]
     async fn encrypt(
         masked_data: Secret<serde_json::Value, S>,
         key: &[u8],
@@ -81,6 +85,7 @@ impl<
         Ok(Self::new(masked_data, encrypted_data))
     }
 
+    #[instrument(skip_all)]
     async fn decrypt(
         encrypted_data: Encryption,
         key: &[u8],
@@ -96,11 +101,47 @@ impl<
     }
 }
 
-pub async fn get_key_and_algo(
-    _db: &dyn crate::db::StorageInterface,
-    _merchant_id: String,
-) -> CustomResult<Vec<u8>, core_errors::StorageError> {
-    Ok(Vec::new())
+#[async_trait]
+impl<
+        V: crypto::DecodeMessage + crypto::EncodeMessage + Send + 'static,
+        S: masking::Strategy<Vec<u8>> + Send,
+    > TypeEncryption<Vec<u8>, V, S> for crypto::Encryptable<Secret<Vec<u8>, S>>
+{
+    #[instrument(skip_all)]
+    async fn encrypt(
+        masked_data: Secret<Vec<u8>, S>,
+        key: &[u8],
+        crypt_algo: V,
+    ) -> CustomResult<Self, errors::CryptoError> {
+        let encrypted_data = crypt_algo.encode_message(key, masked_data.peek())?;
+
+        Ok(Self::new(masked_data, encrypted_data))
+    }
+
+    #[instrument(skip_all)]
+    async fn decrypt(
+        encrypted_data: Encryption,
+        key: &[u8],
+        crypt_algo: V,
+    ) -> CustomResult<Self, errors::CryptoError> {
+        let encrypted = encrypted_data.into_inner();
+        let decrypted_data = crypt_algo.decode_message(key, encrypted.clone())?;
+
+        Ok(Self::new(decrypted_data.into(), encrypted))
+    }
+}
+
+pub async fn get_merchant_enc_key(
+    db: &dyn crate::db::StorageInterface,
+    merchant_id: impl AsRef<str>,
+) -> CustomResult<Vec<u8>, crate::core::errors::StorageError> {
+    let merchant_id = merchant_id.as_ref();
+    let key = db
+        .get_merchant_key_store_by_merchant_id(merchant_id)
+        .await?
+        .key
+        .into_inner();
+    Ok(key.expose())
 }
 
 pub trait Lift<U> {
@@ -149,6 +190,21 @@ impl<U, V: Lift<U> + Lift<U, SelfWrapper<U> = V> + Send> AsyncLift<U> for V {
     }
 }
 
+pub(crate) async fn encrypt<E: Clone, S>(
+    inner: Secret<E, S>,
+    key: &[u8],
+) -> CustomResult<crypto::Encryptable<Secret<E, S>>, errors::CryptoError>
+where
+    S: masking::Strategy<E>,
+    crypto::Encryptable<Secret<E, S>>: TypeEncryption<E, crypto::GcmAes256, S>,
+{
+    request::record_operation_time(
+        crypto::Encryptable::encrypt(inner, key, crypto::GcmAes256 {}),
+        &ENCRYPTION_TIME,
+    )
+    .await
+}
+
 pub(crate) async fn decrypt<T: Clone, S: masking::Strategy<T>>(
     inner: Option<Encryption>,
     key: &[u8],
@@ -156,8 +212,10 @@ pub(crate) async fn decrypt<T: Clone, S: masking::Strategy<T>>(
 where
     crypto::Encryptable<Secret<T, S>>: TypeEncryption<T, crypto::GcmAes256, S>,
 {
-    inner
-        .async_map(|item| crypto::Encryptable::decrypt(item, key, crypto::GcmAes256 {}))
-        .await
-        .transpose()
+    request::record_operation_time(
+        inner.async_map(|item| crypto::Encryptable::decrypt(item, key, crypto::GcmAes256 {})),
+        &DECRYPTION_TIME,
+    )
+    .await
+    .transpose()
 }
diff --git a/crates/router/src/types/storage/merchant_connector_account.rs b/crates/router/src/types/storage/merchant_connector_account.rs
index 249538f5823..700a97048a9 100644
--- a/crates/router/src/types/storage/merchant_connector_account.rs
+++ b/crates/router/src/types/storage/merchant_connector_account.rs
@@ -1,4 +1,5 @@
 pub use storage_models::merchant_connector_account::{
-    MerchantConnectorAccount, MerchantConnectorAccountNew, MerchantConnectorAccountUpdate,
-    MerchantConnectorAccountUpdateInternal,
+    MerchantConnectorAccount, MerchantConnectorAccountNew, MerchantConnectorAccountUpdateInternal,
 };
+
+pub use crate::types::domain::merchant_connector_account::MerchantConnectorAccountUpdate;
diff --git a/crates/router/src/types/transformers.rs b/crates/router/src/types/transformers.rs
index 2f387f26509..f4d8a70f817 100644
--- a/crates/router/src/types/transformers.rs
+++ b/crates/router/src/types/transformers.rs
@@ -332,7 +332,7 @@ impl TryFrom<domain::merchant_connector_account::MerchantConnectorAccount>
             connector_type: merchant_ca.connector_type.foreign_into(),
             connector_name: merchant_ca.connector_name,
             merchant_connector_id: Some(merchant_ca.merchant_connector_id),
-            connector_account_details: Some(merchant_ca.connector_account_details),
+            connector_account_details: Some(merchant_ca.connector_account_details.into_inner()),
             test_mode: merchant_ca.test_mode,
             disabled: merchant_ca.disabled,
             metadata: merchant_ca.metadata,
diff --git a/crates/storage_models/src/encryption.rs b/crates/storage_models/src/encryption.rs
index e0d9e054f21..a32dd45682e 100644
--- a/crates/storage_models/src/encryption.rs
+++ b/crates/storage_models/src/encryption.rs
@@ -1,4 +1,9 @@
-use diesel::{backend::Backend, deserialize::FromSql, serialize::ToSql, sql_types, AsExpression};
+use diesel::{
+    backend::Backend,
+    deserialize::{self, FromSql, Queryable},
+    serialize::ToSql,
+    sql_types, AsExpression,
+};
 
 #[derive(Debug, AsExpression, Clone, serde::Serialize, serde::Deserialize)]
 #[diesel(sql_type = diesel::sql_types::Binary)]
@@ -51,3 +56,14 @@ where
         self.get_inner().to_sql(out)
     }
 }
+
+impl<DB> Queryable<sql_types::Binary, DB> for Encryption
+where
+    DB: Backend,
+    Vec<u8>: FromSql<sql_types::Binary, DB>,
+{
+    type Row = Vec<u8>;
+    fn build(row: Self::Row) -> deserialize::Result<Self> {
+        Ok(Self { inner: row })
+    }
+}
diff --git a/crates/storage_models/src/lib.rs b/crates/storage_models/src/lib.rs
index 4598f4db145..59eed2c9684 100644
--- a/crates/storage_models/src/lib.rs
+++ b/crates/storage_models/src/lib.rs
@@ -16,6 +16,7 @@ pub mod locker_mock_up;
 pub mod mandate;
 pub mod merchant_account;
 pub mod merchant_connector_account;
+pub mod merchant_key_store;
 pub mod payment_attempt;
 pub mod payment_intent;
 pub mod payment_method;
diff --git a/crates/storage_models/src/merchant_connector_account.rs b/crates/storage_models/src/merchant_connector_account.rs
index 08d3fa21fa3..189dd0770f6 100644
--- a/crates/storage_models/src/merchant_connector_account.rs
+++ b/crates/storage_models/src/merchant_connector_account.rs
@@ -1,16 +1,13 @@
 use common_utils::pii;
 use diesel::{AsChangeset, Identifiable, Insertable, Queryable};
-use masking::Secret;
 
-use crate::{enums as storage_enums, schema::merchant_connector_account};
+use crate::{encryption::Encryption, enums as storage_enums, schema::merchant_connector_account};
 
 #[derive(
     Clone,
     Debug,
-    Eq,
     serde::Serialize,
     serde::Deserialize,
-    PartialEq,
     Identifiable,
     Queryable,
     router_derive::DebugAsDisplay,
@@ -20,7 +17,7 @@ pub struct MerchantConnectorAccount {
     pub id: i32,
     pub merchant_id: String,
     pub connector_name: String,
-    pub connector_account_details: serde_json::Value,
+    pub connector_account_details: Encryption,
     pub test_mode: Option<bool>,
     pub disabled: Option<bool>,
     pub merchant_connector_id: String,
@@ -36,7 +33,7 @@ pub struct MerchantConnectorAccountNew {
     pub merchant_id: Option<String>,
     pub connector_type: Option<storage_enums::ConnectorType>,
     pub connector_name: Option<String>,
-    pub connector_account_details: Option<Secret<serde_json::Value>>,
+    pub connector_account_details: Option<Encryption>,
     pub test_mode: Option<bool>,
     pub disabled: Option<bool>,
     pub merchant_connector_id: String,
@@ -44,58 +41,16 @@ pub struct MerchantConnectorAccountNew {
     pub metadata: Option<pii::SecretSerdeValue>,
 }
 
-#[derive(Debug)]
-pub enum MerchantConnectorAccountUpdate {
-    Update {
-        merchant_id: Option<String>,
-        connector_type: Option<storage_enums::ConnectorType>,
-        connector_name: Option<String>,
-        connector_account_details: Option<Secret<serde_json::Value>>,
-        test_mode: Option<bool>,
-        disabled: Option<bool>,
-        merchant_connector_id: Option<String>,
-        payment_methods_enabled: Option<Vec<serde_json::Value>>,
-        metadata: Option<pii::SecretSerdeValue>,
-    },
-}
 #[derive(Clone, Debug, Default, AsChangeset, router_derive::DebugAsDisplay)]
 #[diesel(table_name = merchant_connector_account)]
 pub struct MerchantConnectorAccountUpdateInternal {
-    merchant_id: Option<String>,
-    connector_type: Option<storage_enums::ConnectorType>,
-    connector_name: Option<String>,
-    connector_account_details: Option<Secret<serde_json::Value>>,
-    test_mode: Option<bool>,
-    disabled: Option<bool>,
-    merchant_connector_id: Option<String>,
-    payment_methods_enabled: Option<Vec<serde_json::Value>>,
-    metadata: Option<pii::SecretSerdeValue>,
-}
-
-impl From<MerchantConnectorAccountUpdate> for MerchantConnectorAccountUpdateInternal {
-    fn from(merchant_connector_account_update: MerchantConnectorAccountUpdate) -> Self {
-        match merchant_connector_account_update {
-            MerchantConnectorAccountUpdate::Update {
-                merchant_id,
-                connector_type,
-                connector_name,
-                connector_account_details,
-                test_mode,
-                disabled,
-                merchant_connector_id,
-                payment_methods_enabled,
-                metadata,
-            } => Self {
-                merchant_id,
-                connector_type,
-                connector_name,
-                connector_account_details,
-                test_mode,
-                disabled,
-                merchant_connector_id,
-                payment_methods_enabled,
-                metadata,
-            },
-        }
-    }
+    pub merchant_id: Option<String>,
+    pub connector_type: Option<storage_enums::ConnectorType>,
+    pub connector_name: Option<String>,
+    pub connector_account_details: Option<Encryption>,
+    pub test_mode: Option<bool>,
+    pub disabled: Option<bool>,
+    pub merchant_connector_id: Option<String>,
+    pub payment_methods_enabled: Option<Vec<serde_json::Value>>,
+    pub metadata: Option<pii::SecretSerdeValue>,
 }
diff --git a/crates/storage_models/src/merchant_key_store.rs b/crates/storage_models/src/merchant_key_store.rs
new file mode 100644
index 00000000000..90d24f5b932
--- /dev/null
+++ b/crates/storage_models/src/merchant_key_store.rs
@@ -0,0 +1,41 @@
+use common_utils::custom_serde;
+use diesel::{AsChangeset, Identifiable, Insertable, Queryable};
+use time::PrimitiveDateTime;
+
+use crate::{encryption::Encryption, schema::merchantkeystore};
+
+#[derive(
+    Clone,
+    Debug,
+    serde::Serialize,
+    serde::Deserialize,
+    Identifiable,
+    Queryable,
+    router_derive::DebugAsDisplay,
+)]
+#[diesel(table_name = merchantkeystore)]
+#[diesel(primary_key(merchant_id))]
+pub struct MerchantKeyStore {
+    pub merchant_id: String,
+    pub key: Encryption,
+    #[serde(with = "custom_serde::iso8601")]
+    pub created_at: PrimitiveDateTime,
+}
+
+#[derive(
+    Clone, Debug, serde::Serialize, serde::Deserialize, Insertable, router_derive::DebugAsDisplay,
+)]
+#[diesel(table_name = merchantkeystore)]
+pub struct MerchantKeyStoreNew {
+    pub merchant_id: String,
+    pub key: Encryption,
+}
+
+#[derive(
+    Clone, Debug, serde::Serialize, serde::Deserialize, AsChangeset, router_derive::DebugAsDisplay,
+)]
+#[diesel(table_name = merchantkeystore)]
+pub struct MerchantKeyStoreUpdateInternal {
+    pub merchant_id: String,
+    pub key: Encryption,
+}
diff --git a/crates/storage_models/src/query.rs b/crates/storage_models/src/query.rs
index 83baa4502c2..271bf9e1aef 100644
--- a/crates/storage_models/src/query.rs
+++ b/crates/storage_models/src/query.rs
@@ -11,6 +11,7 @@ pub mod locker_mock_up;
 pub mod mandate;
 pub mod merchant_account;
 pub mod merchant_connector_account;
+pub mod merchant_key_store;
 pub mod payment_attempt;
 pub mod payment_intent;
 pub mod payment_method;
diff --git a/crates/storage_models/src/query/merchant_connector_account.rs b/crates/storage_models/src/query/merchant_connector_account.rs
index 6e63cdf6e66..18aab8a0802 100644
--- a/crates/storage_models/src/query/merchant_connector_account.rs
+++ b/crates/storage_models/src/query/merchant_connector_account.rs
@@ -5,7 +5,7 @@ use super::generics;
 use crate::{
     errors,
     merchant_connector_account::{
-        MerchantConnectorAccount, MerchantConnectorAccountNew, MerchantConnectorAccountUpdate,
+        MerchantConnectorAccount, MerchantConnectorAccountNew,
         MerchantConnectorAccountUpdateInternal,
     },
     schema::merchant_connector_account::dsl,
@@ -24,12 +24,12 @@ impl MerchantConnectorAccount {
     pub async fn update(
         self,
         conn: &PgPooledConn,
-        merchant_connector_account: MerchantConnectorAccountUpdate,
+        merchant_connector_account: MerchantConnectorAccountUpdateInternal,
     ) -> StorageResult<Self> {
         match generics::generic_update_by_id::<<Self as HasTable>::Table, _, _, _>(
             conn,
             self.id,
-            MerchantConnectorAccountUpdateInternal::from(merchant_connector_account),
+            merchant_connector_account,
         )
         .await
         {
diff --git a/crates/storage_models/src/query/merchant_key_store.rs b/crates/storage_models/src/query/merchant_key_store.rs
new file mode 100644
index 00000000000..e681c8c1dff
--- /dev/null
+++ b/crates/storage_models/src/query/merchant_key_store.rs
@@ -0,0 +1,30 @@
+use diesel::{associations::HasTable, ExpressionMethods};
+use router_env::{instrument, tracing};
+
+use super::generics;
+use crate::{
+    merchant_key_store::{MerchantKeyStore, MerchantKeyStoreNew},
+    schema::merchantkeystore::dsl,
+    PgPooledConn, StorageResult,
+};
+
+impl MerchantKeyStoreNew {
+    #[instrument(skip(conn))]
+    pub async fn insert(self, conn: &PgPooledConn) -> StorageResult<MerchantKeyStore> {
+        generics::generic_insert(conn, self).await
+    }
+}
+
+impl MerchantKeyStore {
+    #[instrument(skip(conn))]
+    pub async fn find_by_merchant_id(
+        conn: &PgPooledConn,
+        merchant_id: &str,
+    ) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::merchant_id.eq(merchant_id.to_owned()),
+        )
+        .await
+    }
+}
diff --git a/crates/storage_models/src/schema.rs b/crates/storage_models/src/schema.rs
index 340c602efe2..06bb2163c52 100644
--- a/crates/storage_models/src/schema.rs
+++ b/crates/storage_models/src/schema.rs
@@ -9,14 +9,14 @@ diesel::table! {
         address_id -> Varchar,
         city -> Nullable<Varchar>,
         country -> Nullable<CountryCode>,
-        line1 -> Nullable<Binary>,
-        line2 -> Nullable<Binary>,
-        line3 -> Nullable<Binary>,
-        state -> Nullable<Binary>,
-        zip -> Nullable<Binary>,
-        first_name -> Nullable<Binary>,
-        last_name -> Nullable<Binary>,
-        phone_number -> Nullable<Binary>,
+        line1 -> Nullable<Bytea>,
+        line2 -> Nullable<Bytea>,
+        line3 -> Nullable<Bytea>,
+        state -> Nullable<Bytea>,
+        zip -> Nullable<Bytea>,
+        first_name -> Nullable<Bytea>,
+        last_name -> Nullable<Bytea>,
+        phone_number -> Nullable<Bytea>,
         country_code -> Nullable<Varchar>,
         created_at -> Timestamp,
         modified_at -> Timestamp,
@@ -99,9 +99,9 @@ diesel::table! {
         id -> Int4,
         customer_id -> Varchar,
         merchant_id -> Varchar,
-        name -> Nullable<Binary>,
-        email -> Nullable<Binary>,
-        phone -> Nullable<Binary>,
+        name -> Nullable<Bytea>,
+        email -> Nullable<Bytea>,
+        phone -> Nullable<Bytea>,
         phone_country_code -> Nullable<Varchar>,
         description -> Nullable<Varchar>,
         created_at -> Timestamp,
@@ -212,8 +212,8 @@ diesel::table! {
         enable_payment_response_hash -> Bool,
         payment_response_hash_key -> Nullable<Varchar>,
         redirect_to_merchant_with_http_post -> Bool,
-        merchant_name -> Nullable<Binary>,
-        merchant_details -> Nullable<Binary>,
+        merchant_name -> Nullable<Bytea>,
+        merchant_details -> Nullable<Bytea>,
         webhook_details -> Nullable<Json>,
         sub_merchants_enabled -> Nullable<Bool>,
         parent_merchant_id -> Nullable<Varchar>,
@@ -222,7 +222,7 @@ diesel::table! {
         locker_id -> Nullable<Varchar>,
         metadata -> Nullable<Jsonb>,
         routing_algorithm -> Nullable<Json>,
-        api_key -> Nullable<Binary>,
+        api_key -> Nullable<Bytea>,
     }
 }
 
@@ -234,7 +234,7 @@ diesel::table! {
         id -> Int4,
         merchant_id -> Varchar,
         connector_name -> Varchar,
-        connector_account_details -> Json,
+        connector_account_details -> Bytea,
         test_mode -> Nullable<Bool>,
         disabled -> Nullable<Bool>,
         merchant_connector_id -> Varchar,
@@ -244,6 +244,17 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    use diesel::sql_types::*;
+    use crate::enums::diesel_exports::*;
+
+    merchantkeystore (merchant_id) {
+        merchant_id -> Varchar,
+        key -> Bytea,
+        created_at -> Timestamp,
+    }
+}
+
 diesel::table! {
     use diesel::sql_types::*;
     use crate::enums::diesel_exports::*;
@@ -423,6 +434,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     mandate,
     merchant_account,
     merchant_connector_account,
+    merchantkeystore,
     payment_attempt,
     payment_intent,
     payment_methods,
diff --git a/migrations/2023-04-06-092008_create_merchant_ek/down.sql b/migrations/2023-04-06-092008_create_merchant_ek/down.sql
new file mode 100644
index 00000000000..de0fb814836
--- /dev/null
+++ b/migrations/2023-04-06-092008_create_merchant_ek/down.sql
@@ -0,0 +1 @@
+DROP TABLE MERCHANTKEYSTORE;
diff --git a/migrations/2023-04-06-092008_create_merchant_ek/up.sql b/migrations/2023-04-06-092008_create_merchant_ek/up.sql
new file mode 100644
index 00000000000..3703f68b937
--- /dev/null
+++ b/migrations/2023-04-06-092008_create_merchant_ek/up.sql
@@ -0,0 +1,7 @@
+CREATE TABLE MERCHANTKEYSTORE(
+    MERCHANT_ID VARCHAR(255) NOT NULL PRIMARY KEY,
+    KEY BYTEA NOT NULL,
+    CREATED_AT TIMESTAMP NOT NULL DEFAULT now()
+);
+
+CREATE UNIQUE INDEX merchant_key_store_unique_index ON merchantkeystore(merchant_id);

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

## Description
<!-- Describe your changes in detail -->
This PR adds key store DB which stores encryption key per merchant account. The encryption key will be further encrypted by `master_key` which will further be KMS encrypted. 

### Additional Changes

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

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

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


## Motivation and Context
<!--
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 its an obvious bug or documentation fix
that will have little conversation).
-->


## 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)?
-->
Manual

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