Issue:
Bug: feat(events): add payment metadata to the payment intent events



We generate payment intent events on db changes,

for this we rely on the database payment intent struct for generating a corresponding struct (`KafkaPaymentIntent`).

we need to add the metadata, hashed customer_email to the generated event as well.

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [x] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
 - `metadata` is marked as `pii::SecretSerdeValue`, now changed it to `serde_json::Value` to remove the masking.
- Added the `metadata` field for `hyperswitch-payment-intent-events` (type - String).

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
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).
-->
Adding the metadata to the payment-intent-event helps in providing better information about the event.

## 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)?
-->
Make a normal payment through Postman by following the steps:
- Merchant Account - Create
- API Key - Create
- Payment Connector - Create
- Payments - Create

You should now be able to see the `metadata` field in hyperswitch-payment-intent-events when viewed in kafka-ui.
<img width="1495" alt="Screenshot 2024-07-01 at 6 57 29 PM" src="https://github.com/juspay/hyperswitch/assets/83278309/f678591d-1d83-4b18-856e-f650cb2eeb6f">
And can also be searched using global search in the dashboard.
<img width="805" alt="Screenshot 2024-07-02 at 1 06 11 PM" src="https://github.com/juspay/hyperswitch/assets/83278309/b2c0d10f-d55d-487c-8950-4aa22ca21559">
<img width="418" alt="Screenshot 2024-07-02 at 1 06 54 PM" src="https://github.com/juspay/hyperswitch/assets/83278309/e467ed28-b95c-4301-b90d-47b672f142f2">

Metadata is of type - String
<img width="1327" alt="Screenshot 2024-07-02 at 3 55 37 PM" src="https://github.com/juspay/hyperswitch/assets/83278309/ebc47d7b-6427-43d3-84c4-4302ccf62746">

## 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`
- [ ] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible


Testing:

Make a normal payment through Postman by following the steps:
- Merchant Account - Create
- API Key - Create
- Payment Connector - Create
- Payments - Create

You should now be able to see the `metadata` field in hyperswitch-payment-intent-events when viewed in kafka-ui.
<img width="1495" alt="Screenshot 2024-07-01 at 6 57 29 PM" src="https://github.com/juspay/hyperswitch/assets/83278309/f678591d-1d83-4b18-856e-f650cb2eeb6f">
And can also be searched using global search in the dashboard.
<img width="805" alt="Screenshot 2024-07-02 at 1 06 11 PM" src="https://github.com/juspay/hyperswitch/assets/83278309/b2c0d10f-d55d-487c-8950-4aa22ca21559">
<img width="418" alt="Screenshot 2024-07-02 at 1 06 54 PM" src="https://github.com/juspay/hyperswitch/assets/83278309/e467ed28-b95c-4301-b90d-47b672f142f2">

Metadata is of type - String
<img width="1327" alt="Screenshot 2024-07-02 at 3 55 37 PM" src="https://github.com/juspay/hyperswitch/assets/83278309/ebc47d7b-6427-43d3-84c4-4302ccf62746">

Patch:
diff --git a/config/development.toml b/config/development.toml
index 556b2e24afe..31a507a69a2 100644
--- a/config/development.toml
+++ b/config/development.toml
@@ -663,7 +663,7 @@ host = "https://localhost:9200"
 [opensearch.auth]
 auth = "basic"
 username = "admin"
-password = "admin"
+password = "0penS3arc#"
 region = "eu-central-1"
 
 [opensearch.indexes]
diff --git a/config/docker_compose.toml b/config/docker_compose.toml
index 28bf8fe8bb4..e2ac7b9ed9d 100644
--- a/config/docker_compose.toml
+++ b/config/docker_compose.toml
@@ -495,7 +495,7 @@ host = "https://opensearch:9200"
 [opensearch.auth]
 auth = "basic"
 username = "admin"
-password = "admin"
+password = "0penS3arc#"
 region = "eu-central-1"
 
 [opensearch.indexes]
diff --git a/crates/api_models/src/payments.rs b/crates/api_models/src/payments.rs
index c0b9ce756f7..31f75c05c32 100644
--- a/crates/api_models/src/payments.rs
+++ b/crates/api_models/src/payments.rs
@@ -454,7 +454,7 @@ pub struct PaymentsRequest {
 
     /// You can specify up to 50 keys, with key names up to 40 characters long and values up to 500 characters long. Metadata is useful for storing additional, structured information on an object.
     #[schema(value_type = Option<Object>, example = r#"{ "udf1": "some-value", "udf2": "some-value" }"#)]
-    pub metadata: Option<pii::SecretSerdeValue>,
+    pub metadata: Option<serde_json::Value>,
 
     /// Some connectors like Apple pay, Airwallex and Noon might require some additional information, find specific details in the child attributes below.
     pub connector_metadata: Option<ConnectorMetadata>,
@@ -3557,7 +3557,7 @@ pub struct PaymentsResponse {
 
     /// You can specify up to 50 keys, with key names up to 40 characters long and values up to 500 characters long. Metadata is useful for storing additional, structured information on an object.
     #[schema(value_type = Option<Object>, example = r#"{ "udf1": "some-value", "udf2": "some-value" }"#)]
-    pub metadata: Option<pii::SecretSerdeValue>,
+    pub metadata: Option<serde_json::Value>,
 
     /// Additional data related to some connectors
     #[schema(value_type = Option<ConnectorMetadata>)]
diff --git a/crates/diesel_models/src/payment_intent.rs b/crates/diesel_models/src/payment_intent.rs
index bc480c497ea..7f4aaeef29e 100644
--- a/crates/diesel_models/src/payment_intent.rs
+++ b/crates/diesel_models/src/payment_intent.rs
@@ -18,7 +18,7 @@ pub struct PaymentIntent {
     pub customer_id: Option<id_type::CustomerId>,
     pub description: Option<String>,
     pub return_url: Option<String>,
-    pub metadata: Option<pii::SecretSerdeValue>,
+    pub metadata: Option<serde_json::Value>,
     pub connector_id: Option<String>,
     pub shipping_address_id: Option<String>,
     pub billing_address_id: Option<String>,
@@ -77,7 +77,7 @@ pub struct PaymentIntentNew {
     pub customer_id: Option<id_type::CustomerId>,
     pub description: Option<String>,
     pub return_url: Option<String>,
-    pub metadata: Option<pii::SecretSerdeValue>,
+    pub metadata: Option<serde_json::Value>,
     pub connector_id: Option<String>,
     pub shipping_address_id: Option<String>,
     pub billing_address_id: Option<String>,
@@ -131,7 +131,7 @@ pub enum PaymentIntentUpdate {
         incremental_authorization_allowed: Option<bool>,
     },
     MetadataUpdate {
-        metadata: pii::SecretSerdeValue,
+        metadata: serde_json::Value,
         updated_by: String,
     },
     PaymentCreateUpdate {
@@ -169,7 +169,7 @@ pub enum PaymentIntentUpdate {
         statement_descriptor_name: Option<String>,
         statement_descriptor_suffix: Option<String>,
         order_details: Option<Vec<pii::SecretSerdeValue>>,
-        metadata: Option<pii::SecretSerdeValue>,
+        metadata: Option<serde_json::Value>,
         payment_confirm_source: Option<storage_enums::PaymentSource>,
         updated_by: String,
         session_expiry: Option<PrimitiveDateTime>,
@@ -230,7 +230,7 @@ pub struct PaymentIntentUpdateInternal {
     pub return_url: Option<String>,
     pub setup_future_usage: Option<storage_enums::FutureUsage>,
     pub off_session: Option<bool>,
-    pub metadata: Option<pii::SecretSerdeValue>,
+    pub metadata: Option<serde_json::Value>,
     pub billing_address_id: Option<String>,
     pub shipping_address_id: Option<String>,
     pub modified_at: Option<PrimitiveDateTime>,
diff --git a/crates/hyperswitch_domain_models/src/payments.rs b/crates/hyperswitch_domain_models/src/payments.rs
index 0c22c70bc36..fff5043ad95 100644
--- a/crates/hyperswitch_domain_models/src/payments.rs
+++ b/crates/hyperswitch_domain_models/src/payments.rs
@@ -21,7 +21,7 @@ pub struct PaymentIntent {
     pub customer_id: Option<id_type::CustomerId>,
     pub description: Option<String>,
     pub return_url: Option<String>,
-    pub metadata: Option<pii::SecretSerdeValue>,
+    pub metadata: Option<serde_json::Value>,
     pub connector_id: Option<String>,
     pub shipping_address_id: Option<String>,
     pub billing_address_id: Option<String>,
diff --git a/crates/hyperswitch_domain_models/src/payments/payment_intent.rs b/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
index fb824a80798..48e0c63acfa 100644
--- a/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
+++ b/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
@@ -98,7 +98,7 @@ pub struct PaymentIntentNew {
     pub customer_id: Option<id_type::CustomerId>,
     pub description: Option<String>,
     pub return_url: Option<String>,
-    pub metadata: Option<pii::SecretSerdeValue>,
+    pub metadata: Option<serde_json::Value>,
     pub frm_metadata: Option<pii::SecretSerdeValue>,
     pub connector_id: Option<String>,
     pub shipping_address_id: Option<String>,
@@ -147,7 +147,7 @@ pub enum PaymentIntentUpdate {
         incremental_authorization_allowed: Option<bool>,
     },
     MetadataUpdate {
-        metadata: pii::SecretSerdeValue,
+        metadata: serde_json::Value,
         updated_by: String,
     },
     PaymentCreateUpdate {
@@ -185,7 +185,7 @@ pub enum PaymentIntentUpdate {
         statement_descriptor_name: Option<String>,
         statement_descriptor_suffix: Option<String>,
         order_details: Option<Vec<pii::SecretSerdeValue>>,
-        metadata: Option<pii::SecretSerdeValue>,
+        metadata: Option<serde_json::Value>,
         frm_metadata: Option<pii::SecretSerdeValue>,
         payment_confirm_source: Option<storage_enums::PaymentSource>,
         updated_by: String,
@@ -245,7 +245,7 @@ pub struct PaymentIntentUpdateInternal {
     pub return_url: Option<String>,
     pub setup_future_usage: Option<storage_enums::FutureUsage>,
     pub off_session: Option<bool>,
-    pub metadata: Option<pii::SecretSerdeValue>,
+    pub metadata: Option<serde_json::Value>,
     pub billing_address_id: Option<String>,
     pub shipping_address_id: Option<String>,
     pub modified_at: Option<PrimitiveDateTime>,
diff --git a/crates/hyperswitch_domain_models/src/router_request_types.rs b/crates/hyperswitch_domain_models/src/router_request_types.rs
index efd01397f69..7121ebefc6d 100644
--- a/crates/hyperswitch_domain_models/src/router_request_types.rs
+++ b/crates/hyperswitch_domain_models/src/router_request_types.rs
@@ -56,7 +56,7 @@ pub struct PaymentsAuthorizeData {
     pub surcharge_details: Option<SurchargeDetails>,
     pub customer_id: Option<String>,
     pub request_incremental_authorization: bool,
-    pub metadata: Option<pii::SecretSerdeValue>,
+    pub metadata: Option<serde_json::Value>,
     pub authentication_data: Option<AuthenticationData>,
     pub charges: Option<PaymentCharges>,
 
@@ -102,7 +102,7 @@ pub struct PaymentsCaptureData {
     pub multiple_capture_data: Option<MultipleCaptureRequestData>,
     pub connector_meta: Option<serde_json::Value>,
     pub browser_info: Option<BrowserInformation>,
-    pub metadata: Option<pii::SecretSerdeValue>,
+    pub metadata: Option<serde_json::Value>,
     // This metadata is used to store the metadata shared during the payment intent request.
 
     // New amount for amount frame work
@@ -346,7 +346,7 @@ pub struct CompleteAuthorizeData {
     pub connector_transaction_id: Option<String>,
     pub connector_meta: Option<serde_json::Value>,
     pub complete_authorize_url: Option<String>,
-    pub metadata: Option<pii::SecretSerdeValue>,
+    pub metadata: Option<serde_json::Value>,
     pub customer_acceptance: Option<mandates::CustomerAcceptance>,
     // New amount for amount frame work
     pub minor_amount: MinorUnit,
@@ -390,7 +390,7 @@ pub struct PaymentsCancelData {
     pub cancellation_reason: Option<String>,
     pub connector_meta: Option<serde_json::Value>,
     pub browser_info: Option<BrowserInformation>,
-    pub metadata: Option<pii::SecretSerdeValue>,
+    pub metadata: Option<serde_json::Value>,
     // This metadata is used to store the metadata shared during the payment intent request.
 
     // minor amount data for amount framework
diff --git a/crates/router/src/compatibility/stripe/payment_intents/types.rs b/crates/router/src/compatibility/stripe/payment_intents/types.rs
index 6b3054e7c38..81dbbb748e3 100644
--- a/crates/router/src/compatibility/stripe/payment_intents/types.rs
+++ b/crates/router/src/compatibility/stripe/payment_intents/types.rs
@@ -261,7 +261,7 @@ pub struct StripePaymentIntentRequest {
     pub shipping: Option<Shipping>,
     pub statement_descriptor: Option<String>,
     pub statement_descriptor_suffix: Option<String>,
-    pub metadata: Option<SecretSerdeValue>,
+    pub metadata: Option<serde_json::Value>,
     pub client_secret: Option<masking::Secret<String>>,
     pub payment_method_options: Option<StripePaymentMethodOptions>,
     pub merchant_connector_details: Option<admin::MerchantConnectorDetailsWrap>,
@@ -478,7 +478,7 @@ pub struct StripePaymentIntentResponse {
     pub customer: Option<id_type::CustomerId>,
     pub refunds: Option<Vec<stripe_refunds::StripeRefundResponse>>,
     pub mandate: Option<String>,
-    pub metadata: Option<SecretSerdeValue>,
+    pub metadata: Option<serde_json::Value>,
     pub charges: Charges,
     pub connector: Option<String>,
     pub description: Option<String>,
diff --git a/crates/router/src/compatibility/stripe/setup_intents/types.rs b/crates/router/src/compatibility/stripe/setup_intents/types.rs
index 2f01da29832..db5ada51529 100644
--- a/crates/router/src/compatibility/stripe/setup_intents/types.rs
+++ b/crates/router/src/compatibility/stripe/setup_intents/types.rs
@@ -458,7 +458,7 @@ pub struct StripeSetupIntentResponse {
     pub object: String,
     pub status: StripeSetupStatus,
     pub client_secret: Option<masking::Secret<String>>,
-    pub metadata: Option<secret::SecretSerdeValue>,
+    pub metadata: Option<Value>,
     #[serde(with = "common_utils::custom_serde::iso8601::option")]
     pub created: Option<time::PrimitiveDateTime>,
     pub customer: Option<id_type::CustomerId>,
diff --git a/crates/router/src/connector/adyen/transformers.rs b/crates/router/src/connector/adyen/transformers.rs
index 9a85522396d..63c7bed5d0d 100644
--- a/crates/router/src/connector/adyen/transformers.rs
+++ b/crates/router/src/connector/adyen/transformers.rs
@@ -2610,7 +2610,7 @@ impl<'a>
             channel: None,
             shopper_statement: item.router_data.request.statement_descriptor.clone(),
             shopper_ip: item.router_data.request.get_ip_address_as_optional(),
-            metadata: item.router_data.request.metadata.clone(),
+            metadata: item.router_data.request.metadata.clone().map(Into::into),
             merchant_order_reference: item.router_data.request.merchant_order_reference_id.clone(),
         })
     }
@@ -2673,7 +2673,7 @@ impl<'a>
             channel: None,
             shopper_statement: item.router_data.request.statement_descriptor.clone(),
             shopper_ip: item.router_data.request.get_ip_address_as_optional(),
-            metadata: item.router_data.request.metadata.clone(),
+            metadata: item.router_data.request.metadata.clone().map(Into::into),
             merchant_order_reference: item.router_data.request.merchant_order_reference_id.clone(),
         })
     }
@@ -2727,7 +2727,7 @@ impl<'a>
             channel: None,
             shopper_statement: item.router_data.request.statement_descriptor.clone(),
             shopper_ip: item.router_data.request.get_ip_address_as_optional(),
-            metadata: item.router_data.request.metadata.clone(),
+            metadata: item.router_data.request.metadata.clone().map(Into::into),
             merchant_order_reference: item.router_data.request.merchant_order_reference_id.clone(),
         };
         Ok(request)
@@ -2782,7 +2782,7 @@ impl<'a>
             channel: None,
             shopper_statement: item.router_data.request.statement_descriptor.clone(),
             shopper_ip: item.router_data.request.get_ip_address_as_optional(),
-            metadata: item.router_data.request.metadata.clone(),
+            metadata: item.router_data.request.metadata.clone().map(Into::into),
             merchant_order_reference: item.router_data.request.merchant_order_reference_id.clone(),
         };
         Ok(request)
@@ -2833,7 +2833,7 @@ impl<'a>
             channel: None,
             shopper_statement: item.router_data.request.statement_descriptor.clone(),
             shopper_ip: item.router_data.request.get_ip_address_as_optional(),
-            metadata: item.router_data.request.metadata.clone(),
+            metadata: item.router_data.request.metadata.clone().map(Into::into),
             merchant_order_reference: item.router_data.request.merchant_order_reference_id.clone(),
         };
         Ok(request)
@@ -2884,7 +2884,7 @@ impl<'a>
             social_security_number: None,
             shopper_statement: item.router_data.request.statement_descriptor.clone(),
             shopper_ip: item.router_data.request.get_ip_address_as_optional(),
-            metadata: item.router_data.request.metadata.clone(),
+            metadata: item.router_data.request.metadata.clone().map(Into::into),
             merchant_order_reference: item.router_data.request.merchant_order_reference_id.clone(),
         };
         Ok(request)
@@ -2945,7 +2945,7 @@ impl<'a>
             channel: None,
             shopper_statement: item.router_data.request.statement_descriptor.clone(),
             shopper_ip: item.router_data.request.get_ip_address_as_optional(),
-            metadata: item.router_data.request.metadata.clone(),
+            metadata: item.router_data.request.metadata.clone().map(Into::into),
             merchant_order_reference: item.router_data.request.merchant_order_reference_id.clone(),
         })
     }
@@ -3041,7 +3041,7 @@ impl<'a>
             channel,
             shopper_statement: item.router_data.request.statement_descriptor.clone(),
             shopper_ip: item.router_data.request.get_ip_address_as_optional(),
-            metadata: item.router_data.request.metadata.clone(),
+            metadata: item.router_data.request.metadata.clone().map(Into::into),
             merchant_order_reference: item.router_data.request.merchant_order_reference_id.clone(),
         })
     }
@@ -3117,7 +3117,7 @@ impl<'a>
             channel: None,
             shopper_statement: item.router_data.request.statement_descriptor.clone(),
             shopper_ip: item.router_data.request.get_ip_address_as_optional(),
-            metadata: item.router_data.request.metadata.clone(),
+            metadata: item.router_data.request.metadata.clone().map(Into::into),
             merchant_order_reference: item.router_data.request.merchant_order_reference_id.clone(),
         })
     }
@@ -3176,7 +3176,7 @@ impl<'a>
             social_security_number: None,
             shopper_statement: item.router_data.request.statement_descriptor.clone(),
             shopper_ip: item.router_data.request.get_ip_address_as_optional(),
-            metadata: item.router_data.request.metadata.clone(),
+            metadata: item.router_data.request.metadata.clone().map(Into::into),
             merchant_order_reference: item.router_data.request.merchant_order_reference_id.clone(),
         })
     }
diff --git a/crates/router/src/connector/bankofamerica/transformers.rs b/crates/router/src/connector/bankofamerica/transformers.rs
index 1230843afa9..d26c01e9a84 100644
--- a/crates/router/src/connector/bankofamerica/transformers.rs
+++ b/crates/router/src/connector/bankofamerica/transformers.rs
@@ -941,10 +941,12 @@ impl
             specification_version: three_ds_info.three_ds_data.specification_version,
         });
 
-        let merchant_defined_information =
-            item.router_data.request.metadata.clone().map(|metadata| {
-                Vec::<MerchantDefinedInformation>::foreign_from(metadata.peek().to_owned())
-            });
+        let merchant_defined_information = item
+            .router_data
+            .request
+            .metadata
+            .clone()
+            .map(Vec::<MerchantDefinedInformation>::foreign_from);
 
         Ok(Self {
             processing_information,
@@ -976,10 +978,12 @@ impl
         let payment_information = PaymentInformation::try_from(&ccard)?;
         let processing_information = ProcessingInformation::try_from((item, None, None))?;
         let client_reference_information = ClientReferenceInformation::from(item);
-        let merchant_defined_information =
-            item.router_data.request.metadata.clone().map(|metadata| {
-                Vec::<MerchantDefinedInformation>::foreign_from(metadata.peek().to_owned())
-            });
+        let merchant_defined_information = item
+            .router_data
+            .request
+            .metadata
+            .clone()
+            .map(Vec::<MerchantDefinedInformation>::foreign_from);
 
         Ok(Self {
             processing_information,
@@ -1017,10 +1021,12 @@ impl
         ))?;
         let client_reference_information = ClientReferenceInformation::from(item);
         let payment_information = PaymentInformation::try_from(&apple_pay_data)?;
-        let merchant_defined_information =
-            item.router_data.request.metadata.clone().map(|metadata| {
-                Vec::<MerchantDefinedInformation>::foreign_from(metadata.peek().to_owned())
-            });
+        let merchant_defined_information = item
+            .router_data
+            .request
+            .metadata
+            .clone()
+            .map(Vec::<MerchantDefinedInformation>::foreign_from);
         let ucaf_collection_indicator = match apple_pay_wallet_data
             .payment_method
             .network
@@ -1068,10 +1074,12 @@ impl
         let processing_information =
             ProcessingInformation::try_from((item, Some(PaymentSolution::GooglePay), None))?;
         let client_reference_information = ClientReferenceInformation::from(item);
-        let merchant_defined_information =
-            item.router_data.request.metadata.clone().map(|metadata| {
-                Vec::<MerchantDefinedInformation>::foreign_from(metadata.peek().to_owned())
-            });
+        let merchant_defined_information = item
+            .router_data
+            .request
+            .metadata
+            .clone()
+            .map(Vec::<MerchantDefinedInformation>::foreign_from);
 
         Ok(Self {
             processing_information,
@@ -1132,7 +1140,7 @@ impl TryFrom<&BankOfAmericaRouterData<&types::PaymentsAuthorizeRouterData>>
                                     let merchant_defined_information =
                                         item.router_data.request.metadata.clone().map(|metadata| {
                                             Vec::<MerchantDefinedInformation>::foreign_from(
-                                                metadata.peek().to_owned(),
+                                                metadata,
                                             )
                                         });
                                     let ucaf_collection_indicator = match apple_pay_data
@@ -1307,10 +1315,12 @@ impl
                 payment_instrument,
             }));
         let client_reference_information = ClientReferenceInformation::from(item);
-        let merchant_defined_information =
-            item.router_data.request.metadata.clone().map(|metadata| {
-                Vec::<MerchantDefinedInformation>::foreign_from(metadata.peek().to_owned())
-            });
+        let merchant_defined_information = item
+            .router_data
+            .request
+            .metadata
+            .clone()
+            .map(Vec::<MerchantDefinedInformation>::foreign_from);
         Ok(Self {
             processing_information,
             payment_information,
@@ -2610,10 +2620,12 @@ impl TryFrom<&BankOfAmericaRouterData<&types::PaymentsCaptureRouterData>>
     fn try_from(
         value: &BankOfAmericaRouterData<&types::PaymentsCaptureRouterData>,
     ) -> Result<Self, Self::Error> {
-        let merchant_defined_information =
-            value.router_data.request.metadata.clone().map(|metadata| {
-                Vec::<MerchantDefinedInformation>::foreign_from(metadata.peek().to_owned())
-            });
+        let merchant_defined_information = value
+            .router_data
+            .request
+            .metadata
+            .clone()
+            .map(Vec::<MerchantDefinedInformation>::foreign_from);
         Ok(Self {
             order_information: OrderInformation {
                 amount_details: Amount {
@@ -2653,10 +2665,12 @@ impl TryFrom<&BankOfAmericaRouterData<&types::PaymentsCancelRouterData>>
     fn try_from(
         value: &BankOfAmericaRouterData<&types::PaymentsCancelRouterData>,
     ) -> Result<Self, Self::Error> {
-        let merchant_defined_information =
-            value.router_data.request.metadata.clone().map(|metadata| {
-                Vec::<MerchantDefinedInformation>::foreign_from(metadata.peek().to_owned())
-            });
+        let merchant_defined_information = value
+            .router_data
+            .request
+            .metadata
+            .clone()
+            .map(Vec::<MerchantDefinedInformation>::foreign_from);
         Ok(Self {
             client_reference_information: ClientReferenceInformation {
                 code: Some(value.router_data.connector_request_reference_id.clone()),
diff --git a/crates/router/src/connector/billwerk/transformers.rs b/crates/router/src/connector/billwerk/transformers.rs
index ddaa58cfd04..c00303fc5e9 100644
--- a/crates/router/src/connector/billwerk/transformers.rs
+++ b/crates/router/src/connector/billwerk/transformers.rs
@@ -202,7 +202,7 @@ impl TryFrom<&BillwerkRouterData<&types::PaymentsAuthorizeRouterData>> for Billw
                 first_name: item.router_data.get_optional_billing_first_name(),
                 last_name: item.router_data.get_optional_billing_last_name(),
             },
-            metadata: item.router_data.request.metadata.clone(),
+            metadata: item.router_data.request.metadata.clone().map(Into::into),
             settle: item.router_data.request.is_auto_capture()?,
         })
     }
diff --git a/crates/router/src/connector/bluesnap/transformers.rs b/crates/router/src/connector/bluesnap/transformers.rs
index fbc7c661e4c..d1d5f1d8c9e 100644
--- a/crates/router/src/connector/bluesnap/transformers.rs
+++ b/crates/router/src/connector/bluesnap/transformers.rs
@@ -258,7 +258,7 @@ impl TryFrom<&BluesnapRouterData<&types::PaymentsAuthorizeRouterData>> for Blues
                 .metadata
                 .as_ref()
                 .map(|metadata| BluesnapMetadata {
-                    meta_data: Vec::<RequestMetadata>::foreign_from(metadata.peek().to_owned()),
+                    meta_data: Vec::<RequestMetadata>::foreign_from(metadata.to_owned()),
                 });
 
         let (payment_method, card_holder_info) = match item
@@ -609,7 +609,7 @@ impl TryFrom<&BluesnapRouterData<&types::PaymentsCompleteAuthorizeRouterData>>
                 .metadata
                 .as_ref()
                 .map(|metadata| BluesnapMetadata {
-                    meta_data: Vec::<RequestMetadata>::foreign_from(metadata.peek().to_owned()),
+                    meta_data: Vec::<RequestMetadata>::foreign_from(metadata.to_owned()),
                 });
 
         let token = item
diff --git a/crates/router/src/connector/checkout/transformers.rs b/crates/router/src/connector/checkout/transformers.rs
index 0a73d135f43..e035c102d20 100644
--- a/crates/router/src/connector/checkout/transformers.rs
+++ b/crates/router/src/connector/checkout/transformers.rs
@@ -418,7 +418,7 @@ impl TryFrom<&CheckoutRouterData<&types::PaymentsAuthorizeRouterData>> for Payme
         let connector_auth = &item.router_data.connector_auth_type;
         let auth_type: CheckoutAuthType = connector_auth.try_into()?;
         let processing_channel_id = auth_type.processing_channel_id;
-        let metadata = item.router_data.request.metadata.clone();
+        let metadata = item.router_data.request.metadata.clone().map(Into::into);
         Ok(Self {
             source: source_var,
             amount: item.amount.to_owned(),
diff --git a/crates/router/src/connector/cybersource/transformers.rs b/crates/router/src/connector/cybersource/transformers.rs
index dcaa2ffb2f1..38514fc45f9 100644
--- a/crates/router/src/connector/cybersource/transformers.rs
+++ b/crates/router/src/connector/cybersource/transformers.rs
@@ -991,10 +991,12 @@ impl
 
         let processing_information = ProcessingInformation::try_from((item, None, card_type))?;
         let client_reference_information = ClientReferenceInformation::from(item);
-        let merchant_defined_information =
-            item.router_data.request.metadata.clone().map(|metadata| {
-                Vec::<MerchantDefinedInformation>::foreign_from(metadata.peek().to_owned())
-            });
+        let merchant_defined_information = item
+            .router_data
+            .request
+            .metadata
+            .clone()
+            .map(Vec::<MerchantDefinedInformation>::foreign_from);
 
         let consumer_authentication_information = item
             .router_data
@@ -1097,10 +1099,12 @@ impl
             veres_enrolled: None,
         });
 
-        let merchant_defined_information =
-            item.router_data.request.metadata.clone().map(|metadata| {
-                Vec::<MerchantDefinedInformation>::foreign_from(metadata.peek().to_owned())
-            });
+        let merchant_defined_information = item
+            .router_data
+            .request
+            .metadata
+            .clone()
+            .map(Vec::<MerchantDefinedInformation>::foreign_from);
 
         Ok(Self {
             processing_information,
@@ -1149,10 +1153,12 @@ impl
                     expiration_month,
                 },
             }));
-        let merchant_defined_information =
-            item.router_data.request.metadata.clone().map(|metadata| {
-                Vec::<MerchantDefinedInformation>::foreign_from(metadata.peek().to_owned())
-            });
+        let merchant_defined_information = item
+            .router_data
+            .request
+            .metadata
+            .clone()
+            .map(Vec::<MerchantDefinedInformation>::foreign_from);
         let ucaf_collection_indicator = match apple_pay_wallet_data
             .payment_method
             .network
@@ -1211,10 +1217,12 @@ impl
         let processing_information =
             ProcessingInformation::try_from((item, Some(PaymentSolution::GooglePay), None))?;
         let client_reference_information = ClientReferenceInformation::from(item);
-        let merchant_defined_information =
-            item.router_data.request.metadata.clone().map(|metadata| {
-                Vec::<MerchantDefinedInformation>::foreign_from(metadata.peek().to_owned())
-            });
+        let merchant_defined_information = item
+            .router_data
+            .request
+            .metadata
+            .clone()
+            .map(Vec::<MerchantDefinedInformation>::foreign_from);
 
         Ok(Self {
             processing_information,
@@ -1284,7 +1292,7 @@ impl TryFrom<&CybersourceRouterData<&types::PaymentsAuthorizeRouterData>>
                                     let merchant_defined_information =
                                         item.router_data.request.metadata.clone().map(|metadata| {
                                             Vec::<MerchantDefinedInformation>::foreign_from(
-                                                metadata.peek().to_owned(),
+                                                metadata,
                                             )
                                         });
                                     let ucaf_collection_indicator = match apple_pay_data
@@ -1411,10 +1419,12 @@ impl
                 payment_instrument,
             }));
         let client_reference_information = ClientReferenceInformation::from(item);
-        let merchant_defined_information =
-            item.router_data.request.metadata.clone().map(|metadata| {
-                Vec::<MerchantDefinedInformation>::foreign_from(metadata.peek().to_owned())
-            });
+        let merchant_defined_information = item
+            .router_data
+            .request
+            .metadata
+            .clone()
+            .map(Vec::<MerchantDefinedInformation>::foreign_from);
         Ok(Self {
             processing_information,
             payment_information,
@@ -1510,10 +1520,12 @@ impl TryFrom<&CybersourceRouterData<&types::PaymentsCaptureRouterData>>
     fn try_from(
         item: &CybersourceRouterData<&types::PaymentsCaptureRouterData>,
     ) -> Result<Self, Self::Error> {
-        let merchant_defined_information =
-            item.router_data.request.metadata.clone().map(|metadata| {
-                Vec::<MerchantDefinedInformation>::foreign_from(metadata.peek().to_owned())
-            });
+        let merchant_defined_information = item
+            .router_data
+            .request
+            .metadata
+            .clone()
+            .map(Vec::<MerchantDefinedInformation>::foreign_from);
         Ok(Self {
             processing_information: ProcessingInformation {
                 capture_options: Some(CaptureOptions {
@@ -1602,10 +1614,12 @@ impl TryFrom<&CybersourceRouterData<&types::PaymentsCancelRouterData>> for Cyber
     fn try_from(
         value: &CybersourceRouterData<&types::PaymentsCancelRouterData>,
     ) -> Result<Self, Self::Error> {
-        let merchant_defined_information =
-            value.router_data.request.metadata.clone().map(|metadata| {
-                Vec::<MerchantDefinedInformation>::foreign_from(metadata.peek().to_owned())
-            });
+        let merchant_defined_information = value
+            .router_data
+            .request
+            .metadata
+            .clone()
+            .map(Vec::<MerchantDefinedInformation>::foreign_from);
         Ok(Self {
             client_reference_information: ClientReferenceInformation {
                 code: Some(value.router_data.connector_request_reference_id.clone()),
diff --git a/crates/router/src/connector/nmi/transformers.rs b/crates/router/src/connector/nmi/transformers.rs
index c9d86b17ba8..a87b8e6176d 100644
--- a/crates/router/src/connector/nmi/transformers.rs
+++ b/crates/router/src/connector/nmi/transformers.rs
@@ -1,12 +1,7 @@
 use api_models::webhooks;
 use cards::CardNumber;
 use common_enums::CountryAlpha2;
-use common_utils::{
-    errors::CustomResult,
-    ext_traits::XmlExt,
-    pii::{self, Email},
-    types::FloatMajorUnit,
-};
+use common_utils::{errors::CustomResult, ext_traits::XmlExt, pii::Email, types::FloatMajorUnit};
 use error_stack::{report, Report, ResultExt};
 use masking::{ExposeInterface, PeekInterface, Secret};
 use serde::{Deserialize, Serialize};
@@ -431,8 +426,8 @@ pub struct NmiMerchantDefinedField {
 }
 
 impl NmiMerchantDefinedField {
-    pub fn new(metadata: &pii::SecretSerdeValue) -> Self {
-        let metadata_as_string = metadata.peek().to_string();
+    pub fn new(metadata: &serde_json::Value) -> Self {
+        let metadata_as_string = metadata.to_string();
         let hash_map: std::collections::BTreeMap<String, serde_json::Value> =
             serde_json::from_str(&metadata_as_string).unwrap_or(std::collections::BTreeMap::new());
         let inner = hash_map
diff --git a/crates/router/src/connector/noon/transformers.rs b/crates/router/src/connector/noon/transformers.rs
index d79774b2e87..6de5ba5e6d1 100644
--- a/crates/router/src/connector/noon/transformers.rs
+++ b/crates/router/src/connector/noon/transformers.rs
@@ -1,6 +1,6 @@
 use common_utils::{ext_traits::Encode, pii, types::StringMajorUnit};
 use error_stack::ResultExt;
-use masking::{ExposeInterface, PeekInterface, Secret};
+use masking::{ExposeInterface, Secret};
 use serde::{Deserialize, Serialize};
 
 use crate::{
@@ -107,8 +107,8 @@ fn get_value_as_string(value: &serde_json::Value) -> String {
 }
 
 impl NoonOrderNvp {
-    pub fn new(metadata: &pii::SecretSerdeValue) -> Self {
-        let metadata_as_string = metadata.peek().to_string();
+    pub fn new(metadata: &serde_json::Value) -> Self {
+        let metadata_as_string = metadata.to_string();
         let hash_map: std::collections::BTreeMap<String, serde_json::Value> =
             serde_json::from_str(&metadata_as_string).unwrap_or(std::collections::BTreeMap::new());
         let inner = hash_map
diff --git a/crates/router/src/connector/stripe/transformers.rs b/crates/router/src/connector/stripe/transformers.rs
index 53850e97afc..ddd56846741 100644
--- a/crates/router/src/connector/stripe/transformers.rs
+++ b/crates/router/src/connector/stripe/transformers.rs
@@ -1840,7 +1840,8 @@ impl TryFrom<(&types::PaymentsAuthorizeRouterData, MinorUnit)> for PaymentIntent
                 }
             });
 
-        let meta_data = get_transaction_metadata(item.request.metadata.clone(), order_id);
+        let meta_data =
+            get_transaction_metadata(item.request.metadata.clone().map(Into::into), order_id);
 
         // We pass browser_info only when payment_data exists.
         // Hence, we're pass Null during recurring payments as payment_method_data[type] is not passed
@@ -3334,7 +3335,7 @@ impl TryFrom<(&types::PaymentsAuthorizeRouterData, MinorUnit)> for ChargesReques
             let amount = data.1;
             let order_id = value.connector_request_reference_id.clone();
             let meta_data = Some(get_transaction_metadata(
-                value.request.metadata.clone(),
+                value.request.metadata.clone().map(Into::into),
                 order_id,
             ));
             Ok(Self {
diff --git a/crates/router/src/connector/utils.rs b/crates/router/src/connector/utils.rs
index 67ba0219642..e9657c6cfc5 100644
--- a/crates/router/src/connector/utils.rs
+++ b/crates/router/src/connector/utils.rs
@@ -922,16 +922,14 @@ impl PaymentsAuthorizeRequestData for types::PaymentsAuthorizeData {
     }
 
     fn get_metadata_as_object(&self) -> Option<pii::SecretSerdeValue> {
-        self.metadata
-            .clone()
-            .and_then(|meta_data| match meta_data.peek() {
-                serde_json::Value::Null
-                | serde_json::Value::Bool(_)
-                | serde_json::Value::Number(_)
-                | serde_json::Value::String(_)
-                | serde_json::Value::Array(_) => None,
-                serde_json::Value::Object(_) => Some(meta_data),
-            })
+        self.metadata.clone().and_then(|meta_data| match meta_data {
+            serde_json::Value::Null
+            | serde_json::Value::Bool(_)
+            | serde_json::Value::Number(_)
+            | serde_json::Value::String(_)
+            | serde_json::Value::Array(_) => None,
+            serde_json::Value::Object(_) => Some(meta_data.into()),
+        })
     }
 
     fn get_authentication_data(&self) -> Result<AuthenticationData, Error> {
diff --git a/crates/router/src/core/fraud_check.rs b/crates/router/src/core/fraud_check.rs
index 69c82d4f2d7..0d0eace13e3 100644
--- a/crates/router/src/core/fraud_check.rs
+++ b/crates/router/src/core/fraud_check.rs
@@ -320,7 +320,7 @@ where
         .or_else(||
             // when the order_details are present within the meta_data, we need to take those to support backward compatibility
             payment_data.payment_intent.metadata.clone().and_then(|meta| {
-                let order_details = meta.peek().get("order_details").to_owned();
+                let order_details = meta.get("order_details").to_owned();
                 order_details.map(|order| vec![masking::Secret::new(order.to_owned())])
             }))
         .map(|order_details_value| {
diff --git a/crates/router/src/core/payments/transformers.rs b/crates/router/src/core/payments/transformers.rs
index 62167d592e9..b9b3aa134dd 100644
--- a/crates/router/src/core/payments/transformers.rs
+++ b/crates/router/src/core/payments/transformers.rs
@@ -1690,7 +1690,7 @@ impl<F: Clone> TryFrom<PaymentAdditionalData<'_, F>> for types::SetupMandateRequ
                 Some(RequestIncrementalAuthorization::True)
                     | Some(RequestIncrementalAuthorization::Default)
             ),
-            metadata: payment_data.payment_intent.metadata.clone(),
+            metadata: payment_data.payment_intent.metadata.clone().map(Into::into),
         })
     }
 }
diff --git a/crates/router/src/services/kafka/payment_intent.rs b/crates/router/src/services/kafka/payment_intent.rs
index 7a74f07e3b2..8491ddbb91e 100644
--- a/crates/router/src/services/kafka/payment_intent.rs
+++ b/crates/router/src/services/kafka/payment_intent.rs
@@ -14,6 +14,7 @@ pub struct KafkaPaymentIntent<'a> {
     pub customer_id: Option<&'a id_type::CustomerId>,
     pub description: Option<&'a String>,
     pub return_url: Option<&'a String>,
+    pub metadata: Option<String>,
     pub connector_id: Option<&'a String>,
     pub statement_descriptor_name: Option<&'a String>,
     pub statement_descriptor_suffix: Option<&'a String>,
@@ -45,6 +46,7 @@ impl<'a> KafkaPaymentIntent<'a> {
             customer_id: intent.customer_id.as_ref(),
             description: intent.description.as_ref(),
             return_url: intent.return_url.as_ref(),
+            metadata: intent.metadata.as_ref().map(|x| x.to_string()),
             connector_id: intent.connector_id.as_ref(),
             statement_descriptor_name: intent.statement_descriptor_name.as_ref(),
             statement_descriptor_suffix: intent.statement_descriptor_suffix.as_ref(),
diff --git a/crates/router/src/services/kafka/payment_intent_event.rs b/crates/router/src/services/kafka/payment_intent_event.rs
index a3fbd9ddc45..dccd301151a 100644
--- a/crates/router/src/services/kafka/payment_intent_event.rs
+++ b/crates/router/src/services/kafka/payment_intent_event.rs
@@ -15,6 +15,7 @@ pub struct KafkaPaymentIntentEvent<'a> {
     pub customer_id: Option<&'a id_type::CustomerId>,
     pub description: Option<&'a String>,
     pub return_url: Option<&'a String>,
+    pub metadata: Option<String>,
     pub connector_id: Option<&'a String>,
     pub statement_descriptor_name: Option<&'a String>,
     pub statement_descriptor_suffix: Option<&'a String>,
@@ -46,6 +47,7 @@ impl<'a> KafkaPaymentIntentEvent<'a> {
             customer_id: intent.customer_id.as_ref(),
             description: intent.description.as_ref(),
             return_url: intent.return_url.as_ref(),
+            metadata: intent.metadata.as_ref().map(|x| x.to_string()),
             connector_id: intent.connector_id.as_ref(),
             statement_descriptor_name: intent.statement_descriptor_name.as_ref(),
             statement_descriptor_suffix: intent.statement_descriptor_suffix.as_ref(),
diff --git a/docker-compose.yml b/docker-compose.yml
index 3a6ff1e4047..9c8aa61dc90 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -379,6 +379,7 @@ services:
     environment:
       - "discovery.type=single-node"
       - OPENSEARCH_INITIAL_ADMIN_PASSWORD=0penS3arc#
+      - LOG_LEVEL=DEBUG
     profiles:
       - olap
     ports:
diff --git a/docker/fluentd/conf/fluent.conf b/docker/fluentd/conf/fluent.conf
index 7aec9dd7cd8..785254a3b25 100644
--- a/docker/fluentd/conf/fluent.conf
+++ b/docker/fluentd/conf/fluent.conf
@@ -125,7 +125,7 @@
     index_name hyperswitch-dispute-events
     id_key dispute_id
     user admin
-    password admin
+    password '0penS3arc#'
     ssl_verify false
     prefer_oj_serializer true
     reload_on_failure true