feat(routing): Add audit trail for routing (#8188)

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

crates/analytics/src/clickhouse.rs

@@ -32,6 +32,7 @@ use crate::{
     connector_events::events::ConnectorEventsResult,
     disputes::{filters::DisputeFilterRow, metrics::DisputeMetricRow},
     outgoing_webhook_event::events::OutgoingWebhookLogsResult,
+    routing_events::events::RoutingEventsResult,
     sdk_events::events::SdkEventsResult,
     types::TableEngine,
 };
@@ -150,6 +151,7 @@ impl AnalyticsDataSource for ClickhouseClient {
             | AnalyticsCollection::SdkEventsAnalytics
             | AnalyticsCollection::ApiEvents
             | AnalyticsCollection::ConnectorEvents
+            | AnalyticsCollection::RoutingEvents
             | AnalyticsCollection::ApiEventsAnalytics
             | AnalyticsCollection::OutgoingWebhookEvent
             | AnalyticsCollection::ActivePaymentsAnalytics => TableEngine::BasicTree,
@@ -187,6 +189,7 @@ impl super::api_event::events::ApiLogsFilterAnalytics for ClickhouseClient {}
 impl super::api_event::filters::ApiEventFilterAnalytics for ClickhouseClient {}
 impl super::api_event::metrics::ApiEventMetricAnalytics for ClickhouseClient {}
 impl super::connector_events::events::ConnectorEventLogAnalytics for ClickhouseClient {}
+impl super::routing_events::events::RoutingEventLogAnalytics for ClickhouseClient {}
 impl super::outgoing_webhook_event::events::OutgoingWebhookLogsFilterAnalytics
     for ClickhouseClient
 {
@@ -236,6 +239,16 @@ impl TryInto<ConnectorEventsResult> for serde_json::Value {
     }
 }
 
+impl TryInto<RoutingEventsResult> for serde_json::Value {
+    type Error = Report<ParsingError>;
+
+    fn try_into(self) -> Result<RoutingEventsResult, Self::Error> {
+        serde_json::from_value(self).change_context(ParsingError::StructParseFailure(
+            "Failed to parse RoutingEventsResult in clickhouse results",
+        ))
+    }
+}
+
 impl TryInto<PaymentMetricRow> for serde_json::Value {
     type Error = Report<ParsingError>;
 
@@ -471,6 +484,7 @@ impl ToSql<ClickhouseClient> for AnalyticsCollection {
             Self::DisputeSessionized => Ok("sessionizer_dispute".to_string()),
             Self::ActivePaymentsAnalytics => Ok("active_payments".to_string()),
             Self::Authentications => Ok("authentications".to_string()),
+            Self::RoutingEvents => Ok("routing_events_audit".to_string()),
         }
     }
 }

crates/analytics/src/lib.rs

@@ -16,6 +16,7 @@ pub mod payment_intents;
 pub mod payments;
 mod query;
 pub mod refunds;
+pub mod routing_events;
 pub mod sdk_events;
 pub mod search;
 mod sqlx;
@@ -1165,6 +1166,7 @@ pub enum AnalyticsFlow {
     GetDisputeFilters,
     GetDisputeMetrics,
     GetSankey,
+    GetRoutingEvents,
 }
 
 impl FlowMetric for AnalyticsFlow {}

crates/analytics/src/routing_events.rs

@@ -0,0 +1,5 @@
+mod core;
+pub mod events;
+pub trait RoutingEventAnalytics: events::RoutingEventLogAnalytics {}
+
+pub use self::core::routing_events_core;

crates/analytics/src/routing_events/core.rs

@@ -0,0 +1,26 @@
+use api_models::analytics::routing_events::RoutingEventsRequest;
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+
+use super::events::{get_routing_events, RoutingEventsResult};
+use crate::{errors::AnalyticsResult, types::FiltersError, AnalyticsProvider};
+
+pub async fn routing_events_core(
+    pool: &AnalyticsProvider,
+    req: RoutingEventsRequest,
+    merchant_id: &common_utils::id_type::MerchantId,
+) -> AnalyticsResult<Vec<RoutingEventsResult>> {
+    let data = match pool {
+        AnalyticsProvider::Sqlx(_) => Err(FiltersError::NotImplemented(
+            "Connector Events not implemented for SQLX",
+        ))
+        .attach_printable("SQL Analytics is not implemented for Connector Events"),
+        AnalyticsProvider::Clickhouse(ckh_pool)
+        | AnalyticsProvider::CombinedSqlx(_, ckh_pool)
+        | AnalyticsProvider::CombinedCkh(_, ckh_pool) => {
+            get_routing_events(merchant_id, req, ckh_pool).await
+        }
+    }
+    .switch()?;
+    Ok(data)
+}

crates/analytics/src/routing_events/events.rs

@@ -0,0 +1,73 @@
+use api_models::analytics::{routing_events::RoutingEventsRequest, Granularity};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use crate::{
+    query::{Aggregate, GroupByClause, QueryBuilder, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, FiltersError, FiltersResult, LoadRow},
+};
+pub trait RoutingEventLogAnalytics: LoadRow<RoutingEventsResult> {}
+
+pub async fn get_routing_events<T>(
+    merchant_id: &common_utils::id_type::MerchantId,
+    query_param: RoutingEventsRequest,
+    pool: &T,
+) -> FiltersResult<Vec<RoutingEventsResult>>
+where
+    T: AnalyticsDataSource + RoutingEventLogAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::RoutingEvents);
+    query_builder.add_select_column("*").switch()?;
+
+    query_builder
+        .add_filter_clause("merchant_id", merchant_id)
+        .switch()?;
+
+    query_builder
+        .add_filter_clause("payment_id", &query_param.payment_id)
+        .switch()?;
+
+    if let Some(refund_id) = query_param.refund_id {
+        query_builder
+            .add_filter_clause("refund_id", &refund_id)
+            .switch()?;
+    }
+
+    if let Some(dispute_id) = query_param.dispute_id {
+        query_builder
+            .add_filter_clause("dispute_id", &dispute_id)
+            .switch()?;
+    }
+
+    query_builder
+        .execute_query::<RoutingEventsResult, _>(pool)
+        .await
+        .change_context(FiltersError::QueryBuildingError)?
+        .change_context(FiltersError::QueryExecutionFailure)
+}
+
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub struct RoutingEventsResult {
+    pub merchant_id: common_utils::id_type::MerchantId,
+    pub profile_id: common_utils::id_type::ProfileId,
+    pub payment_id: String,
+    pub routable_connectors: String,
+    pub payment_connector: Option<String>,
+    pub request_id: Option<String>,
+    pub flow: String,
+    pub url: Option<String>,
+    pub request: String,
+    pub response: Option<String>,
+    pub error: Option<String>,
+    pub status_code: Option<u16>,
+    #[serde(with = "common_utils::custom_serde::iso8601")]
+    pub created_at: PrimitiveDateTime,
+    pub method: String,
+    pub routing_engine: String,
+}

crates/analytics/src/sqlx.rs

@@ -1400,6 +1400,8 @@ impl ToSql<SqlxClient> for AnalyticsCollection {
                 .attach_printable("DisputeSessionized table is not implemented for Sqlx"))?,
             Self::Authentications => Err(error_stack::report!(ParsingError::UnknownError)
                 .attach_printable("Authentications table is not implemented for Sqlx"))?,
+            Self::RoutingEvents => Err(error_stack::report!(ParsingError::UnknownError)
+                .attach_printable("RoutingEvents table is not implemented for Sqlx"))?,
         }
     }
 }

crates/analytics/src/types.rs

@@ -42,6 +42,7 @@ pub enum AnalyticsCollection {
     DisputeSessionized,
     ApiEventsAnalytics,
     ActivePaymentsAnalytics,
+    RoutingEvents,
 }
 
 #[allow(dead_code)]