Issue:
Bug: feat(disputes): add support for disputes aggregate



1. Support for disputes aggregate 
2. Support for disputes aggregate in 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 disputes aggregate 
- Add support for profile level disputes 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 #5898 

## 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)?
-->
**Dispute Aggregate** 
Request:
```
curl --location 'http://localhost:8080/disputes/aggregate?start_time=2022-08-10T18%3A30%3A00Z' \
--header 'Authorization: Bearer JWT' \
--data ''
```

Response : 
```
{
    "status_with_count": {
        "dispute_cancelled": 0,
        "dispute_expired": 3,
        "dispute_lost": 2,
        "dispute_opened": 0,
        "dispute_accepted": 0,
        "dispute_challenged": 0,
        "dispute_won": 0
    }
}

```

**Profile level Dispute aggregate** 
Request 
```
curl --location 'http://localhost:8080/disputes/profile/aggregate?start_time=2022-08-10T18%3A30%3A00Z' \
--header 'Authorization: Bearer JWT' \
--data ''
```

Response
```
{
    "status_with_count": {
        "dispute_lost": 2,
        "dispute_accepted": 0,
        "dispute_opened": 0,
        "dispute_challenged": 0,
        "dispute_expired": 3,
        "dispute_cancelled": 0,
        "dispute_won": 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:

**Dispute Aggregate** 
Request:
```
curl --location 'http://localhost:8080/disputes/aggregate?start_time=2022-08-10T18%3A30%3A00Z' \
--header 'Authorization: Bearer JWT' \
--data ''
```

Response : 
```
{
    "status_with_count": {
        "dispute_cancelled": 0,
        "dispute_expired": 3,
        "dispute_lost": 2,
        "dispute_opened": 0,
        "dispute_accepted": 0,
        "dispute_challenged": 0,
        "dispute_won": 0
    }
}

```

**Profile level Dispute aggregate** 
Request 
```
curl --location 'http://localhost:8080/disputes/profile/aggregate?start_time=2022-08-10T18%3A30%3A00Z' \
--header 'Authorization: Bearer JWT' \
--data ''
```

Response
```
{
    "status_with_count": {
        "dispute_lost": 2,
        "dispute_accepted": 0,
        "dispute_opened": 0,
        "dispute_challenged": 0,
        "dispute_expired": 3,
        "dispute_cancelled": 0,
        "dispute_won": 0
    }
}
```

Patch:
diff --git a/crates/api_models/src/disputes.rs b/crates/api_models/src/disputes.rs
index 5c2a2cc03d2..a1340b5aa5a 100644
--- a/crates/api_models/src/disputes.rs
+++ b/crates/api_models/src/disputes.rs
@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use masking::{Deserialize, Serialize};
 use time::PrimitiveDateTime;
 use utoipa::ToSchema;
@@ -208,3 +210,9 @@ pub struct DeleteEvidenceRequest {
     /// Evidence Type to be deleted
     pub evidence_type: EvidenceType,
 }
+
+#[derive(Clone, Debug, serde::Serialize)]
+pub struct DisputesAggregateResponse {
+    /// Different status of disputes with their count
+    pub status_with_count: HashMap<DisputeStatus, i64>,
+}
diff --git a/crates/api_models/src/events/dispute.rs b/crates/api_models/src/events/dispute.rs
index 1ea6f4e8552..57f91330cc9 100644
--- a/crates/api_models/src/events/dispute.rs
+++ b/crates/api_models/src/events/dispute.rs
@@ -1,7 +1,8 @@
 use common_utils::events::{ApiEventMetric, ApiEventsType};
 
 use super::{
-    DeleteEvidenceRequest, DisputeResponse, DisputeResponsePaymentsRetrieve, SubmitEvidenceRequest,
+    DeleteEvidenceRequest, DisputeResponse, DisputeResponsePaymentsRetrieve,
+    DisputesAggregateResponse, SubmitEvidenceRequest,
 };
 
 impl ApiEventMetric for SubmitEvidenceRequest {
@@ -32,3 +33,9 @@ impl ApiEventMetric for DeleteEvidenceRequest {
         })
     }
 }
+
+impl ApiEventMetric for DisputesAggregateResponse {
+    fn get_api_event_type(&self) -> Option<ApiEventsType> {
+        Some(ApiEventsType::ResourceListAPI)
+    }
+}
diff --git a/crates/common_enums/src/enums.rs b/crates/common_enums/src/enums.rs
index 10c9dae2cc1..691b683aa3e 100644
--- a/crates/common_enums/src/enums.rs
+++ b/crates/common_enums/src/enums.rs
@@ -1832,6 +1832,7 @@ pub enum DisputeStage {
     serde::Serialize,
     strum::Display,
     strum::EnumString,
+    strum::EnumIter,
     ToSchema,
 )]
 #[router_derive::diesel_enum(storage_type = "db_enum")]
diff --git a/crates/router/src/core/disputes.rs b/crates/router/src/core/disputes.rs
index 8835f5f95a8..ff5936fc296 100644
--- a/crates/router/src/core/disputes.rs
+++ b/crates/router/src/core/disputes.rs
@@ -3,6 +3,9 @@ use common_utils::ext_traits::{Encode, ValueExt};
 use error_stack::ResultExt;
 use router_env::{instrument, tracing};
 pub mod transformers;
+use std::collections::HashMap;
+
+use strum::IntoEnumIterator;
 
 use super::{
     errors::{self, ConnectorErrorExt, RouterResponse, StorageErrorExt},
@@ -507,3 +510,31 @@ pub async fn delete_evidence(
         })?;
     Ok(services::ApplicationResponse::StatusOk)
 }
+
+#[instrument(skip(state))]
+pub async fn get_aggregates_for_disputes(
+    state: SessionState,
+    merchant: domain::MerchantAccount,
+    profile_id_list: Option<Vec<common_utils::id_type::ProfileId>>,
+    time_range: api::TimeRange,
+) -> RouterResponse<dispute_models::DisputesAggregateResponse> {
+    let db = state.store.as_ref();
+    let dispute_status_with_count = db
+        .get_dispute_status_with_count(merchant.get_id(), profile_id_list, &time_range)
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Unable to retrieve disputes aggregate")?;
+
+    let mut status_map: HashMap<storage_enums::DisputeStatus, i64> =
+        dispute_status_with_count.into_iter().collect();
+
+    for status in storage_enums::DisputeStatus::iter() {
+        status_map.entry(status).or_default();
+    }
+
+    Ok(services::ApplicationResponse::Json(
+        dispute_models::DisputesAggregateResponse {
+            status_with_count: status_map,
+        },
+    ))
+}
diff --git a/crates/router/src/db/dispute.rs b/crates/router/src/db/dispute.rs
index abe469dc32a..1d45cfd405f 100644
--- a/crates/router/src/db/dispute.rs
+++ b/crates/router/src/db/dispute.rs
@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use error_stack::report;
 use router_env::{instrument, tracing};
 
@@ -45,6 +47,13 @@ pub trait DisputeInterface {
         this: storage::Dispute,
         dispute: storage::DisputeUpdate,
     ) -> CustomResult<storage::Dispute, errors::StorageError>;
+
+    async fn get_dispute_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::enums::DisputeStatus, i64)>, errors::StorageError>;
 }
 
 #[async_trait::async_trait]
@@ -126,6 +135,24 @@ impl DisputeInterface for Store {
             .await
             .map_err(|error| report!(errors::StorageError::from(error)))
     }
+
+    #[instrument(skip_all)]
+    async fn get_dispute_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::DisputeStatus, i64)>, errors::StorageError> {
+        let conn = connection::pg_connection_read(self).await?;
+        storage::Dispute::get_dispute_status_with_count(
+            &conn,
+            merchant_id,
+            profile_id_list,
+            time_range,
+        )
+        .await
+        .map_err(|error| report!(errors::StorageError::from(error)))
+    }
 }
 
 #[async_trait::async_trait]
@@ -358,6 +385,50 @@ impl DisputeInterface for MockDb {
 
         Ok(dispute_to_update.clone())
     }
+
+    async fn get_dispute_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::DisputeStatus, i64)>, errors::StorageError> {
+        let locked_disputes = self.disputes.lock().await;
+
+        let filtered_disputes_data = locked_disputes
+            .iter()
+            .filter(|d| {
+                d.merchant_id == *merchant_id
+                    && d.created_at >= time_range.start_time
+                    && time_range
+                        .end_time
+                        .as_ref()
+                        .map(|received_end_time| received_end_time >= &d.created_at)
+                        .unwrap_or(true)
+                    && profile_id_list
+                        .as_ref()
+                        .zip(d.profile_id.as_ref())
+                        .map(|(received_profile_list, received_profile_id)| {
+                            received_profile_list.contains(received_profile_id)
+                        })
+                        .unwrap_or(true)
+            })
+            .cloned()
+            .collect::<Vec<storage::Dispute>>();
+
+        Ok(filtered_disputes_data
+            .into_iter()
+            .fold(
+                HashMap::new(),
+                |mut acc: HashMap<common_enums::DisputeStatus, i64>, value| {
+                    acc.entry(value.dispute_status)
+                        .and_modify(|value| *value += 1)
+                        .or_insert(1);
+                    acc
+                },
+            )
+            .into_iter()
+            .collect::<Vec<(common_enums::DisputeStatus, i64)>>())
+    }
 }
 
 #[cfg(test)]
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 6c620c4fcdc..3a473891feb 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -629,6 +629,17 @@ impl DisputeInterface for KafkaStore {
             .find_disputes_by_merchant_id_payment_id(merchant_id, payment_id)
             .await
     }
+
+    async fn get_dispute_status_with_count(
+        &self,
+        merchant_id: &id_type::MerchantId,
+        profile_id_list: Option<Vec<id_type::ProfileId>>,
+        time_range: &api_models::payments::TimeRange,
+    ) -> CustomResult<Vec<(common_enums::DisputeStatus, i64)>, errors::StorageError> {
+        self.diesel_store
+            .get_dispute_status_with_count(merchant_id, profile_id_list, time_range)
+            .await
+    }
 }
 
 #[async_trait::async_trait]
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index da1a50b82ee..b0fd1091296 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -1463,6 +1463,13 @@ impl Disputes {
                 web::resource("/accept/{dispute_id}")
                     .route(web::post().to(disputes::accept_dispute)),
             )
+            .service(
+                web::resource("/aggregate").route(web::get().to(disputes::get_disputes_aggregate)),
+            )
+            .service(
+                web::resource("/profile/aggregate")
+                    .route(web::get().to(disputes::get_disputes_aggregate_profile)),
+            )
             .service(
                 web::resource("/evidence")
                     .route(web::post().to(disputes::submit_dispute_evidence))
diff --git a/crates/router/src/routes/disputes.rs b/crates/router/src/routes/disputes.rs
index 98ba33a64b4..5d8f7120f40 100644
--- a/crates/router/src/routes/disputes.rs
+++ b/crates/router/src/routes/disputes.rs
@@ -11,7 +11,7 @@ use super::app::AppState;
 use crate::{
     core::disputes,
     services::{api, authentication as auth},
-    types::api::disputes as dispute_types,
+    types::api::{disputes as dispute_types, payments::TimeRange},
 };
 
 /// Disputes - Retrieve Dispute
@@ -408,3 +408,68 @@ pub async fn delete_dispute_evidence(
     ))
     .await
 }
+
+#[instrument(skip_all, fields(flow = ?Flow::DisputesAggregate))]
+pub async fn get_disputes_aggregate(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    query_param: web::Query<TimeRange>,
+) -> HttpResponse {
+    let flow = Flow::DisputesAggregate;
+    let query_param = query_param.into_inner();
+
+    Box::pin(api::server_wrap(
+        flow,
+        state,
+        &req,
+        query_param,
+        |state, auth, req, _| {
+            disputes::get_aggregates_for_disputes(state, auth.merchant_account, None, req)
+        },
+        auth::auth_type(
+            &auth::HeaderAuth(auth::ApiKeyAuth),
+            &auth::JWTAuth {
+                permission: Permission::DisputeRead,
+                minimum_entity_level: EntityType::Merchant,
+            },
+            req.headers(),
+        ),
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
+
+#[instrument(skip_all, fields(flow = ?Flow::DisputesAggregate))]
+pub async fn get_disputes_aggregate_profile(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    query_param: web::Query<TimeRange>,
+) -> HttpResponse {
+    let flow = Flow::DisputesAggregate;
+    let query_param = query_param.into_inner();
+
+    Box::pin(api::server_wrap(
+        flow,
+        state,
+        &req,
+        query_param,
+        |state, auth, req, _| {
+            disputes::get_aggregates_for_disputes(
+                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::DisputeRead,
+                minimum_entity_level: EntityType::Profile,
+            },
+            req.headers(),
+        ),
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
diff --git a/crates/router/src/routes/lock_utils.rs b/crates/router/src/routes/lock_utils.rs
index e63b00cd9bf..774e6c75bb4 100644
--- a/crates/router/src/routes/lock_utils.rs
+++ b/crates/router/src/routes/lock_utils.rs
@@ -175,6 +175,7 @@ impl From<Flow> for ApiIdentifier {
             | Flow::DisputesEvidenceSubmit
             | Flow::AttachDisputeEvidence
             | Flow::RetrieveDisputeEvidence
+            | Flow::DisputesAggregate
             | Flow::DeleteDisputeEvidence => Self::Disputes,
 
             Flow::CardsInfo => Self::CardsInfo,
diff --git a/crates/router/src/types/storage/dispute.rs b/crates/router/src/types/storage/dispute.rs
index 2728b29bc07..79b9a90343e 100644
--- a/crates/router/src/types/storage/dispute.rs
+++ b/crates/router/src/types/storage/dispute.rs
@@ -14,6 +14,13 @@ pub trait DisputeDbExt: Sized {
         merchant_id: &common_utils::id_type::MerchantId,
         dispute_list_constraints: api_models::disputes::DisputeListConstraints,
     ) -> CustomResult<Vec<Self>, errors::DatabaseError>;
+
+    async fn get_dispute_status_with_count(
+        conn: &PgPooledConn,
+        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::enums::DisputeStatus, i64)>, errors::DatabaseError>;
 }
 
 #[async_trait::async_trait]
@@ -72,4 +79,38 @@ impl DisputeDbExt for Dispute {
         .change_context(errors::DatabaseError::NotFound)
         .attach_printable_lazy(|| "Error filtering records by predicate")
     }
+
+    async fn get_dispute_status_with_count(
+        conn: &PgPooledConn,
+        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::DisputeStatus, i64)>, errors::DatabaseError> {
+        let mut query = <Self as HasTable>::table()
+            .group_by(dsl::dispute_status)
+            .select((dsl::dispute_status, diesel::dsl::count_star()))
+            .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 {
+            Some(ending_at) => query.filter(dsl::created_at.le(ending_at)),
+            None => query,
+        };
+
+        logger::debug!(query = %diesel::debug_query::<diesel::pg::Pg,_>(&query).to_string());
+
+        db_metrics::track_database_call::<<Self as HasTable>::Table, _, _>(
+            query.get_results_async::<(common_enums::DisputeStatus, i64)>(conn),
+            db_metrics::DatabaseOperation::Count,
+        )
+        .await
+        .change_context(errors::DatabaseError::NotFound)
+        .attach_printable_lazy(|| "Error filtering records by predicate")
+    }
 }
diff --git a/crates/router_env/src/logger/types.rs b/crates/router_env/src/logger/types.rs
index d815325a814..310f5ccecbf 100644
--- a/crates/router_env/src/logger/types.rs
+++ b/crates/router_env/src/logger/types.rs
@@ -290,6 +290,8 @@ pub enum Flow {
     AttachDisputeEvidence,
     /// Delete Dispute Evidence flow
     DeleteDisputeEvidence,
+    /// Disputes aggregate flow
+    DisputesAggregate,
     /// Retrieve Dispute Evidence flow
     RetrieveDisputeEvidence,
     /// Invalidate cache flow