feat(analytics): add new filters, dimensions and metrics for authentication analytics (#7451)

Co-authored-by: Sandeep Kumar <[email protected]>

Author: 2 people

crates/analytics/src/auth_events.rs

@@ -1,6 +1,8 @@
 pub mod accumulator;
 mod core;
+pub mod filters;
 pub mod metrics;
+pub mod types;
 pub use accumulator::{AuthEventMetricAccumulator, AuthEventMetricsAccumulator};
 
-pub use self::core::get_metrics;
+pub use self::core::{get_filters, get_metrics};

crates/analytics/src/auth_events/accumulator.rs

@@ -6,12 +6,14 @@ use super::metrics::AuthEventMetricRow;
 pub struct AuthEventMetricsAccumulator {
     pub authentication_count: CountAccumulator,
     pub authentication_attempt_count: CountAccumulator,
+    pub authentication_error_message: AuthenticationErrorMessageAccumulator,
     pub authentication_success_count: CountAccumulator,
     pub challenge_flow_count: CountAccumulator,
     pub challenge_attempt_count: CountAccumulator,
     pub challenge_success_count: CountAccumulator,
     pub frictionless_flow_count: CountAccumulator,
     pub frictionless_success_count: CountAccumulator,
+    pub authentication_funnel: CountAccumulator,
 }
 
 #[derive(Debug, Default)]
@@ -20,6 +22,11 @@ pub struct CountAccumulator {
     pub count: Option<i64>,
 }
 
+#[derive(Debug, Default)]
+pub struct AuthenticationErrorMessageAccumulator {
+    pub count: Option<i64>,
+}
+
 pub trait AuthEventMetricAccumulator {
     type MetricOutput;
 
@@ -44,6 +51,22 @@ impl AuthEventMetricAccumulator for CountAccumulator {
     }
 }
 
+impl AuthEventMetricAccumulator for AuthenticationErrorMessageAccumulator {
+    type MetricOutput = Option<u64>;
+    #[inline]
+    fn add_metrics_bucket(&mut self, metrics: &AuthEventMetricRow) {
+        self.count = match (self.count, metrics.count) {
+            (None, None) => None,
+            (None, i @ Some(_)) | (i @ Some(_), None) => i,
+            (Some(a), Some(b)) => Some(a + b),
+        }
+    }
+    #[inline]
+    fn collect(self) -> Self::MetricOutput {
+        self.count.and_then(|i| u64::try_from(i).ok())
+    }
+}
+
 impl AuthEventMetricsAccumulator {
     pub fn collect(self) -> AuthEventMetricsBucketValue {
         AuthEventMetricsBucketValue {
@@ -55,6 +78,8 @@ impl AuthEventMetricsAccumulator {
             challenge_success_count: self.challenge_success_count.collect(),
             frictionless_flow_count: self.frictionless_flow_count.collect(),
             frictionless_success_count: self.frictionless_success_count.collect(),
+            error_message_count: self.authentication_error_message.collect(),
+            authentication_funnel: self.authentication_funnel.collect(),
         }
     }
 }

crates/analytics/src/auth_events/core.rs

@@ -1,13 +1,20 @@
 use std::collections::HashMap;
 
 use api_models::analytics::{
-    auth_events::{AuthEventMetrics, AuthEventMetricsBucketIdentifier, MetricsBucketResponse},
-    AnalyticsMetadata, GetAuthEventMetricRequest, MetricsResponse,
+    auth_events::{
+        AuthEventDimensions, AuthEventMetrics, AuthEventMetricsBucketIdentifier,
+        MetricsBucketResponse,
+    },
+    AuthEventFilterValue, AuthEventFiltersResponse, AuthEventMetricsResponse,
+    AuthEventsAnalyticsMetadata, GetAuthEventFilterRequest, GetAuthEventMetricRequest,
 };
-use error_stack::ResultExt;
+use error_stack::{report, ResultExt};
 use router_env::{instrument, tracing};
 
-use super::AuthEventMetricsAccumulator;
+use super::{
+    filters::{get_auth_events_filter_for_dimension, AuthEventFilterRow},
+    AuthEventMetricsAccumulator,
+};
 use crate::{
     auth_events::AuthEventMetricAccumulator,
     errors::{AnalyticsError, AnalyticsResult},
@@ -19,7 +26,7 @@ pub async fn get_metrics(
     pool: &AnalyticsProvider,
     merchant_id: &common_utils::id_type::MerchantId,
     req: GetAuthEventMetricRequest,
-) -> AnalyticsResult<MetricsResponse<MetricsBucketResponse>> {
+) -> AnalyticsResult<AuthEventMetricsResponse<MetricsBucketResponse>> {
     let mut metrics_accumulator: HashMap<
         AuthEventMetricsBucketIdentifier,
         AuthEventMetricsAccumulator,
@@ -34,7 +41,9 @@ pub async fn get_metrics(
             let data = pool
                 .get_auth_event_metrics(
                     &metric_type,
+                    &req.group_by_names.clone(),
                     &merchant_id_scoped,
+                    &req.filters,
                     req.time_series.map(|t| t.granularity),
                     &req.time_range,
                 )
@@ -77,22 +86,94 @@ pub async fn get_metrics(
                 AuthEventMetrics::FrictionlessSuccessCount => metrics_builder
                     .frictionless_success_count
                     .add_metrics_bucket(&value),
+                AuthEventMetrics::AuthenticationErrorMessage => metrics_builder
+                    .authentication_error_message
+                    .add_metrics_bucket(&value),
+                AuthEventMetrics::AuthenticationFunnel => metrics_builder
+                    .authentication_funnel
+                    .add_metrics_bucket(&value),
             }
         }
     }
 
+    let mut total_error_message_count = 0;
     let query_data: Vec<MetricsBucketResponse> = metrics_accumulator
         .into_iter()
-        .map(|(id, val)| MetricsBucketResponse {
-            values: val.collect(),
-            dimensions: id,
+        .map(|(id, val)| {
+            let collected_values = val.collect();
+            if let Some(count) = collected_values.error_message_count {
+                total_error_message_count += count;
+            }
+            MetricsBucketResponse {
+                values: collected_values,
+                dimensions: id,
+            }
         })
         .collect();
-
-    Ok(MetricsResponse {
+    Ok(AuthEventMetricsResponse {
         query_data,
-        meta_data: [AnalyticsMetadata {
-            current_time_range: req.time_range,
+        meta_data: [AuthEventsAnalyticsMetadata {
+            total_error_message_count: Some(total_error_message_count),
         }],
     })
 }
+
+pub async fn get_filters(
+    pool: &AnalyticsProvider,
+    req: GetAuthEventFilterRequest,
+    merchant_id: &common_utils::id_type::MerchantId,
+) -> AnalyticsResult<AuthEventFiltersResponse> {
+    let mut res = AuthEventFiltersResponse::default();
+    for dim in req.group_by_names {
+        let values = match pool {
+                        AnalyticsProvider::Sqlx(_pool) => {
+                            Err(report!(AnalyticsError::UnknownError))
+            }
+                        AnalyticsProvider::Clickhouse(pool) => {
+                get_auth_events_filter_for_dimension(dim, merchant_id, &req.time_range, pool)
+                    .await
+                    .map_err(|e| e.change_context(AnalyticsError::UnknownError))
+            }
+                    AnalyticsProvider::CombinedCkh(sqlx_pool, ckh_pool) | AnalyticsProvider::CombinedSqlx(sqlx_pool, ckh_pool) => {
+                let ckh_result = get_auth_events_filter_for_dimension(
+                    dim,
+                    merchant_id,
+                    &req.time_range,
+                    ckh_pool,
+                )
+                .await
+                .map_err(|e| e.change_context(AnalyticsError::UnknownError));
+                let sqlx_result = get_auth_events_filter_for_dimension(
+                    dim,
+                    merchant_id,
+                    &req.time_range,
+                    sqlx_pool,
+                )
+                .await
+                .map_err(|e| e.change_context(AnalyticsError::UnknownError));
+                match (&sqlx_result, &ckh_result) {
+                    (Ok(ref sqlx_res), Ok(ref ckh_res)) if sqlx_res != ckh_res => {
+                        router_env::logger::error!(clickhouse_result=?ckh_res, postgres_result=?sqlx_res, "Mismatch between clickhouse & postgres refunds analytics filters")
+                    },
+                    _ => {}
+                };
+                ckh_result
+            }
+        }
+        .change_context(AnalyticsError::UnknownError)?
+        .into_iter()
+        .filter_map(|fil: AuthEventFilterRow| match dim {
+            AuthEventDimensions::AuthenticationStatus => fil.authentication_status.map(|i| i.as_ref().to_string()),
+            AuthEventDimensions::TransactionStatus => fil.trans_status.map(|i| i.as_ref().to_string()),
+            AuthEventDimensions::ErrorMessage => fil.error_message,
+            AuthEventDimensions::AuthenticationConnector => fil.authentication_connector.map(|i| i.as_ref().to_string()),
+            AuthEventDimensions::MessageVersion => fil.message_version,
+        })
+        .collect::<Vec<String>>();
+        res.query_data.push(AuthEventFilterValue {
+            dimension: dim,
+            values,
+        })
+    }
+    Ok(res)
+}

crates/analytics/src/auth_events/filters.rs

@@ -0,0 +1,60 @@
+use api_models::analytics::{auth_events::AuthEventDimensions, Granularity, TimeRange};
+use common_utils::errors::ReportSwitchExt;
+use diesel_models::enums::{AuthenticationConnectors, AuthenticationStatus, TransactionStatus};
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use crate::{
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, ToSql, Window},
+    types::{
+        AnalyticsCollection, AnalyticsDataSource, DBEnumWrapper, FiltersError, FiltersResult,
+        LoadRow,
+    },
+};
+
+pub trait AuthEventFilterAnalytics: LoadRow<AuthEventFilterRow> {}
+
+pub async fn get_auth_events_filter_for_dimension<T>(
+    dimension: AuthEventDimensions,
+    merchant_id: &common_utils::id_type::MerchantId,
+    time_range: &TimeRange,
+    pool: &T,
+) -> FiltersResult<Vec<AuthEventFilterRow>>
+where
+    T: AnalyticsDataSource + AuthEventFilterAnalytics,
+    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::Authentications);
+
+    query_builder.add_select_column(dimension).switch()?;
+    time_range
+        .set_filter_clause(&mut query_builder)
+        .attach_printable("Error filtering time range")
+        .switch()?;
+
+    query_builder
+        .add_filter_clause("merchant_id", merchant_id)
+        .switch()?;
+
+    query_builder.set_distinct();
+
+    query_builder
+        .execute_query::<AuthEventFilterRow, _>(pool)
+        .await
+        .change_context(FiltersError::QueryBuildingError)?
+        .change_context(FiltersError::QueryExecutionFailure)
+}
+
+#[derive(Debug, serde::Serialize, Eq, PartialEq, serde::Deserialize)]
+pub struct AuthEventFilterRow {
+    pub authentication_status: Option<DBEnumWrapper<AuthenticationStatus>>,
+    pub trans_status: Option<DBEnumWrapper<TransactionStatus>>,
+    pub error_message: Option<String>,
+    pub authentication_connector: Option<DBEnumWrapper<AuthenticationConnectors>>,
+    pub message_version: Option<String>,
+}