feat(events): Add events framework for registering events

Makefile

@@ -65,10 +65,10 @@ endif
 # Format Rust sources with rustfmt.
 #
 # Usage :
-#	make fmt [writing=(no|yes)]
+#	make fmt [dry_run=(no|yes)]
 
 fmt :
-	cargo +nightly fmt --all $(if $(call eq,$(writing),yes),,-- --check)
+	cargo +nightly fmt --all $(if $(call eq,$(dry_run),yes),-- --check,)
 
 # Lint Rust sources with Clippy.
 #

config/config.example.toml

@@ -563,6 +563,7 @@ api_logs_topic = "topic"              # Kafka topic to be used for incoming api
 connector_logs_topic = "topic"        # Kafka topic to be used for connector api events
 outgoing_webhook_logs_topic = "topic" # Kafka topic to be used for outgoing webhook events
 dispute_analytics_topic = "topic"     # Kafka topic to be used for Dispute events
+audit_events_topic = "topic"          # Kafka topic to be used for Payment Audit events
 
 # File storage configuration
 [file_storage]

config/deployments/env_specific.toml

@@ -78,6 +78,8 @@ 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
 outgoing_webhook_logs_topic = "topic" # Kafka topic to be used for outgoing webhook events
+dispute_analytics_topic = "topic"     # Kafka topic to be used for Dispute events
+audit_events_topic = "topic"          # Kafka topic to be used for Payment Audit events
 
 # File storage configuration
 [file_storage]

config/development.toml

@@ -551,6 +551,7 @@ api_logs_topic = "hyperswitch-api-log-events"
 connector_logs_topic = "hyperswitch-connector-api-events"
 outgoing_webhook_logs_topic = "hyperswitch-outgoing-webhook-events"
 dispute_analytics_topic = "hyperswitch-dispute-events"
+audit_events_topic = "hyperswitch-audit-events"
 
 [analytics]
 source = "sqlx"

config/docker_compose.toml

@@ -401,6 +401,7 @@ api_logs_topic = "hyperswitch-api-log-events"
 connector_logs_topic = "hyperswitch-connector-api-events"
 outgoing_webhook_logs_topic = "hyperswitch-outgoing-webhook-events"
 dispute_analytics_topic = "hyperswitch-dispute-events"
+audit_events_topic = "hyperswitch-audit-events"
 
 [analytics]
 source = "sqlx"

crates/data_models/src/lib.rs

@@ -2,7 +2,7 @@ pub mod errors;
 pub mod mandates;
 pub mod payments;
 
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
 pub enum RemoteStorageObject<T: ForeignIDRef> {
     ForeignID(String),
     Object(T),

crates/data_models/src/payments.rs

@@ -9,7 +9,7 @@ use common_enums as storage_enums;
 use self::payment_attempt::PaymentAttempt;
 use crate::RemoteStorageObject;
 
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
 pub struct PaymentIntent {
     pub id: i32,
     pub payment_id: String,

crates/events/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "events"
+version = "0.1.0"
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+error-stack = "0.3.1"
+masking = { version = "0.1.0", path = "../masking" }
+serde_json = "1.0.114"
+thiserror = "1.0.58"
+time = "0.3.34"

crates/events/src/lib.rs

@@ -0,0 +1,149 @@
+#![cfg_attr(docsrs, feature(doc_auto_cfg, doc_cfg_hide))]
+#![cfg_attr(docsrs, doc(cfg_hide(doc)))]
+#![forbid(unsafe_code)]
+#![warn(missing_docs)]
+
+//!
+//! A generic event handler system.
+//! This library consists of 4 parts:
+//! Event Sink: A trait that defines how events are published. This could be a simple logger, a message queue, or a database.
+//! EventContext: A struct that holds the event sink and metadata about the event. This is used to create events. This can be used to add metadata to all events, such as the user who triggered the event.
+//! EventInfo: A trait that defines the metadata that is sent with the event. This trait is used to define the data that is sent with the event it works with the EventContext to add metadata to all events.
+//! Event: A trait that defines the event itself. This trait is used to define the data that is sent with the event and defines the event's type & identifier.
+//!
+
+use std::rc::Rc;
+
+use error_stack::Result;
+use time::PrimitiveDateTime;
+
+/// Errors that can occur when working with events.
+#[derive(Debug, Clone, thiserror::Error)]
+pub enum EventsError {
+    /// An error occurred when publishing the event.
+    #[error("Generic Error")]
+    GenericError,
+    #[error("Event serialization error: {0}")]
+    SerializationError(String),
+}
+
+/// An event that can be published.
+pub trait Event: EventInfo {
+    /// The type of the event.
+    type EventType;
+    /// The timestamp of the event.
+    fn timestamp(&self) -> PrimitiveDateTime;
+
+    /// The (unique) identifier of the event.
+    fn identifier(&self) -> String;
+
+    /// The class/type of the event. This is used to group/categorize events together.
+    fn class(&self) -> Self::EventType;
+}
+
+/// An Event sink that can publish events.
+/// This could be a simple logger, a message queue, or a database.
+pub trait EventSink<T> {
+    /// Publish an event.
+    /// The parameters for this function are determined from the Event trait.
+    fn publish_event(
+        &self,
+        data: serde_json::Value,
+        identifier: String,
+        topic: T,
+        timestamp: PrimitiveDateTime,
+    ) -> Result<(), EventsError>;
+}
+
+/// Hold the context information for any events
+#[derive(Clone)]
+pub struct EventContext<T> {
+    event_sink: Rc<Box<dyn EventSink<T>>>,
+    metadata: Vec<Rc<Box<dyn EventInfo>>>,
+}
+
+/// intermediary structure to build events with in-place info.
+pub struct EventBuilder<T> {
+    event_sink: Rc<Box<dyn EventSink<T>>>,
+    src_metadata: Vec<Rc<Box<dyn EventInfo>>>,
+    event_metadata: Vec<Rc<Box<dyn EventInfo>>>,
+    event: Box<dyn Event<EventType = T>>,
+}
+
+impl<T> EventBuilder<T> {
+    /// Add metadata to the event.
+    pub fn with<E: EventInfo + 'static>(mut self, info: E) -> Self {
+        let boxed_event: Box<dyn EventInfo> = Box::new(info);
+        self.event_metadata.push(boxed_event.into());
+        self
+    }
+    /// Emit the event.
+    pub fn emit(self) -> Result<(), EventsError> {
+        self.event_sink.publish_event(
+            self.data()?,
+            self.event.identifier(),
+            self.event.class(),
+            self.event.timestamp(),
+        )
+    }
+}
+
+impl<T> EventInfo for EventBuilder<T> {
+    fn data(&self) -> Result<serde_json::Value, EventsError> {
+        self.src_metadata
+            .iter()
+            .chain(self.event_metadata.iter())
+            .map(|info| info.data().map(|d| (info.key(), d)))
+            .collect()
+    }
+
+    fn key(&self) -> String {
+        self.event.key()
+    }
+}
+
+impl<T> EventContext<T> {
+    /// Create a new event context.
+    pub fn new(event_sink: Rc<Box<dyn EventSink<T>>>) -> Self {
+        Self {
+            event_sink,
+            metadata: Vec::new(),
+        }
+    }
+
+    /// Add metadata to the event context.
+    pub fn record_info<E: EventInfo + 'static>(&mut self, info: E) {
+        let boxed_event: Box<dyn EventInfo> = Box::new(info);
+        self.metadata.push(boxed_event.into());
+    }
+
+    /// Emit an event.
+    pub fn emit(&self, event: Box<dyn Event<EventType = T>>) -> Result<(), EventsError> {
+        EventBuilder {
+            event_sink: self.event_sink.clone(),
+            src_metadata: self.metadata.clone(),
+            event_metadata: vec![],
+            event,
+        }
+        .emit()
+    }
+
+    /// Create an event builder.
+    pub fn event(&self, event: Box<dyn Event<EventType = T>>) -> EventBuilder<T> {
+        EventBuilder {
+            event_sink: self.event_sink.clone(),
+            src_metadata: self.metadata.clone(),
+            event_metadata: vec![],
+            event,
+        }
+    }
+}
+
+/// Add information/metadata to the current context of an event.
+pub trait EventInfo {
+    /// The data that is sent with the event.
+    fn data(&self) -> Result<serde_json::Value, EventsError>;
+
+    /// The key identifying the data for an event.
+    fn key(&self) -> String;
+}

crates/router/Cargo.toml

@@ -124,6 +124,7 @@ rdkafka = "0.36.0"
 isocountry = "0.3.2"
 iso_currency = "0.4.4"
 actix-http = "3.3.1"
+events = { version = "0.1.0", path = "../events" }
 
 [build-dependencies]
 router_env = { version = "0.1.0", path = "../router_env", default-features = false }

crates/router/src/core/payments.rs

@@ -393,7 +393,7 @@ where
                             &connectors,
                         )
                         .await?;
-                    call_multiple_connectors_service(
+                    Box::pin(call_multiple_connectors_service(
                         state,
                         &merchant_account,
                         &key_store,
@@ -402,7 +402,7 @@ where
                         payment_data,
                         &customer,
                         session_surcharge_details,
-                    )
+                    ))
                     .await?
                 }
             };
@@ -3427,7 +3427,7 @@ pub async fn payment_external_authentication(
             id: profile_id.to_string(),
         })?;
 
-    let authentication_response = authentication_core::perform_authentication(
+    let authentication_response = Box::pin(authentication_core::perform_authentication(
         &state,
         authentication_connector,
         payment_method_details.0,
@@ -3451,7 +3451,7 @@ pub async fn payment_external_authentication(
         req.sdk_information,
         req.threeds_method_comp_ind,
         optional_customer.and_then(|customer| customer.email.map(common_utils::pii::Email::from)),
-    )
+    ))
     .await?;
     Ok(services::ApplicationResponse::Json(
         api_models::payments::PaymentsExternalAuthenticationResponse {