Issue:
Bug:  feat(events): add payment feature metadata & customer email to the payment intent events



Add a hashed value of these fields and use them to provide global search

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 -->
- Implemented hashing of `customer_email` using `blake3::hash`, and added this to PaymentIntent kafka events
- Enabled search using `customer_email` filter in dashboard globalsearch as well
- Added `feature_metadata` with `search_tags` for dashboard globalsearch
### 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).
-->
Hashing of customer_email helps in masking of data to help protect privacy

## 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)?
-->
Tested it using Postman curls:
- Made a payment with the `customer_email` and `search_tags` in `feature_metadata` set
```bash
curl --location 'http://localhost:8080/payments' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: dev_IG9ateOUhGE0KLFEItNN1tsJXcr5M62N7zlbtOlUtStTfPpzSomswbQa4aO2k1B4' \
--data-raw '{
  "amount": 6540,
  "currency": "INR",
  "confirm": true,
  "capture_method": "automatic",
  "capture_on": "2022-09-10T10:11:12Z",
  "amount_to_capture": 6540,
  "customer_id": "StripeCustomer",
  "email": "test@hyperswitch.com",
  "name": "John Doe",
  "phone": "999999999",
  "phone_country_code": "+1",
  "description": "Its my first payment request",
  "authentication_type": "no_three_ds",
  "return_url": "https://google.com",
  "payment_method": "card",
  "payment_method_type": "credit",
  "payment_method_data": {
    "card": {
      "card_number": "4242424242424242",
      "card_exp_month": "10",
      "card_exp_year": "25",
      "card_holder_name": "joseph Doe",
      "card_cvc": "123"
    }
  },
  "billing": {
    "address": {
      "line1": "1467",
      "line2": "Harrison Street",
      "line3": "Harrison Street",
      "city": "San Fransico",
      "state": "California",
      "zip": "94122",
      "country": "US",
      "first_name": "joseph",
      "last_name": "Doe"
    },
    "phone": {
      "number": "8056594427",
      "country_code": "+91"
    }
  },
  "shipping": {
    "address": {
      "line1": "1467",
      "line2": "Harrison Street",
      "line3": "Harrison Street",
      "city": "San Fransico",
      "state": "California",
      "zip": "94122",
      "country": "US",
      "first_name": "joseph",
      "last_name": "Doe"
    },
    "phone": {
      "number": "8056594427",
      "country_code": "+91"
    }
  },
  "statement_descriptor_name": "joseph",
  "statement_descriptor_suffix": "JS",
  "feature_metadata": {
    "search_tags": ["dubai", "tokyo"]
  },
  "metadata": {
    "data2": "camel",
    "new_customer": "true",
    "login_date": "2019-09-10T10:11:12Z"
  }
}'
```
- Hit the `analytics/v1/search` with the following curl:
```bash
curl --location 'http://localhost:8080/analytics/v1/search' \
--header 'Accept: */*' \
--header 'Accept-Language: en-US,en;q=0.9' \
--header 'Connection: keep-alive' \
--header 'Content-Type: application/json' \
--header 'Origin: http://localhost:9000' \
--header 'Referer: http://localhost:9000/' \
--header 'Sec-Fetch-Dest: empty' \
--header 'Sec-Fetch-Mode: cors' \
--header 'Sec-Fetch-Site: same-site' \
--header 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36' \
--header 'api-key: hyperswitch' \
--header 'authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiNGU2ZWVhYzEtMWVjNi00ZWQyLWJhY2YtMDUzOTFiZGE3YjI1IiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNzE5OTA0MzA1Iiwicm9sZV9pZCI6Im9yZ19hZG1pbiIsImV4cCI6MTcyMDI3MTg0NSwib3JnX2lkIjoib3JnX0c4Zk85a09UTXl1OHBXUEVQSUcwIn0.2Pdjr1w82ImWbccLcAthHzJ8LU959xSHRZ5PrJCFtoE' \
--header 'sec-ch-ua: "Not/A)Brand";v="8", "Chromium";v="126", "Google Chrome";v="126"' \
--header 'sec-ch-ua-mobile: ?0' \
--header 'sec-ch-ua-platform: "macOS"' \
--data-raw '{
    "query": "merchant_1719904305",
    "filters": {
        "currency": [
            "INR"
        ],
        "customer_email": [
            "test@hyperswitch.com"
        ]
    }
}'
```
You should be able to get the results in the `payment_intents` index
## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible


Testing:

Tested it using Postman curls:
- Made a payment with the `customer_email` and `search_tags` in `feature_metadata` set
```bash
curl --location 'http://localhost:8080/payments' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: dev_IG9ateOUhGE0KLFEItNN1tsJXcr5M62N7zlbtOlUtStTfPpzSomswbQa4aO2k1B4' \
--data-raw '{
  "amount": 6540,
  "currency": "INR",
  "confirm": true,
  "capture_method": "automatic",
  "capture_on": "2022-09-10T10:11:12Z",
  "amount_to_capture": 6540,
  "customer_id": "StripeCustomer",
  "email": "test@hyperswitch.com",
  "name": "John Doe",
  "phone": "999999999",
  "phone_country_code": "+1",
  "description": "Its my first payment request",
  "authentication_type": "no_three_ds",
  "return_url": "https://google.com",
  "payment_method": "card",
  "payment_method_type": "credit",
  "payment_method_data": {
    "card": {
      "card_number": "4242424242424242",
      "card_exp_month": "10",
      "card_exp_year": "25",
      "card_holder_name": "joseph Doe",
      "card_cvc": "123"
    }
  },
  "billing": {
    "address": {
      "line1": "1467",
      "line2": "Harrison Street",
      "line3": "Harrison Street",
      "city": "San Fransico",
      "state": "California",
      "zip": "94122",
      "country": "US",
      "first_name": "joseph",
      "last_name": "Doe"
    },
    "phone": {
      "number": "8056594427",
      "country_code": "+91"
    }
  },
  "shipping": {
    "address": {
      "line1": "1467",
      "line2": "Harrison Street",
      "line3": "Harrison Street",
      "city": "San Fransico",
      "state": "California",
      "zip": "94122",
      "country": "US",
      "first_name": "joseph",
      "last_name": "Doe"
    },
    "phone": {
      "number": "8056594427",
      "country_code": "+91"
    }
  },
  "statement_descriptor_name": "joseph",
  "statement_descriptor_suffix": "JS",
  "feature_metadata": {
    "search_tags": ["dubai", "tokyo"]
  },
  "metadata": {
    "data2": "camel",
    "new_customer": "true",
    "login_date": "2019-09-10T10:11:12Z"
  }
}'
```
- Hit the `analytics/v1/search` with the following curl:
```bash
curl --location 'http://localhost:8080/analytics/v1/search' \
--header 'Accept: */*' \
--header 'Accept-Language: en-US,en;q=0.9' \
--header 'Connection: keep-alive' \
--header 'Content-Type: application/json' \
--header 'Origin: http://localhost:9000' \
--header 'Referer: http://localhost:9000/' \
--header 'Sec-Fetch-Dest: empty' \
--header 'Sec-Fetch-Mode: cors' \
--header 'Sec-Fetch-Site: same-site' \
--header 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36' \
--header 'api-key: hyperswitch' \
--header 'authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiNGU2ZWVhYzEtMWVjNi00ZWQyLWJhY2YtMDUzOTFiZGE3YjI1IiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNzE5OTA0MzA1Iiwicm9sZV9pZCI6Im9yZ19hZG1pbiIsImV4cCI6MTcyMDI3MTg0NSwib3JnX2lkIjoib3JnX0c4Zk85a09UTXl1OHBXUEVQSUcwIn0.2Pdjr1w82ImWbccLcAthHzJ8LU959xSHRZ5PrJCFtoE' \
--header 'sec-ch-ua: "Not/A)Brand";v="8", "Chromium";v="126", "Google Chrome";v="126"' \
--header 'sec-ch-ua-mobile: ?0' \
--header 'sec-ch-ua-platform: "macOS"' \
--data-raw '{
    "query": "merchant_1719904305",
    "filters": {
        "currency": [
            "INR"
        ],
        "customer_email": [
            "test@hyperswitch.com"
        ]
    }
}'
```
You should be able to get the results in the `payment_intents` index

Patch:
diff --git a/Cargo.lock b/Cargo.lock
index d0b9894970b..d40d7e2ee05 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1577,6 +1577,7 @@ dependencies = [
  "cc",
  "cfg-if 1.0.0",
  "constant_time_eq 0.3.0",
+ "serde",
 ]
 
 [[package]]
@@ -1958,6 +1959,7 @@ name = "common_utils"
 version = "0.1.0"
 dependencies = [
  "async-trait",
+ "blake3",
  "bytes 1.6.0",
  "common_enums",
  "diesel",
diff --git a/config/deployments/env_specific.toml b/config/deployments/env_specific.toml
index 7d32e76928b..67f32396b74 100644
--- a/config/deployments/env_specific.toml
+++ b/config/deployments/env_specific.toml
@@ -82,6 +82,7 @@ audit_events_topic = "topic"               # Kafka topic to be used for Payment
 payout_analytics_topic = "topic"           # Kafka topic to be used for Payouts and PayoutAttempt events
 consolidated_events_topic = "topic"        # Kafka topic to be used for Consolidated events
 authentication_analytics_topic = "topic"   # Kafka topic to be used for Authentication events
+fraud_check_analytics_topic = "topic"      # Kafka topic to be used for Fraud Check events
 
 # File storage configuration
 [file_storage]
diff --git a/crates/analytics/src/search.rs b/crates/analytics/src/search.rs
index 1237db061bd..269864bf44a 100644
--- a/crates/analytics/src/search.rs
+++ b/crates/analytics/src/search.rs
@@ -51,7 +51,18 @@ pub async fn msearch_results(
         if let Some(customer_email) = filters.customer_email {
             if !customer_email.is_empty() {
                 query_builder
-                    .add_filter_clause("customer_email.keyword".to_string(), customer_email.clone())
+                    .add_filter_clause(
+                        "customer_email.keyword".to_string(),
+                        customer_email
+                            .iter()
+                            .filter_map(|email| {
+                                // TODO: Add trait based inputs instead of converting this to strings
+                                serde_json::to_value(email)
+                                    .ok()
+                                    .and_then(|a| a.as_str().map(|a| a.to_string()))
+                            })
+                            .collect(),
+                    )
                     .switch()?;
             }
         };
@@ -147,7 +158,18 @@ pub async fn search_results(
         if let Some(customer_email) = filters.customer_email {
             if !customer_email.is_empty() {
                 query_builder
-                    .add_filter_clause("customer_email.keyword".to_string(), customer_email.clone())
+                    .add_filter_clause(
+                        "customer_email.keyword".to_string(),
+                        customer_email
+                            .iter()
+                            .filter_map(|email| {
+                                // TODO: Add trait based inputs instead of converting this to strings
+                                serde_json::to_value(email)
+                                    .ok()
+                                    .and_then(|a| a.as_str().map(|a| a.to_string()))
+                            })
+                            .collect(),
+                    )
                     .switch()?;
             }
         };
diff --git a/crates/api_models/src/analytics/search.rs b/crates/api_models/src/analytics/search.rs
index b2af4f6759c..034a2a94356 100644
--- a/crates/api_models/src/analytics/search.rs
+++ b/crates/api_models/src/analytics/search.rs
@@ -1,3 +1,4 @@
+use common_utils::hashing::HashedString;
 use serde_json::Value;
 
 #[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)]
@@ -5,7 +6,7 @@ pub struct SearchFilters {
     pub payment_method: Option<Vec<String>>,
     pub currency: Option<Vec<String>>,
     pub status: Option<Vec<String>>,
-    pub customer_email: Option<Vec<String>>,
+    pub customer_email: Option<Vec<HashedString<common_utils::pii::EmailStrategy>>>,
 }
 
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
diff --git a/crates/api_models/src/payments.rs b/crates/api_models/src/payments.rs
index c0b9ce756f7..311686f27de 100644
--- a/crates/api_models/src/payments.rs
+++ b/crates/api_models/src/payments.rs
@@ -4821,8 +4821,11 @@ pub struct PaymentsStartRequest {
 #[derive(Debug, Clone, serde::Deserialize, serde::Serialize, ToSchema)]
 pub struct FeatureMetadata {
     /// Redirection response coming in request as metadata field only for redirection scenarios
-    #[schema(value_type = Option<RedirectResponse>)]
     pub redirect_response: Option<RedirectResponse>,
+    // TODO: Convert this to hashedstrings to avoid PII sensitive data
+    /// Additional tags to be used for global search
+    #[schema(value_type = Option<RedirectResponse>)]
+    pub search_tags: Option<Vec<Secret<String>>>,
 }
 
 ///frm message is an object sent inside the payments response...when frm is invoked, its value is Some(...), else its None
diff --git a/crates/common_utils/Cargo.toml b/crates/common_utils/Cargo.toml
index 58cdf447fd4..568dc00a970 100644
--- a/crates/common_utils/Cargo.toml
+++ b/crates/common_utils/Cargo.toml
@@ -44,6 +44,8 @@ tokio = { version = "1.37.0", features = ["macros", "rt-multi-thread"], optional
 url = { version = "2.5.0", features = ["serde"] }
 utoipa = { version = "4.2.0", features = ["preserve_order", "preserve_path_order"] }
 uuid = { version = "1.8.0", features = ["v7"] }
+blake3 = { version = "1.5.1", features = ["serde"] }
+
 
 # First party crates
 rusty-money = { git = "https://github.com/varunsrin/rusty_money", rev = "bbc0150742a0fff905225ff11ee09388e9babdcc", features = ["iso", "crypto"] }
diff --git a/crates/common_utils/src/hashing.rs b/crates/common_utils/src/hashing.rs
new file mode 100644
index 00000000000..d08cd9f0868
--- /dev/null
+++ b/crates/common_utils/src/hashing.rs
@@ -0,0 +1,22 @@
+use masking::{PeekInterface, Secret, Strategy};
+use serde::{Deserialize, Serialize, Serializer};
+
+#[derive(Clone, Debug, Deserialize)]
+/// Represents a hashed string using blake3's hashing strategy.
+pub struct HashedString<T: Strategy<String>>(Secret<String, T>);
+
+impl<T: Strategy<String>> Serialize for HashedString<T> {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        let hashed_value = blake3::hash(self.0.peek().as_bytes()).to_hex();
+        hashed_value.serialize(serializer)
+    }
+}
+
+impl<T: Strategy<String>> From<Secret<String, T>> for HashedString<T> {
+    fn from(value: Secret<String, T>) -> Self {
+        Self(value)
+    }
+}
diff --git a/crates/common_utils/src/lib.rs b/crates/common_utils/src/lib.rs
index 8e305dd3d67..d1449f2a7d2 100644
--- a/crates/common_utils/src/lib.rs
+++ b/crates/common_utils/src/lib.rs
@@ -31,6 +31,8 @@ pub mod static_cache;
 pub mod types;
 pub mod validation;
 
+/// Used for hashing
+pub mod hashing;
 #[cfg(feature = "metrics")]
 pub mod metrics;
 
diff --git a/crates/common_utils/src/pii.rs b/crates/common_utils/src/pii.rs
index 2921b237177..9d4b96200ba 100644
--- a/crates/common_utils/src/pii.rs
+++ b/crates/common_utils/src/pii.rs
@@ -14,6 +14,7 @@ use error_stack::ResultExt;
 use masking::{ExposeInterface, Secret, Strategy, WithType};
 #[cfg(feature = "logs")]
 use router_env::logger;
+use serde::Deserialize;
 
 use crate::{
     crypto::Encryptable,
@@ -205,7 +206,7 @@ where
 }
 
 /// Strategy for masking Email
-#[derive(Debug)]
+#[derive(Debug, Copy, Clone, Deserialize)]
 pub enum EmailStrategy {}
 
 impl<T> Strategy<T> for EmailStrategy
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index 4af5a2ed308..18e62b9c1c8 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -944,6 +944,7 @@ impl PaymentRedirectFlow for PaymentRedirectCompleteAuthorize {
                     param: req.param.map(Secret::new),
                     json_payload: Some(req.json_payload.unwrap_or(serde_json::json!({})).into()),
                 }),
+                search_tags: None,
             }),
             ..Default::default()
         };
@@ -1230,6 +1231,7 @@ impl PaymentRedirectFlow for PaymentAuthenticateCompleteAuthorize {
                             req.json_payload.unwrap_or(serde_json::json!({})).into(),
                         ),
                     }),
+                    search_tags: None,
                 }),
                 ..Default::default()
             };
diff --git a/crates/router/src/services/kafka/payment_intent.rs b/crates/router/src/services/kafka/payment_intent.rs
index 0a4ac6d6c2c..0dd11087896 100644
--- a/crates/router/src/services/kafka/payment_intent.rs
+++ b/crates/router/src/services/kafka/payment_intent.rs
@@ -1,7 +1,8 @@
-use common_utils::{crypto::Encryptable, id_type, types::MinorUnit};
+use common_utils::{crypto::Encryptable, hashing::HashedString, id_type, pii, types::MinorUnit};
 use diesel_models::enums as storage_enums;
 use hyperswitch_domain_models::payments::PaymentIntent;
-use masking::Secret;
+use masking::{PeekInterface, Secret};
+use serde_json::Value;
 use time::OffsetDateTime;
 
 #[derive(serde::Serialize, Debug)]
@@ -32,7 +33,10 @@ pub struct KafkaPaymentIntent<'a> {
     pub business_label: Option<&'a String>,
     pub attempt_count: i16,
     pub payment_confirm_source: Option<storage_enums::PaymentSource>,
-    pub billing_details: Option<Encryptable<Secret<serde_json::Value>>>,
+    pub customer_email: Option<HashedString<pii::EmailStrategy>>,
+    pub feature_metadata: Option<&'a Value>,
+    pub merchant_order_reference_id: Option<&'a String>,
+    pub billing_details: Option<Encryptable<Secret<Value>>>,
 }
 
 impl<'a> KafkaPaymentIntent<'a> {
@@ -61,7 +65,16 @@ impl<'a> KafkaPaymentIntent<'a> {
             business_label: intent.business_label.as_ref(),
             attempt_count: intent.attempt_count,
             payment_confirm_source: intent.payment_confirm_source,
-            billing_details: intent.billing_details.clone(),
+            customer_email: intent
+                .customer_details
+                .as_ref()
+                .and_then(|value| value.get_inner().peek().as_object())
+                .and_then(|obj| obj.get("email"))
+                .and_then(|email| email.as_str())
+                .map(|email| HashedString::from(Secret::new(email.to_string()))),
+            feature_metadata: intent.feature_metadata.as_ref(),
+            merchant_order_reference_id: intent.merchant_order_reference_id.as_ref(),
+            billing_details: None,
         }
     }
 }
diff --git a/crates/router/src/services/kafka/payment_intent_event.rs b/crates/router/src/services/kafka/payment_intent_event.rs
index b8ffb659457..1824ad71c7d 100644
--- a/crates/router/src/services/kafka/payment_intent_event.rs
+++ b/crates/router/src/services/kafka/payment_intent_event.rs
@@ -1,7 +1,8 @@
-use common_utils::{crypto::Encryptable, id_type, types::MinorUnit};
+use common_utils::{crypto::Encryptable, hashing::HashedString, id_type, pii, types::MinorUnit};
 use diesel_models::enums as storage_enums;
 use hyperswitch_domain_models::payments::PaymentIntent;
-use masking::Secret;
+use masking::{PeekInterface, Secret};
+use serde_json::Value;
 use time::OffsetDateTime;
 
 #[serde_with::skip_serializing_none]
@@ -33,7 +34,10 @@ pub struct KafkaPaymentIntentEvent<'a> {
     pub business_label: Option<&'a String>,
     pub attempt_count: i16,
     pub payment_confirm_source: Option<storage_enums::PaymentSource>,
-    pub billing_details: Option<Encryptable<Secret<serde_json::Value>>>,
+    pub customer_email: Option<HashedString<pii::EmailStrategy>>,
+    pub feature_metadata: Option<&'a Value>,
+    pub merchant_order_reference_id: Option<&'a String>,
+    pub billing_details: Option<Encryptable<Secret<Value>>>,
 }
 
 impl<'a> KafkaPaymentIntentEvent<'a> {
@@ -62,7 +66,16 @@ impl<'a> KafkaPaymentIntentEvent<'a> {
             business_label: intent.business_label.as_ref(),
             attempt_count: intent.attempt_count,
             payment_confirm_source: intent.payment_confirm_source,
-            billing_details: intent.billing_details.clone(),
+            customer_email: intent
+                .customer_details
+                .as_ref()
+                .and_then(|value| value.get_inner().peek().as_object())
+                .and_then(|obj| obj.get("email"))
+                .and_then(|email| email.as_str())
+                .map(|email| HashedString::from(Secret::new(email.to_string()))),
+            feature_metadata: intent.feature_metadata.as_ref(),
+            merchant_order_reference_id: intent.merchant_order_reference_id.as_ref(),
+            billing_details: None,
         }
     }
 }