Issue:
Bug: feat(analytics): add customer_id as filter for payment intents



Add customer_id as a filter for payment intents

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
- Added `customer_id` as a filter for payment intents
- Added `payment_processed_amount` metric for `payment_intents` table 
### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->
Filter out payments for a particular customer through the `customer_id`

## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->
Hit the curl:
Metric data would be obtained based on the customer_id filter applied.
```bash
curl --location 'http://localhost:8080/analytics/v2/org/metrics/payments' \
--header 'Accept: */*' \
--header 'Accept-Language: en-US,en;q=0.9' \
--header 'Connection: keep-alive' \
--header 'Content-Type: application/json' \
--header 'Origin: http://localhost:9000' \
--header 'QueryType: SingleStatTimeseries' \
--header 'Referer: http://localhost:9000/' \
--header 'Sec-Fetch-Dest: empty' \
--header 'Sec-Fetch-Mode: cors' \
--header 'Sec-Fetch-Site: same-site' \
--header 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36' \
--header 'api-key: dev_LZcIA7XeMpnK5satp4CDvjcnHCaeKTBcosyuBBJaknZu1odhFo96cwS0nSdfzuJF' \
--header 'authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiNTQ5ZTNkMmItMTY5Yi00NzUzLWJmNTQtZDcxMTM2YjRiN2JkIiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNzI2MDQ2MzI4Iiwicm9sZV9pZCI6Im9yZ19hZG1pbiIsImV4cCI6MTcyOTI0ODM1MCwib3JnX2lkIjoib3JnX1ZwU0hPanNZZkR2YWJWWUpnQ0FKIiwicHJvZmlsZV9pZCI6InByb192NXNGb0hlODBPZWlVbElvbm9jTSJ9.TQD6wYkiTXlTvPW3-mH8FzEP1d7hXw9SmWgjVMH7flk' \
--header 'sec-ch-ua: "Chromium";v="128", "Not;A=Brand";v="24", "Google Chrome";v="128"' \
--header 'sec-ch-ua-mobile: ?0' \
--header 'sec-ch-ua-platform: "macOS"' \
--data '[
    {
        "timeRange": {
            "startTime": "2024-09-11T00:00:00Z",
            "endTime": "2024-10-26T09:22:30Z"
        },
        "filters": {
            "customer_id": ["StripeCustomer"]
        },
        "mode": "ORDER",
        "source": "BATCH",
        "metrics": [
            "payment_processed_amount"
        ]
    }
]'
```

## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible


Testing:

Hit the curl:
Metric data would be obtained based on the customer_id filter applied.
```bash
curl --location 'http://localhost:8080/analytics/v2/org/metrics/payments' \
--header 'Accept: */*' \
--header 'Accept-Language: en-US,en;q=0.9' \
--header 'Connection: keep-alive' \
--header 'Content-Type: application/json' \
--header 'Origin: http://localhost:9000' \
--header 'QueryType: SingleStatTimeseries' \
--header 'Referer: http://localhost:9000/' \
--header 'Sec-Fetch-Dest: empty' \
--header 'Sec-Fetch-Mode: cors' \
--header 'Sec-Fetch-Site: same-site' \
--header 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36' \
--header 'api-key: dev_LZcIA7XeMpnK5satp4CDvjcnHCaeKTBcosyuBBJaknZu1odhFo96cwS0nSdfzuJF' \
--header 'authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiNTQ5ZTNkMmItMTY5Yi00NzUzLWJmNTQtZDcxMTM2YjRiN2JkIiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNzI2MDQ2MzI4Iiwicm9sZV9pZCI6Im9yZ19hZG1pbiIsImV4cCI6MTcyOTI0ODM1MCwib3JnX2lkIjoib3JnX1ZwU0hPanNZZkR2YWJWWUpnQ0FKIiwicHJvZmlsZV9pZCI6InByb192NXNGb0hlODBPZWlVbElvbm9jTSJ9.TQD6wYkiTXlTvPW3-mH8FzEP1d7hXw9SmWgjVMH7flk' \
--header 'sec-ch-ua: "Chromium";v="128", "Not;A=Brand";v="24", "Google Chrome";v="128"' \
--header 'sec-ch-ua-mobile: ?0' \
--header 'sec-ch-ua-platform: "macOS"' \
--data '[
    {
        "timeRange": {
            "startTime": "2024-09-11T00:00:00Z",
            "endTime": "2024-10-26T09:22:30Z"
        },
        "filters": {
            "customer_id": ["StripeCustomer"]
        },
        "mode": "ORDER",
        "source": "BATCH",
        "metrics": [
            "payment_processed_amount"
        ]
    }
]'
```

Patch:
diff --git a/crates/analytics/src/payment_intents/core.rs b/crates/analytics/src/payment_intents/core.rs
index af46baed23e..3e8915c60a2 100644
--- a/crates/analytics/src/payment_intents/core.rs
+++ b/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
diff --git a/crates/analytics/src/payment_intents/filters.rs b/crates/analytics/src/payment_intents/filters.rs
index e81b050214c..d03d6c2a15f 100644
--- a/crates/analytics/src/payment_intents/filters.rs
+++ b/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>,
 }
diff --git a/crates/analytics/src/payment_intents/metrics.rs b/crates/analytics/src/payment_intents/metrics.rs
index e9d7f244306..9aa7d3e9771 100644
--- a/crates/analytics/src/payment_intents/metrics.rs
+++ b/crates/analytics/src/payment_intents/metrics.rs
@@ -17,6 +17,7 @@ use crate::{
 };
 
 mod payment_intent_count;
+mod payment_processed_amount;
 mod payments_success_rate;
 mod sessionized_metrics;
 mod smart_retried_amount;
@@ -24,6 +25,7 @@ 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)
diff --git a/crates/analytics/src/payment_intents/metrics/payment_processed_amount.rs b/crates/analytics/src/payment_intents/metrics/payment_processed_amount.rs
new file mode 100644
index 00000000000..51b574f4ad3
--- /dev/null
+++ b/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)
+    }
+}
diff --git a/crates/analytics/src/payment_intents/types.rs b/crates/analytics/src/payment_intents/types.rs
index 03f2a196c20..bb5141297c5 100644
--- a/crates/analytics/src/payment_intents/types.rs
+++ b/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(())
     }
 }
diff --git a/crates/analytics/src/query.rs b/crates/analytics/src/query.rs
index 7ce338f7db3..d746594e36e 100644
--- a/crates/analytics/src/query.rs
+++ b/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),+) => {
diff --git a/crates/analytics/src/sqlx.rs b/crates/analytics/src/sqlx.rs
index 89eb2ee0bde..7c90e37c55f 100644
--- a/crates/analytics/src/sqlx.rs
+++ b/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,
         })
     }
 }
diff --git a/crates/api_models/src/analytics/payment_intents.rs b/crates/api_models/src/analytics/payment_intents.rs
index d018437ae8c..60662f2e90a 100644
--- a/crates/api_models/src/analytics/payment_intents.rs
+++ b/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,