feat(analytics): add customer_id as filter for payment intents (#6344)

crates/analytics/src/payment_intents/core.rs

@@ -205,7 +205,8 @@ pub async fn get_metrics(
                         | PaymentIntentMetrics::SessionizedPaymentsSuccessRate => metrics_builder
                             .payments_success_rate
                             .add_metrics_bucket(&value),
-                        PaymentIntentMetrics::SessionizedPaymentProcessedAmount => metrics_builder
+                        PaymentIntentMetrics::SessionizedPaymentProcessedAmount
+                        | PaymentIntentMetrics::PaymentProcessedAmount => metrics_builder
                             .payment_processed_amount
                             .add_metrics_bucket(&value),
                         PaymentIntentMetrics::SessionizedPaymentsDistribution => metrics_builder

crates/analytics/src/payment_intents/filters.rs

@@ -54,4 +54,5 @@ pub struct PaymentIntentFilterRow {
     pub status: Option<DBEnumWrapper<IntentStatus>>,
     pub currency: Option<DBEnumWrapper<Currency>>,
     pub profile_id: Option<String>,
+    pub customer_id: Option<String>,
 }

crates/analytics/src/payment_intents/metrics.rs

@@ -17,13 +17,15 @@ use crate::{
 };
 
 mod payment_intent_count;
+mod payment_processed_amount;
 mod payments_success_rate;
 mod sessionized_metrics;
 mod smart_retried_amount;
 mod successful_smart_retries;
 mod total_smart_retries;
 
 use payment_intent_count::PaymentIntentCount;
+use payment_processed_amount::PaymentProcessedAmount;
 use payments_success_rate::PaymentsSuccessRate;
 use smart_retried_amount::SmartRetriedAmount;
 use successful_smart_retries::SuccessfulSmartRetries;
@@ -107,6 +109,11 @@ where
                     .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
                     .await
             }
+            Self::PaymentProcessedAmount => {
+                PaymentProcessedAmount
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
             Self::SessionizedSuccessfulSmartRetries => {
                 sessionized_metrics::SuccessfulSmartRetries
                     .load_metrics(dimensions, auth, filters, granularity, time_range, pool)

crates/analytics/src/payment_intents/metrics/payment_processed_amount.rs

@@ -0,0 +1,161 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payment_intents::{
+        PaymentIntentDimensions, PaymentIntentFilters, PaymentIntentMetricsBucketIdentifier,
+    },
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use diesel_models::enums as storage_enums;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentIntentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(super) struct PaymentProcessedAmount;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentIntentMetric<T> for PaymentProcessedAmount
+where
+    T: AnalyticsDataSource + super::PaymentIntentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentIntentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>>
+    {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentIntent);
+
+        let mut dimensions = dimensions.to_vec();
+
+        dimensions.push(PaymentIntentDimensions::PaymentIntentStatus);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+
+        query_builder
+            .add_select_column("attempt_count == 1 as first_attempt")
+            .switch()?;
+
+        query_builder.add_select_column("currency").switch()?;
+
+        query_builder
+            .add_select_column(Aggregate::Sum {
+                field: "amount",
+                alias: Some("total"),
+            })
+            .switch()?;
+
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+
+        query_builder
+            .add_group_by_clause("attempt_count")
+            .attach_printable("Error grouping by attempt_count")
+            .switch()?;
+
+        query_builder
+            .add_group_by_clause("currency")
+            .attach_printable("Error grouping by currency")
+            .switch()?;
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .add_filter_clause(
+                PaymentIntentDimensions::PaymentIntentStatus,
+                storage_enums::IntentStatus::Succeeded,
+            )
+            .switch()?;
+
+        query_builder
+            .execute_query::<PaymentIntentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentIntentMetricsBucketIdentifier::new(
+                        None,
+                        i.currency.as_ref().map(|i| i.0),
+                        i.profile_id.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}

crates/analytics/src/payment_intents/types.rs

@@ -30,6 +30,11 @@ where
                 .add_filter_in_range_clause(PaymentIntentDimensions::ProfileId, &self.profile_id)
                 .attach_printable("Error adding profile id filter")?;
         }
+        if !self.customer_id.is_empty() {
+            builder
+                .add_filter_in_range_clause("customer_id", &self.customer_id)
+                .attach_printable("Error adding customer id filter")?;
+        }
         Ok(())
     }
 }

crates/analytics/src/query.rs

@@ -451,6 +451,12 @@ impl<T: AnalyticsDataSource> ToSql<T> for &common_utils::id_type::PaymentId {
     }
 }
 
+impl<T: AnalyticsDataSource> ToSql<T> for common_utils::id_type::CustomerId {
+    fn to_sql(&self, _table_engine: &TableEngine) -> error_stack::Result<String, ParsingError> {
+        Ok(self.get_string_repr().to_owned())
+    }
+}
+
 /// Implement `ToSql` on arrays of types that impl `ToString`.
 macro_rules! impl_to_sql_for_to_string {
     ($($type:ty),+) => {

crates/analytics/src/sqlx.rs

@@ -652,10 +652,15 @@ impl<'a> FromRow<'a, PgRow> for super::payment_intents::filters::PaymentIntentFi
             ColumnNotFound(_) => Ok(Default::default()),
             e => Err(e),
         })?;
+        let customer_id: Option<String> = row.try_get("customer_id").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
         Ok(Self {
             status,
             currency,
             profile_id,
+            customer_id,
         })
     }
 }

crates/api_models/src/analytics/payment_intents.rs

@@ -16,6 +16,8 @@ pub struct PaymentIntentFilters {
     pub currency: Vec<Currency>,
     #[serde(default)]
     pub profile_id: Vec<id_type::ProfileId>,
+    #[serde(default)]
+    pub customer_id: Vec<id_type::CustomerId>,
 }
 
 #[derive(
@@ -62,6 +64,7 @@ pub enum PaymentIntentMetrics {
     SmartRetriedAmount,
     PaymentIntentCount,
     PaymentsSuccessRate,
+    PaymentProcessedAmount,
     SessionizedSuccessfulSmartRetries,
     SessionizedTotalSmartRetries,
     SessionizedSmartRetriedAmount,