Issue:
Bug: feat(payments): add support for profile aggregates



Show count of aggregate payments status in the given time range for a particular profile 

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 -->
- Add support for profile level aggregates in payments.
- For now it will have list of intent status along with the their count for a given time range.

### Additional Changes

- [X] 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).
-->
Closes #[#5846](https://github.com/juspay/hyperswitch/issues/5846)

## 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)?
-->
Request : 
```
curl --location 'http://localhost:8080/payments/profile/aggregate?start_time=2022-08-10T18%3A30%3A00Z' \
--header 'Authorization: Bearer JWT token' \
--data ''
```

Response : 
```
{
    "status_with_count": {
        "failed": 5,
        "cancelled": 0,
        "processing": 0,
        "requires_capture": 0,
        "partially_captured_and_capturable": 0,
        "requires_merchant_action": 0,
        "requires_payment_method": 0,
        "requires_customer_action": 0,
        "requires_confirmation": 0,
        "succeeded": 40,
        "partially_captured": 0
    }
}
```

## 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`
- [ ] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible


Testing:

Request : 
```
curl --location 'http://localhost:8080/payments/profile/aggregate?start_time=2022-08-10T18%3A30%3A00Z' \
--header 'Authorization: Bearer JWT token' \
--data ''
```

Response : 
```
{
    "status_with_count": {
        "failed": 5,
        "cancelled": 0,
        "processing": 0,
        "requires_capture": 0,
        "partially_captured_and_capturable": 0,
        "requires_merchant_action": 0,
        "requires_payment_method": 0,
        "requires_customer_action": 0,
        "requires_confirmation": 0,
        "succeeded": 40,
        "partially_captured": 0
    }
}
```

Patch:
diff --git a/crates/hyperswitch_domain_models/src/payments/payment_intent.rs b/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
index e5b0e8b9120..62fb06fd522 100644
--- a/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
+++ b/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>;
 
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index f94a5a4633b..4eedb52c99d 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -3234,11 +3234,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, &time_range)
         .await
         .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
 
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 2088177754e..76111afc6bd 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/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
     }
 
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 4f8e269dff9..210ded9c25d 100644
--- a/crates/router/src/routes/app.rs
+++ b/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)),
diff --git a/crates/router/src/routes/payments.rs b/crates/router/src/routes/payments.rs
index 18460d6329e..8f3ee3dd4bc 100644
--- a/crates/router/src/routes/payments.rs
+++ b/crates/router/src/routes/payments.rs
@@ -1341,7 +1341,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,
@@ -2074,3 +2074,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
+}
diff --git a/crates/storage_impl/src/mock_db/payment_intent.rs b/crates/storage_impl/src/mock_db/payment_intent.rs
index f5db60da3fa..e296c88b3ac 100644
--- a/crates/storage_impl/src/mock_db/payment_intent.rs
+++ b/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`
diff --git a/crates/storage_impl/src/payments/payment_intent.rs b/crates/storage_impl/src/payments/payment_intent.rs
index f69c26e6409..90395e15dac 100644
--- a/crates/storage_impl/src/payments/payment_intent.rs
+++ b/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));
+        }
+
         query = query.filter(pi_dsl::created_at.ge(time_range.start_time));
 
         query = match time_range.end_time {