feat(events): add hashed customer_email and feature_metadata (#5220)

Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>

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",

api-reference/openapi_spec.json

@@ -9485,6 +9485,14 @@
               }
             ],
             "nullable": true
+          },
+          "search_tags": {
+            "allOf": [
+              {
+                "$ref": "#/components/schemas/RedirectResponse"
+              }
+            ],
+            "nullable": true
           }
         }
       },

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]

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()?;
             }
         };

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)]

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

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"] }

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)
+    }
+}

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;
 

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

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()
             };

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)]
@@ -33,7 +34,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> {
@@ -63,7 +67,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,
         }
     }
 }

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]
@@ -34,7 +35,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> {
@@ -64,7 +68,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,
         }
     }
 }