feat(payments): add support for profile aggregates

crates/hyperswitch_domain_models/src/payments/payment_intent.rs

@@ -71,6 +71,7 @@ pub trait PaymentIntentInterface {
     async fn get_intent_status_with_count(
         &self,
         merchant_id: &id_type::MerchantId,
+        profile_id_list: Option<&Vec<id_type::ProfileId>>,
         constraints: &api_models::payments::TimeRange,
     ) -> error_stack::Result<Vec<(common_enums::IntentStatus, i64)>, errors::StorageError>;
 

crates/router/src/core/payments.rs

@@ -3182,11 +3182,12 @@ pub async fn get_payment_filters(
 pub async fn get_aggregates_for_payments(
     state: SessionState,
     merchant: domain::MerchantAccount,
+    profile_id_list: Option<Vec<id_type::ProfileId>>,
     time_range: api::TimeRange,
 ) -> RouterResponse<api::PaymentsAggregateResponse> {
     let db = state.store.as_ref();
     let intent_status_with_count = db
-        .get_intent_status_with_count(merchant.get_id(), &time_range)
+        .get_intent_status_with_count(merchant.get_id(), profile_id_list.as_ref(), &time_range)
         .await
         .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
 

crates/router/src/db/kafka_store.rs

@@ -1640,10 +1640,11 @@ impl PaymentIntentInterface for KafkaStore {
     async fn get_intent_status_with_count(
         &self,
         merchant_id: &id_type::MerchantId,
+        profile_id_list: Option<&Vec<id_type::ProfileId>>,
         time_range: &api_models::payments::TimeRange,
     ) -> error_stack::Result<Vec<(common_enums::IntentStatus, i64)>, errors::DataStorageError> {
         self.diesel_store
-            .get_intent_status_with_count(merchant_id, time_range)
+            .get_intent_status_with_count(merchant_id, profile_id_list, time_range)
             .await
     }
 

crates/router/src/routes/app.rs

@@ -547,6 +547,10 @@ impl Payments {
                 .service(web::resource("/filter").route(web::post().to(get_filters_for_payments)))
                 .service(web::resource("/v2/filter").route(web::get().to(get_payment_filters)))
                 .service(web::resource("/aggregate").route(web::get().to(get_payments_aggregates)))
+                .service(
+                    web::resource("/profile/aggregate")
+                        .route(web::get().to(get_payments_aggregates_profile)),
+                )
                 .service(
                     web::resource("/v2/profile/filter")
                         .route(web::get().to(get_payment_filters_profile)),

crates/router/src/routes/payments.rs

@@ -1310,7 +1310,7 @@ pub async fn get_payments_aggregates(
         &req,
         payload,
         |state, auth: auth::AuthenticationData, req, _| {
-            payments::get_aggregates_for_payments(state, auth.merchant_account, req)
+            payments::get_aggregates_for_payments(state, auth.merchant_account, None, req)
         },
         &auth::JWTAuth {
             permission: Permission::PaymentRead,
@@ -2020,3 +2020,34 @@ impl GetLockingInput for payment_types::PaymentsManualUpdateRequest {
         }
     }
 }
+
+#[instrument(skip_all, fields(flow = ?Flow::PaymentsAggregate))]
+#[cfg(feature = "olap")]
+pub async fn get_payments_aggregates_profile(
+    state: web::Data<app::AppState>,
+    req: actix_web::HttpRequest,
+    payload: web::Query<payment_types::TimeRange>,
+) -> impl Responder {
+    let flow = Flow::PaymentsAggregate;
+    let payload = payload.into_inner();
+    Box::pin(api::server_wrap(
+        flow,
+        state,
+        &req,
+        payload,
+        |state, auth: auth::AuthenticationData, req, _| {
+            payments::get_aggregates_for_payments(
+                state,
+                auth.merchant_account,
+                auth.profile_id.map(|profile_id| vec![profile_id]),
+                req,
+            )
+        },
+        &auth::JWTAuth {
+            permission: Permission::PaymentRead,
+            minimum_entity_level: EntityType::Profile,
+        },
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}

crates/storage_impl/src/mock_db/payment_intent.rs

@@ -44,6 +44,7 @@ impl PaymentIntentInterface for MockDb {
     async fn get_intent_status_with_count(
         &self,
         _merchant_id: &common_utils::id_type::MerchantId,
+        _profile_id_list: Option<&Vec<common_utils::id_type::ProfileId>>,
         _time_range: &api_models::payments::TimeRange,
     ) -> CustomResult<Vec<(common_enums::IntentStatus, i64)>, StorageError> {
         // [#172]: Implement function for `MockDb`

crates/storage_impl/src/payments/payment_intent.rs

@@ -350,10 +350,11 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
     async fn get_intent_status_with_count(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
+        profile_id_list: Option<&Vec<common_utils::id_type::ProfileId>>,
         time_range: &api_models::payments::TimeRange,
     ) -> error_stack::Result<Vec<(common_enums::IntentStatus, i64)>, StorageError> {
         self.router_store
-            .get_intent_status_with_count(merchant_id, time_range)
+            .get_intent_status_with_count(merchant_id, profile_id_list, time_range)
             .await
     }
 
@@ -670,6 +671,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
     async fn get_intent_status_with_count(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
+        profile_id_list: Option<&Vec<common_utils::id_type::ProfileId>>,
         time_range: &api_models::payments::TimeRange,
     ) -> error_stack::Result<Vec<(common_enums::IntentStatus, i64)>, StorageError> {
         let conn = connection::pg_connection_read(self).await.switch()?;
@@ -681,6 +683,10 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
             .filter(pi_dsl::merchant_id.eq(merchant_id.to_owned()))
             .into_boxed();
 
+        if let Some(profile_id) = profile_id_list {
+            query = query.filter(pi_dsl::profile_id.eq_any(profile_id.clone()));
+        }
+
         query = query.filter(pi_dsl::created_at.ge(time_range.start_time));
 
         query = match time_range.end_time {