feat(events): add type info to outgoing requests & maintain structural & PII type info

config/config.example.toml

@@ -469,6 +469,12 @@ pm_auth_key = "Some_pm_auth_key"
 [analytics]
 source = "sqlx" # The Analytics source/strategy to be used
 
+[analytics.clickhouse]
+username = ""      # Clickhouse username
+password = ""      # Clickhouse password (optional)
+host = ""          # Clickhouse host in http(s)://<URL>:<PORT> format
+database_name = "" # Clickhouse database name
+
 [analytics.sqlx]
 username = "db_user"      # Analytics DB Username
 password = "db_pass"      # Analytics DB Password
@@ -477,6 +483,7 @@ port = 5432               # Analytics DB Port
 dbname = "hyperswitch_db" # Name of Database
 pool_size = 5             # Number of connections to keep open
 connection_timeout = 10   # Timeout for database connection in seconds
+queue_strategy = "Fifo"   # Add the queue strategy used by the database bb8 client
 
 # Config for KV setup
 [kv_config]
@@ -491,3 +498,23 @@ client_id = "paypal_client_id" # Client ID for PayPal onboarding
 client_secret = "paypal_secret_key" # Secret key for PayPal onboarding
 partner_id = "paypal_partner_id" # Partner ID for PayPal onboarding
 enabled = true # Switch to enable or disable PayPal onboarding
+
+[frm]
+enabled = true
+
+[paypal_onboarding]
+client_id = "paypal_client_id" # Client ID for PayPal onboarding
+client_secret = "paypal_secret_key" # Secret key for PayPal onboarding
+partner_id = "paypal_partner_id" # Partner ID for PayPal onboarding
+enabled = true # Switch to enable or disable PayPal onboarding
+
+[events]
+source = "logs"                    # The event sink to push events supports kafka or logs (stdout)
+
+[events.kafka]
+brokers = []                       # Kafka broker urls for bootstrapping the client
+intent_analytics_topic = "topic"   # Kafka topic to be used for PaymentIntent events
+attempt_analytics_topic = "topic"  # Kafka topic to be used for PaymentAttempt events
+refund_analytics_topic = "topic"   # Kafka topic to be used for Refund events
+api_logs_topic = "topic"           # Kafka topic to be used for incoming api events
+connector_logs_topic = "topic"     # Kafka topic to be used for connector api events

config/docker_compose.toml

@@ -344,7 +344,7 @@ intent_analytics_topic = "hyperswitch-payment-intent-events"
 attempt_analytics_topic = "hyperswitch-payment-attempt-events"
 refund_analytics_topic = "hyperswitch-refund-events"
 api_logs_topic = "hyperswitch-api-log-events"
-connector_events_topic = "hyperswitch-connector-api-events"
+connector_logs_topic = "hyperswitch-connector-api-events"
 
 [analytics]
 source = "sqlx"

crates/common_utils/src/request.rs

@@ -25,6 +25,7 @@ pub enum ContentType {
     Json,
     FormUrlEncoded,
     FormData,
+    Xml,
 }
 
 fn default_request_headers() -> [(String, Maskable<String>); 1] {
@@ -37,12 +38,28 @@ fn default_request_headers() -> [(String, Maskable<String>); 1] {
 pub struct Request {
     pub url: String,
     pub headers: Headers,
-    pub payload: Option<Secret<String>>,
     pub method: Method,
-    pub content_type: Option<ContentType>,
     pub certificate: Option<String>,
     pub certificate_key: Option<String>,
-    pub form_data: Option<reqwest::multipart::Form>,
+    pub body: Option<RequestContent>,
+}
+
+impl std::fmt::Debug for RequestContent {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(match self {
+            Self::Json(_) => "JsonRequestBody",
+            Self::FormUrlEncoded(_) => "FormUrlEncodedRequestBody",
+            Self::FormData(_) => "FormDataRequestBody",
+            Self::Xml(_) => "XmlRequestBody",
+        })
+    }
+}
+
+pub enum RequestContent {
+    Json(Box<dyn masking::ErasedMaskSerialize + Send>),
+    FormUrlEncoded(Box<dyn masking::ErasedMaskSerialize + Send>),
+    FormData(reqwest::multipart::Form),
+    Xml(Box<dyn masking::ErasedMaskSerialize + Send>),
 }
 
 impl Request {
@@ -51,16 +68,14 @@ impl Request {
             method,
             url: String::from(url),
             headers: std::collections::HashSet::new(),
-            payload: None,
-            content_type: None,
             certificate: None,
             certificate_key: None,
-            form_data: None,
+            body: None,
         }
     }
 
-    pub fn set_body(&mut self, body: String) {
-        self.payload = Some(body.into());
+    pub fn set_body<T: Into<RequestContent>>(&mut self, body: T) {
+        self.body.replace(body.into());
     }
 
     pub fn add_default_headers(&mut self) {
@@ -71,33 +86,23 @@ impl Request {
         self.headers.insert((String::from(header), value));
     }
 
-    pub fn add_content_type(&mut self, content_type: ContentType) {
-        self.content_type = Some(content_type);
-    }
-
     pub fn add_certificate(&mut self, certificate: Option<String>) {
         self.certificate = certificate;
     }
 
     pub fn add_certificate_key(&mut self, certificate_key: Option<String>) {
         self.certificate = certificate_key;
     }
-
-    pub fn set_form_data(&mut self, form_data: reqwest::multipart::Form) {
-        self.form_data = Some(form_data);
-    }
 }
 
 #[derive(Debug)]
 pub struct RequestBuilder {
     pub url: String,
     pub headers: Headers,
-    pub payload: Option<Secret<String>>,
     pub method: Method,
-    pub content_type: Option<ContentType>,
     pub certificate: Option<String>,
     pub certificate_key: Option<String>,
-    pub form_data: Option<reqwest::multipart::Form>,
+    pub body: Option<RequestContent>,
 }
 
 impl RequestBuilder {
@@ -106,11 +111,9 @@ impl RequestBuilder {
             method: Method::Get,
             url: String::with_capacity(1024),
             headers: std::collections::HashSet::new(),
-            payload: None,
-            content_type: None,
             certificate: None,
             certificate_key: None,
-            form_data: None,
+            body: None,
         }
     }
 
@@ -140,18 +143,8 @@ impl RequestBuilder {
         self
     }
 
-    pub fn form_data(mut self, form_data: Option<reqwest::multipart::Form>) -> Self {
-        self.form_data = form_data;
-        self
-    }
-
-    pub fn body(mut self, option_body: Option<RequestBody>) -> Self {
-        self.payload = option_body.map(RequestBody::get_inner_value);
-        self
-    }
-
-    pub fn content_type(mut self, content_type: ContentType) -> Self {
-        self.content_type = Some(content_type);
+    pub fn set_body<T: Into<RequestContent>>(mut self, body: T) -> Self {
+        self.body.replace(body.into());
         self
     }
 
@@ -170,11 +163,9 @@ impl RequestBuilder {
             method: self.method,
             url: self.url,
             headers: self.headers,
-            payload: self.payload,
-            content_type: self.content_type,
             certificate: self.certificate,
             certificate_key: self.certificate_key,
-            form_data: self.form_data,
+            body: self.body,
         }
     }
 }
@@ -201,7 +192,13 @@ impl RequestBody {
         logger::info!(connector_request_body=?body);
         Ok(Self(Secret::new(encoder(body)?)))
     }
-    pub fn get_inner_value(request_body: Self) -> Secret<String> {
-        request_body.0
+
+    pub fn get_inner_value(request_body: RequestContent) -> Secret<String> {
+        match request_body {
+            RequestContent::Json(i)
+            | RequestContent::FormUrlEncoded(i)
+            | RequestContent::Xml(i) => serde_json::to_string(&i).unwrap_or_default().into(),
+            RequestContent::FormData(_) => String::new().into(),
+        }
     }
 }

crates/masking/Cargo.toml

@@ -19,6 +19,7 @@ rustdoc-args = ["--cfg", "docsrs"]
 [dependencies]
 bytes = { version = "1", optional = true }
 diesel = { version = "2.1.0", features = ["postgres", "serde_json", "time"], optional = true }
+erased-serde = "0.3.31"
 serde = { version = "1", features = ["derive"], optional = true }
 serde_json = { version = "1.0.108", optional = true }
 subtle = "=2.4.1"

crates/masking/src/serde.rs

@@ -2,6 +2,7 @@
 //! Serde-related.
 //!
 
+pub use erased_serde::Serialize as ErasedSerialize;
 pub use serde::{de, ser, Deserialize, Serialize, Serializer};
 use serde_json::{value::Serializer as JsonValueSerializer, Value};
 
@@ -99,20 +100,32 @@ pub fn masked_serialize<T: Serialize>(value: &T) -> Result<Value, serde_json::Er
 /// like &dyn Serialize or boxed trait objects like Box<dyn Serialize> because of Rust's "object safety" rules.
 /// In particular, the trait contains generic methods which cannot be made into a trait object.
 /// In this case we remove the generic for assuming the serialization to be of 2 types only raw json or masked json
-pub trait ErasedMaskSerialize {
+pub trait ErasedMaskSerialize: ErasedSerialize {
     /// Masked serialization.
     fn masked_serialize(&self) -> Result<Value, serde_json::Error>;
-    /// Normal serialization.
-    fn raw_serialize(&self) -> Result<Value, serde_json::Error>;
 }
 
-impl<T: Serialize> ErasedMaskSerialize for T {
+impl<T: Serialize + ErasedSerialize> ErasedMaskSerialize for T {
     fn masked_serialize(&self) -> Result<Value, serde_json::Error> {
         masked_serialize(self)
     }
+}
+
+impl<'a> Serialize for dyn ErasedMaskSerialize + 'a {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        erased_serde::serialize(self, serializer)
+    }
+}
 
-    fn raw_serialize(&self) -> Result<Value, serde_json::Error> {
-        serde_json::to_value(self)
+impl<'a> Serialize for dyn ErasedMaskSerialize + 'a + Send {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        erased_serde::serialize(self, serializer)
     }
 }