feat(analytics): Add v2 payment analytics (payment-intents analytics)

crates/analytics/src/clickhouse.rs

@@ -10,6 +10,7 @@ use super::{
     active_payments::metrics::ActivePaymentsMetricRow,
     auth_events::metrics::AuthEventMetricRow,
     health_check::HealthCheck,
+    payment_intents::{filters::PaymentIntentFilterRow, metrics::PaymentIntentMetricRow},
     payments::{
         distribution::PaymentDistributionRow, filters::FilterRow, metrics::PaymentMetricRow,
     },
@@ -157,6 +158,8 @@ where
 impl super::payments::filters::PaymentFilterAnalytics for ClickhouseClient {}
 impl super::payments::metrics::PaymentMetricAnalytics for ClickhouseClient {}
 impl super::payments::distribution::PaymentDistributionAnalytics for ClickhouseClient {}
+impl super::payment_intents::filters::PaymentIntentFilterAnalytics for ClickhouseClient {}
+impl super::payment_intents::metrics::PaymentIntentMetricAnalytics for ClickhouseClient {}
 impl super::refunds::metrics::RefundMetricAnalytics for ClickhouseClient {}
 impl super::refunds::filters::RefundFilterAnalytics for ClickhouseClient {}
 impl super::sdk_events::filters::SdkEventFilterAnalytics for ClickhouseClient {}
@@ -247,6 +250,26 @@ impl TryInto<FilterRow> for serde_json::Value {
     }
 }
 
+impl TryInto<PaymentIntentMetricRow> for serde_json::Value {
+    type Error = Report<ParsingError>;
+
+    fn try_into(self) -> Result<PaymentIntentMetricRow, Self::Error> {
+        serde_json::from_value(self).change_context(ParsingError::StructParseFailure(
+            "Failed to parse PaymentIntentMetricRow in clickhouse results",
+        ))
+    }
+}
+
+impl TryInto<PaymentIntentFilterRow> for serde_json::Value {
+    type Error = Report<ParsingError>;
+
+    fn try_into(self) -> Result<PaymentIntentFilterRow, Self::Error> {
+        serde_json::from_value(self).change_context(ParsingError::StructParseFailure(
+            "Failed to parse PaymentIntentFilterRow in clickhouse results",
+        ))
+    }
+}
+
 impl TryInto<RefundMetricRow> for serde_json::Value {
     type Error = Report<ParsingError>;
 

crates/analytics/src/core.rs

@@ -11,6 +11,11 @@ pub async fn get_domain_info(
             download_dimensions: None,
             dimensions: utils::get_payment_dimensions(),
         },
+        AnalyticsDomain::PaymentIntents => GetInfoResponse {
+            metrics: utils::get_payment_intent_metrics_info(),
+            download_dimensions: None,
+            dimensions: utils::get_payment_intent_dimensions(),
+        },
         AnalyticsDomain::Refunds => GetInfoResponse {
             metrics: utils::get_refund_metrics_info(),
             download_dimensions: None,

crates/analytics/src/lib.rs

@@ -3,6 +3,7 @@ pub mod core;
 pub mod disputes;
 pub mod errors;
 pub mod metrics;
+pub mod payment_intents;
 pub mod payments;
 mod query;
 pub mod refunds;
@@ -39,6 +40,10 @@ use api_models::analytics::{
     },
     auth_events::{AuthEventMetrics, AuthEventMetricsBucketIdentifier},
     disputes::{DisputeDimensions, DisputeFilters, DisputeMetrics, DisputeMetricsBucketIdentifier},
+    payment_intents::{
+        PaymentIntentDimensions, PaymentIntentFilters, PaymentIntentMetrics,
+        PaymentIntentMetricsBucketIdentifier,
+    },
     payments::{PaymentDimensions, PaymentFilters, PaymentMetrics, PaymentMetricsBucketIdentifier},
     refunds::{RefundDimensions, RefundFilters, RefundMetrics, RefundMetricsBucketIdentifier},
     sdk_events::{
@@ -60,6 +65,7 @@ use strum::Display;
 use self::{
     active_payments::metrics::{ActivePaymentsMetric, ActivePaymentsMetricRow},
     auth_events::metrics::{AuthEventMetric, AuthEventMetricRow},
+    payment_intents::metrics::{PaymentIntentMetric, PaymentIntentMetricRow},
     payments::{
         distribution::{PaymentDistribution, PaymentDistributionRow},
         metrics::{PaymentMetric, PaymentMetricRow},
@@ -313,6 +319,111 @@ impl AnalyticsProvider {
         .await
     }
 
+    pub async fn get_payment_intent_metrics(
+        &self,
+        metric: &PaymentIntentMetrics,
+        dimensions: &[PaymentIntentDimensions],
+        merchant_id: &str,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+    ) -> types::MetricsResult<Vec<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>>
+    {
+        // Metrics to get the fetch time for each payment intent metric
+        metrics::request::record_operation_time(
+            async {
+                match self {
+                        Self::Sqlx(pool) => {
+                        metric
+                            .load_metrics(
+                                dimensions,
+                                merchant_id,
+                                filters,
+                                granularity,
+                                time_range,
+                                pool,
+                            )
+                            .await
+                    }
+                                        Self::Clickhouse(pool) => {
+                        metric
+                            .load_metrics(
+                                dimensions,
+                                merchant_id,
+                                filters,
+                                granularity,
+                                time_range,
+                                pool,
+                            )
+                            .await
+                    }
+                                    Self::CombinedCkh(sqlx_pool, ckh_pool) => {
+                        let (ckh_result, sqlx_result) = tokio::join!(metric
+                            .load_metrics(
+                                dimensions,
+                                merchant_id,
+                                filters,
+                                granularity,
+                                time_range,
+                                ckh_pool,
+                            ),
+                            metric
+                            .load_metrics(
+                                dimensions,
+                                merchant_id,
+                                filters,
+                                granularity,
+                                time_range,
+                                sqlx_pool,
+                            ));
+                        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 payment intents analytics metrics")
+                            },
+                            _ => {}
+
+                        };
+
+                        ckh_result
+                    }
+                                    Self::CombinedSqlx(sqlx_pool, ckh_pool) => {
+                        let (ckh_result, sqlx_result) = tokio::join!(metric
+                            .load_metrics(
+                                dimensions,
+                                merchant_id,
+                                filters,
+                                granularity,
+                                time_range,
+                                ckh_pool,
+                            ),
+                            metric
+                            .load_metrics(
+                                dimensions,
+                                merchant_id,
+                                filters,
+                                granularity,
+                                time_range,
+                                sqlx_pool,
+                            ));
+                        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 payment intents analytics metrics")
+                            },
+                            _ => {}
+
+                        };
+
+                        sqlx_result
+                    }
+                }
+            },
+            &metrics::METRIC_FETCH_TIME,
+            metric,
+            self,
+        )
+        .await
+    }
+
     pub async fn get_refund_metrics(
         &self,
         metric: &RefundMetrics,
@@ -756,11 +867,13 @@ pub struct ReportConfig {
 pub enum AnalyticsFlow {
     GetInfo,
     GetPaymentMetrics,
+    GetPaymentIntentMetrics,
     GetRefundsMetrics,
     GetSdkMetrics,
     GetAuthMetrics,
     GetActivePaymentsMetrics,
     GetPaymentFilters,
+    GetPaymentIntentFilters,
     GetRefundFilters,
     GetSdkEventFilters,
     GetApiEvents,

crates/analytics/src/payment_intents.rs

@@ -0,0 +1,13 @@
+pub mod accumulator;
+mod core;
+pub mod filters;
+pub mod metrics;
+pub mod types;
+pub use accumulator::{PaymentIntentMetricAccumulator, PaymentIntentMetricsAccumulator};
+
+pub trait PaymentIntentAnalytics:
+    metrics::PaymentIntentMetricAnalytics + filters::PaymentIntentFilterAnalytics
+{
+}
+
+pub use self::core::{get_filters, get_metrics};

crates/analytics/src/payment_intents/accumulator.rs

@@ -0,0 +1,90 @@
+use api_models::analytics::payment_intents::PaymentIntentMetricsBucketValue;
+use bigdecimal::ToPrimitive;
+
+use super::metrics::PaymentIntentMetricRow;
+
+#[derive(Debug, Default)]
+pub struct PaymentIntentMetricsAccumulator {
+    pub successful_smart_retries: CountAccumulator,
+    pub total_smart_retries: CountAccumulator,
+    pub smart_retried_amount: SumAccumulator,
+    pub payment_intent_count: CountAccumulator,
+}
+
+#[derive(Debug, Default)]
+pub struct ErrorDistributionRow {
+    pub count: i64,
+    pub total: i64,
+    pub error_message: String,
+}
+
+#[derive(Debug, Default)]
+pub struct ErrorDistributionAccumulator {
+    pub error_vec: Vec<ErrorDistributionRow>,
+}
+
+#[derive(Debug, Default)]
+#[repr(transparent)]
+pub struct CountAccumulator {
+    pub count: Option<i64>,
+}
+
+pub trait PaymentIntentMetricAccumulator {
+    type MetricOutput;
+
+    fn add_metrics_bucket(&mut self, metrics: &PaymentIntentMetricRow);
+
+    fn collect(self) -> Self::MetricOutput;
+}
+
+#[derive(Debug, Default)]
+#[repr(transparent)]
+pub struct SumAccumulator {
+    pub total: Option<i64>,
+}
+
+impl PaymentIntentMetricAccumulator for CountAccumulator {
+    type MetricOutput = Option<u64>;
+    #[inline]
+    fn add_metrics_bucket(&mut self, metrics: &PaymentIntentMetricRow) {
+        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 PaymentIntentMetricAccumulator for SumAccumulator {
+    type MetricOutput = Option<u64>;
+    #[inline]
+    fn add_metrics_bucket(&mut self, metrics: &PaymentIntentMetricRow) {
+        self.total = match (
+            self.total,
+            metrics.total.as_ref().and_then(ToPrimitive::to_i64),
+        ) {
+            (None, None) => None,
+            (None, i @ Some(_)) | (i @ Some(_), None) => i,
+            (Some(a), Some(b)) => Some(a + b),
+        }
+    }
+    #[inline]
+    fn collect(self) -> Self::MetricOutput {
+        self.total.and_then(|i| u64::try_from(i).ok())
+    }
+}
+
+impl PaymentIntentMetricsAccumulator {
+    pub fn collect(self) -> PaymentIntentMetricsBucketValue {
+        PaymentIntentMetricsBucketValue {
+            successful_smart_retries: self.successful_smart_retries.collect(),
+            total_smart_retries: self.total_smart_retries.collect(),
+            smart_retried_amount: self.smart_retried_amount.collect(),
+            payment_intent_count: self.payment_intent_count.collect(),
+        }
+    }
+}