Issue:
Bug: feat(refunds) : profile level refunds aggregate



Support for disputes aggregate at profile level

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 refunds aggregate
- 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 


## 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/refunds/profile/aggregate?start_time=2022-08-10T18%3A30%3A00Z' \
--header 'Authorization: Bearer JWT' \
--data ''
```

Response 
```
{
    "status_with_count": {
        "success": 3,
        "pending": 0,
        "failure": 0,
        "manual_review": 0,
        "transaction_failure": 0
    }
}
```


## Checklist
<!-- Put an `x` in the boxes that apply -->

- [ ] I formatted the code `cargo +nightly fmt --all`
- [ ] 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/refunds/profile/aggregate?start_time=2022-08-10T18%3A30%3A00Z' \
--header 'Authorization: Bearer JWT' \
--data ''
```

Response 
```
{
    "status_with_count": {
        "success": 3,
        "pending": 0,
        "failure": 0,
        "manual_review": 0,
        "transaction_failure": 0
    }
}
```

Patch:
diff --git a/crates/router/src/core/refunds.rs b/crates/router/src/core/refunds.rs
index 118f4b81766..0f3dfa44c10 100644
--- a/crates/router/src/core/refunds.rs
+++ b/crates/router/src/core/refunds.rs
@@ -1098,11 +1098,17 @@ pub async fn get_filters_for_refunds(
 pub async fn get_aggregates_for_refunds(
     state: SessionState,
     merchant: domain::MerchantAccount,
+    profile_id_list: Option<Vec<common_utils::id_type::ProfileId>>,
     time_range: common_utils::types::TimeRange,
 ) -> RouterResponse<api_models::refunds::RefundAggregateResponse> {
     let db = state.store.as_ref();
     let refund_status_with_count = db
-        .get_refund_status_with_count(merchant.get_id(), &time_range, merchant.storage_scheme)
+        .get_refund_status_with_count(
+            merchant.get_id(),
+            profile_id_list,
+            &time_range,
+            merchant.storage_scheme,
+        )
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Failed to find status count")?;
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index ff7e8145c33..0190718791f 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -2546,11 +2546,12 @@ impl RefundInterface for KafkaStore {
     async fn get_refund_status_with_count(
         &self,
         merchant_id: &id_type::MerchantId,
+        profile_id_list: Option<Vec<id_type::ProfileId>>,
         constraints: &common_utils::types::TimeRange,
         storage_scheme: MerchantStorageScheme,
     ) -> CustomResult<Vec<(common_enums::RefundStatus, i64)>, errors::StorageError> {
         self.diesel_store
-            .get_refund_status_with_count(merchant_id, constraints, storage_scheme)
+            .get_refund_status_with_count(merchant_id, profile_id_list, constraints, storage_scheme)
             .await
     }
 
diff --git a/crates/router/src/db/refund.rs b/crates/router/src/db/refund.rs
index dbe7534ee28..cf092ad410e 100644
--- a/crates/router/src/db/refund.rs
+++ b/crates/router/src/db/refund.rs
@@ -88,6 +88,7 @@ pub trait RefundInterface {
     async fn get_refund_status_with_count(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
+        profile_id_list: Option<Vec<common_utils::id_type::ProfileId>>,
         constraints: &common_utils::types::TimeRange,
         storage_scheme: enums::MerchantStorageScheme,
     ) -> CustomResult<Vec<(common_enums::RefundStatus, i64)>, errors::StorageError>;
@@ -265,11 +266,12 @@ mod storage {
         async fn get_refund_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,
             _storage_scheme: enums::MerchantStorageScheme,
         ) -> CustomResult<Vec<(common_enums::RefundStatus, i64)>, errors::StorageError> {
             let conn = connection::pg_connection_read(self).await?;
-            <diesel_models::refund::Refund as storage_types::RefundDbExt>::get_refund_status_with_count(&conn, merchant_id, time_range)
+            <diesel_models::refund::Refund as storage_types::RefundDbExt>::get_refund_status_with_count(&conn, merchant_id,profile_id_list, time_range)
             .await
             .map_err(|error|report!(errors::StorageError::from(error)))
         }
@@ -831,11 +833,12 @@ mod storage {
         async fn get_refund_status_with_count(
             &self,
             merchant_id: &common_utils::id_type::MerchantId,
+            profile_id_list: Option<Vec<common_utils::id_type::ProfileId>>,
             constraints: &common_utils::types::TimeRange,
             _storage_scheme: enums::MerchantStorageScheme,
         ) -> CustomResult<Vec<(common_enums::RefundStatus, i64)>, errors::StorageError> {
             let conn = connection::pg_connection_read(self).await?;
-            <diesel_models::refund::Refund as storage_types::RefundDbExt>::get_refund_status_with_count(&conn, merchant_id, constraints)
+            <diesel_models::refund::Refund as storage_types::RefundDbExt>::get_refund_status_with_count(&conn, merchant_id,profile_id_list, constraints)
             .await
             .map_err(|error| report!(errors::StorageError::from(error)))
         }
@@ -1182,6 +1185,7 @@ impl RefundInterface for MockDb {
     async fn get_refund_status_with_count(
         &self,
         _merchant_id: &common_utils::id_type::MerchantId,
+        profile_id_list: Option<Vec<common_utils::id_type::ProfileId>>,
         time_range: &common_utils::types::TimeRange,
         _storage_scheme: enums::MerchantStorageScheme,
     ) -> CustomResult<Vec<(api_models::enums::RefundStatus, i64)>, errors::StorageError> {
@@ -1194,7 +1198,17 @@ impl RefundInterface for MockDb {
 
         let filtered_refunds = refunds
             .iter()
-            .filter(|refund| refund.created_at >= start_time && refund.created_at <= end_time)
+            .filter(|refund| {
+                refund.created_at >= start_time
+                    && refund.created_at <= end_time
+                    && profile_id_list
+                        .as_ref()
+                        .zip(refund.profile_id.as_ref())
+                        .map(|(received_profile_list, received_profile_id)| {
+                            received_profile_list.contains(received_profile_id)
+                        })
+                        .unwrap_or(true)
+            })
             .cloned()
             .collect::<Vec<diesel_models::refund::Refund>>();
 
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 01a48c56401..a8a7ea7d56f 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -1002,6 +1002,10 @@ impl Refunds {
                 .service(web::resource("/filter").route(web::post().to(refunds_filter_list)))
                 .service(web::resource("/v2/filter").route(web::get().to(get_refunds_filters)))
                 .service(web::resource("/aggregate").route(web::get().to(get_refunds_aggregates)))
+                .service(
+                    web::resource("/profile/aggregate")
+                        .route(web::get().to(get_refunds_aggregate_profile)),
+                )
                 .service(
                     web::resource("/v2/profile/filter")
                         .route(web::get().to(get_refunds_filters_profile)),
diff --git a/crates/router/src/routes/refunds.rs b/crates/router/src/routes/refunds.rs
index 40184c5fcfa..aaea6d4c99e 100644
--- a/crates/router/src/routes/refunds.rs
+++ b/crates/router/src/routes/refunds.rs
@@ -436,7 +436,7 @@ pub async fn get_refunds_aggregates(
         &req,
         query_params,
         |state, auth: auth::AuthenticationData, req, _| {
-            get_aggregates_for_refunds(state, auth.merchant_account, req)
+            get_aggregates_for_refunds(state, auth.merchant_account, None, req)
         },
         auth::auth_type(
             &auth::HeaderAuth(auth::ApiKeyAuth),
@@ -473,3 +473,38 @@ pub async fn refunds_manual_update(
     ))
     .await
 }
+
+#[instrument(skip_all, fields(flow = ?Flow::RefundsAggregate))]
+#[cfg(feature = "olap")]
+pub async fn get_refunds_aggregate_profile(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    query_params: web::Query<common_utils::types::TimeRange>,
+) -> HttpResponse {
+    let flow = Flow::RefundsAggregate;
+    let query_params = query_params.into_inner();
+    Box::pin(api::server_wrap(
+        flow,
+        state,
+        &req,
+        query_params,
+        |state, auth: auth::AuthenticationData, req, _| {
+            get_aggregates_for_refunds(
+                state,
+                auth.merchant_account,
+                auth.profile_id.map(|profile_id| vec![profile_id]),
+                req,
+            )
+        },
+        auth::auth_type(
+            &auth::HeaderAuth(auth::ApiKeyAuth),
+            &auth::JWTAuth {
+                permission: Permission::RefundRead,
+                minimum_entity_level: EntityType::Profile,
+            },
+            req.headers(),
+        ),
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
diff --git a/crates/router/src/types/storage/refund.rs b/crates/router/src/types/storage/refund.rs
index 350c1032261..8076c2c7a6e 100644
--- a/crates/router/src/types/storage/refund.rs
+++ b/crates/router/src/types/storage/refund.rs
@@ -41,6 +41,7 @@ pub trait RefundDbExt: Sized {
     async fn get_refund_status_with_count(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
+        profile_id_list: Option<Vec<common_utils::id_type::ProfileId>>,
         time_range: &common_utils::types::TimeRange,
     ) -> CustomResult<Vec<(RefundStatus, i64)>, errors::DatabaseError>;
 }
@@ -299,6 +300,7 @@ impl RefundDbExt for Refund {
     async fn get_refund_status_with_count(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
+        profile_id_list: Option<Vec<common_utils::id_type::ProfileId>>,
         time_range: &common_utils::types::TimeRange,
     ) -> CustomResult<Vec<(RefundStatus, i64)>, errors::DatabaseError> {
         let mut query = <Self as HasTable>::table()
@@ -307,6 +309,10 @@ impl RefundDbExt for Refund {
             .filter(dsl::merchant_id.eq(merchant_id.to_owned()))
             .into_boxed();
 
+        if let Some(profile_id) = profile_id_list {
+            query = query.filter(dsl::profile_id.eq_any(profile_id));
+        }
+
         query = query.filter(dsl::created_at.ge(time_range.start_time));
 
         query = match time_range.end_time {