feat(events): add type info to outgoing requests & maintain structura…

…l & PII type info (#2956) Co-authored-by: Prasunna Soppa <[email protected]>
juspay · Dec 13, 2023 · 6e82b0b · 6e82b0b
1 parent 3f4167d
commit 6e82b0b
Show file tree

Hide file tree

Showing 93 changed files with 1,496 additions and 2,555 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/config/config.example.toml b/config/config.example.toml
@@ -476,6 +476,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
@@ -484,6 +490,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]
@@ -498,3 +505,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
diff --git a/config/development.toml b/config/development.toml
@@ -513,7 +513,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"

diff --git a/config/docker_compose.toml b/config/docker_compose.toml
@@ -360,7 +360,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"

diff --git a/connector-template/mod.rs b/connector-template/mod.rs
@@ -15,6 +15,7 @@ use crate::{
         self,
         api::{self, ConnectorCommon, ConnectorCommonExt},
         ErrorResponse, Response,
+        RequestContent
     }
 };
 
@@ -158,7 +159,7 @@ impl
         Err(errors::ConnectorError::NotImplemented("get_url method".to_string()).into())
     }
 
-    fn get_request_body(&self, req: &types::PaymentsAuthorizeRouterData, _connectors: &settings::Connectors,) -> CustomResult<Option<types::RequestBody>, errors::ConnectorError> {
+    fn get_request_body(&self, req: &types::PaymentsAuthorizeRouterData, _connectors: &settings::Connectors,) -> CustomResult<RequestContent, errors::ConnectorError> {
         let connector_router_data =
             {{project-name | downcase}}::{{project-name | downcase | pascal_case}}RouterData::try_from((
                 &self.get_currency_unit(),
@@ -187,7 +188,7 @@ impl
                 .headers(types::PaymentsAuthorizeType::get_headers(
                     self, req, connectors,
                 )?)
-                .body(types::PaymentsAuthorizeType::get_request_body(self, req, connectors)?)
+                .set_body(types::PaymentsAuthorizeType::get_request_body(self, req, connectors)?)
                 .build(),
         ))
     }
@@ -304,7 +305,7 @@ impl
         &self,
         _req: &types::PaymentsCaptureRouterData,
         _connectors: &settings::Connectors,
-    ) -> CustomResult<Option<types::RequestBody>, errors::ConnectorError> {
+    ) -> CustomResult<RequestContent, errors::ConnectorError> {
         Err(errors::ConnectorError::NotImplemented("get_request_body method".to_string()).into())
     }
 
@@ -321,7 +322,7 @@ impl
                 .headers(types::PaymentsCaptureType::get_headers(
                     self, req, connectors,
                 )?)
-                .body(types::PaymentsCaptureType::get_request_body(self, req, connectors)?)
+                .set_body(types::PaymentsCaptureType::get_request_body(self, req, connectors)?)
                 .build(),
         ))
     }
@@ -376,7 +377,7 @@ impl
         Err(errors::ConnectorError::NotImplemented("get_url method".to_string()).into())
     }
 
-    fn get_request_body(&self, req: &types::RefundsRouterData<api::Execute>, _connectors: &settings::Connectors,) -> CustomResult<Option<types::RequestBody>, errors::ConnectorError> {
+    fn get_request_body(&self, req: &types::RefundsRouterData<api::Execute>, _connectors: &settings::Connectors,) -> CustomResult<RequestContent, errors::ConnectorError> {
         let connector_router_data =
             {{project-name | downcase}}::{{project-name | downcase | pascal_case}}RouterData::try_from((
                 &self.get_currency_unit(),
@@ -396,7 +397,7 @@ impl
             .url(&types::RefundExecuteType::get_url(self, req, connectors)?)
             .attach_default_headers()
             .headers(types::RefundExecuteType::get_headers(self, req, connectors)?)
-            .body(types::RefundExecuteType::get_request_body(self, req, connectors)?)
+            .set_body(types::RefundExecuteType::get_request_body(self, req, connectors)?)
             .build();
         Ok(Some(request))
     }
@@ -444,7 +445,7 @@ impl
                 .url(&types::RefundSyncType::get_url(self, req, connectors)?)
                 .attach_default_headers()
                 .headers(types::RefundSyncType::get_headers(self, req, connectors)?)
-                .body(types::RefundSyncType::get_request_body(self, req, connectors)?)
+                .set_body(types::RefundSyncType::get_request_body(self, req, connectors)?)
                 .build(),
         ))
     }

diff --git a/crates/common_utils/src/request.rs b/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,15 @@ 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) => serde_json::to_string(&i).unwrap_or_default().into(),
+            RequestContent::FormUrlEncoded(i) => {
+                serde_urlencoded::to_string(&i).unwrap_or_default().into()
+            }
+            RequestContent::Xml(i) => quick_xml::se::to_string(&i).unwrap_or_default().into(),
+            RequestContent::FormData(_) => String::new().into(),
+        }
     }
 }
diff --git a/crates/masking/Cargo.toml b/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"