Issue:
Bug: feat(payments): show aggregate count of payment status



Show count of aggregate payments status in the given time range:


![Image](https://github.com/user-attachments/assets/7510b765-03e7-4469-872d-f61f6748312c)


This can be part of filter Api.

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
![image](https://github.com/user-attachments/assets/9fea2624-b844-4318-8595-d35908d1e993)
- The endpoint `/aggregate` to support views
![image](https://github.com/user-attachments/assets/f3cc252d-6e5d-410b-8a8d-162cb10418aa)

- Add support for aggregates in payments.
- For now it will have list of intent status along with the their count for a given time range.

( Profile level support will be covered in separate PR, with new route but same core function having profile_id as option)
### 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
Closes [#5627](https://github.com/juspay/hyperswitch/issues/5627)


## How did you test it?
Request:
```
curl --location 'http://localhost:8080/payments/aggregate?start_time=2022-08-20T11%3A33%3A39.000Z' \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer JWT'
```

Response:
```
{
    "status_with_count": {
        "requires_capture": 0,
        "requires_customer_action": 0,
        "requires_payment_method": 0,
        "processing": 0,
        "cancelled": 0,
        "requires_confirmation": 0,
        "succeeded": 90,
        "partially_captured": 0,
        "partially_captured_and_capturable": 0,
        "requires_merchant_action": 0,
        "failed": 10
    }
}
```


## 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:
Request:
```
curl --location 'http://localhost:8080/payments/aggregate?start_time=2022-08-20T11%3A33%3A39.000Z' \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer JWT'
```

Response:
```
{
    "status_with_count": {
        "requires_capture": 0,
        "requires_customer_action": 0,
        "requires_payment_method": 0,
        "processing": 0,
        "cancelled": 0,
        "requires_confirmation": 0,
        "succeeded": 90,
        "partially_captured": 0,
        "partially_captured_and_capturable": 0,
        "requires_merchant_action": 0,
        "failed": 10
    }
}
```

Patch:
diff --git a/crates/api_models/src/events/payment.rs b/crates/api_models/src/events/payment.rs
index 589dcd8d10b..941cbb33309 100644
--- a/crates/api_models/src/events/payment.rs
+++ b/crates/api_models/src/events/payment.rs
@@ -18,12 +18,13 @@ use crate::{
     payments::{
         ExtendedCardInfoResponse, PaymentIdType, PaymentListConstraints,
         PaymentListFilterConstraints, PaymentListFilters, PaymentListFiltersV2,
-        PaymentListResponse, PaymentListResponseV2, PaymentsApproveRequest, PaymentsCancelRequest,
-        PaymentsCaptureRequest, PaymentsCompleteAuthorizeRequest,
-        PaymentsExternalAuthenticationRequest, PaymentsExternalAuthenticationResponse,
-        PaymentsIncrementalAuthorizationRequest, PaymentsManualUpdateRequest,
-        PaymentsRejectRequest, PaymentsRequest, PaymentsResponse, PaymentsRetrieveRequest,
-        PaymentsSessionResponse, PaymentsStartRequest, RedirectionResponse,
+        PaymentListResponse, PaymentListResponseV2, PaymentsAggregateResponse,
+        PaymentsApproveRequest, PaymentsCancelRequest, PaymentsCaptureRequest,
+        PaymentsCompleteAuthorizeRequest, PaymentsExternalAuthenticationRequest,
+        PaymentsExternalAuthenticationResponse, PaymentsIncrementalAuthorizationRequest,
+        PaymentsManualUpdateRequest, PaymentsRejectRequest, PaymentsRequest, PaymentsResponse,
+        PaymentsRetrieveRequest, PaymentsSessionResponse, PaymentsStartRequest,
+        RedirectionResponse,
     },
 };
 impl ApiEventMetric for PaymentsRetrieveRequest {
@@ -225,6 +226,11 @@ impl ApiEventMetric for PaymentListResponseV2 {
         Some(ApiEventsType::ResourceListAPI)
     }
 }
+impl ApiEventMetric for PaymentsAggregateResponse {
+    fn get_api_event_type(&self) -> Option<ApiEventsType> {
+        Some(ApiEventsType::ResourceListAPI)
+    }
+}
 
 impl ApiEventMetric for RedirectionResponse {}
 
diff --git a/crates/api_models/src/payments.rs b/crates/api_models/src/payments.rs
index 88c915f0e73..f9041b27373 100644
--- a/crates/api_models/src/payments.rs
+++ b/crates/api_models/src/payments.rs
@@ -4099,6 +4099,12 @@ pub struct PaymentListFiltersV2 {
     pub authentication_type: Vec<enums::AuthenticationType>,
 }
 
+#[derive(Clone, Debug, serde::Serialize)]
+pub struct PaymentsAggregateResponse {
+    /// The list of intent status with their count
+    pub status_with_count: HashMap<enums::IntentStatus, i64>,
+}
+
 #[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize, ToSchema)]
 pub struct AmountFilter {
     /// The start amount to filter list of transactions which are greater than or equal to the start amount
diff --git a/crates/hyperswitch_domain_models/src/payments/payment_intent.rs b/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
index 90e3619f02e..c129e83d556 100644
--- a/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
+++ b/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
@@ -67,6 +67,13 @@ pub trait PaymentIntentInterface {
         storage_scheme: storage_enums::MerchantStorageScheme,
     ) -> error_stack::Result<Vec<PaymentIntent>, errors::StorageError>;
 
+    #[cfg(feature = "olap")]
+    async fn get_intent_status_with_count(
+        &self,
+        merchant_id: &id_type::MerchantId,
+        constraints: &api_models::payments::TimeRange,
+    ) -> error_stack::Result<Vec<(common_enums::IntentStatus, i64)>, errors::StorageError>;
+
     #[cfg(feature = "olap")]
     async fn get_filtered_payment_intents_attempt(
         &self,
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index a8ec32cb55b..d49b621b6d1 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -3157,6 +3157,31 @@ pub async fn get_payment_filters(
     ))
 }
 
+#[cfg(feature = "olap")]
+pub async fn get_aggregates_for_payments(
+    state: SessionState,
+    merchant: domain::MerchantAccount,
+    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)
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
+
+    let mut status_map: HashMap<enums::IntentStatus, i64> =
+        intent_status_with_count.into_iter().collect();
+    for status in enums::IntentStatus::iter() {
+        status_map.entry(status).or_default();
+    }
+
+    Ok(services::ApplicationResponse::Json(
+        api::PaymentsAggregateResponse {
+            status_with_count: status_map,
+        },
+    ))
+}
+
 pub async fn add_process_sync_task(
     db: &dyn StorageInterface,
     payment_attempt: &storage::PaymentAttempt,
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 2f5bf8b6791..b8f1e320f47 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -1616,6 +1616,17 @@ impl PaymentIntentInterface for KafkaStore {
             .await
     }
 
+    #[cfg(feature = "olap")]
+    async fn get_intent_status_with_count(
+        &self,
+        merchant_id: &id_type::MerchantId,
+        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)
+            .await
+    }
+
     #[cfg(feature = "olap")]
     async fn get_filtered_payment_intents_attempt(
         &self,
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 5fe71a6a8da..f7c7583ad33 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -539,6 +539,7 @@ 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("/{payment_id}/manual-update")
                         .route(web::put().to(payments_manual_update)),
diff --git a/crates/router/src/routes/lock_utils.rs b/crates/router/src/routes/lock_utils.rs
index 9f073853d88..9f2f99368e2 100644
--- a/crates/router/src/routes/lock_utils.rs
+++ b/crates/router/src/routes/lock_utils.rs
@@ -126,6 +126,7 @@ impl From<Flow> for ApiIdentifier {
             | Flow::PaymentsStart
             | Flow::PaymentsList
             | Flow::PaymentsFilters
+            | Flow::PaymentsAggregate
             | Flow::PaymentsRedirect
             | Flow::PaymentsIncrementalAuthorization
             | Flow::PaymentsExternalAuthentication
diff --git a/crates/router/src/routes/payments.rs b/crates/router/src/routes/payments.rs
index d2f0cfeb775..c79d2d316bc 100644
--- a/crates/router/src/routes/payments.rs
+++ b/crates/router/src/routes/payments.rs
@@ -1069,6 +1069,29 @@ pub async fn get_payment_filters(
     .await
 }
 
+#[instrument(skip_all, fields(flow = ?Flow::PaymentsAggregate))]
+#[cfg(feature = "olap")]
+pub async fn get_payments_aggregates(
+    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, req)
+        },
+        &auth::JWTAuth(Permission::PaymentRead),
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
+
 #[cfg(feature = "oltp")]
 #[instrument(skip_all, fields(flow = ?Flow::PaymentsApprove, payment_id))]
 // #[post("/{payment_id}/approve")]
diff --git a/crates/router/src/types/api/payments.rs b/crates/router/src/types/api/payments.rs
index 6b156098d5e..ab25332a5c5 100644
--- a/crates/router/src/types/api/payments.rs
+++ b/crates/router/src/types/api/payments.rs
@@ -5,14 +5,14 @@ pub use api_models::payments::{
     OnlineMandate, OpenBankingSessionToken, PayLaterData, PaymentIdType, PaymentListConstraints,
     PaymentListFilterConstraints, PaymentListFilters, PaymentListFiltersV2, PaymentListResponse,
     PaymentListResponseV2, PaymentMethodData, PaymentMethodDataRequest, PaymentMethodDataResponse,
-    PaymentOp, PaymentRetrieveBody, PaymentRetrieveBodyWithCredentials, PaymentsApproveRequest,
-    PaymentsCancelRequest, PaymentsCaptureRequest, PaymentsCompleteAuthorizeRequest,
-    PaymentsExternalAuthenticationRequest, PaymentsIncrementalAuthorizationRequest,
-    PaymentsManualUpdateRequest, PaymentsRedirectRequest, PaymentsRedirectionResponse,
-    PaymentsRejectRequest, PaymentsRequest, PaymentsResponse, PaymentsResponseForm,
-    PaymentsRetrieveRequest, PaymentsSessionRequest, PaymentsSessionResponse, PaymentsStartRequest,
-    PgRedirectResponse, PhoneDetails, RedirectionResponse, SessionToken, TimeRange, UrlDetails,
-    VerifyRequest, VerifyResponse, WalletData,
+    PaymentOp, PaymentRetrieveBody, PaymentRetrieveBodyWithCredentials, PaymentsAggregateResponse,
+    PaymentsApproveRequest, PaymentsCancelRequest, PaymentsCaptureRequest,
+    PaymentsCompleteAuthorizeRequest, PaymentsExternalAuthenticationRequest,
+    PaymentsIncrementalAuthorizationRequest, PaymentsManualUpdateRequest, PaymentsRedirectRequest,
+    PaymentsRedirectionResponse, PaymentsRejectRequest, PaymentsRequest, PaymentsResponse,
+    PaymentsResponseForm, PaymentsRetrieveRequest, PaymentsSessionRequest, PaymentsSessionResponse,
+    PaymentsStartRequest, PgRedirectResponse, PhoneDetails, RedirectionResponse, SessionToken,
+    TimeRange, UrlDetails, VerifyRequest, VerifyResponse, WalletData,
 };
 use error_stack::ResultExt;
 pub use hyperswitch_domain_models::router_flow_types::payments::{
diff --git a/crates/router_env/src/logger/types.rs b/crates/router_env/src/logger/types.rs
index f1d2e5d8dcc..ccc2842bf5d 100644
--- a/crates/router_env/src/logger/types.rs
+++ b/crates/router_env/src/logger/types.rs
@@ -167,6 +167,8 @@ pub enum Flow {
     PaymentsList,
     /// Payments filters flow
     PaymentsFilters,
+    /// Payments aggregates flow
+    PaymentsAggregate,
     #[cfg(feature = "payouts")]
     /// Payouts create flow
     PayoutsCreate,
diff --git a/crates/storage_impl/src/mock_db/payment_intent.rs b/crates/storage_impl/src/mock_db/payment_intent.rs
index 59951045a1b..4fb06bc169c 100644
--- a/crates/storage_impl/src/mock_db/payment_intent.rs
+++ b/crates/storage_impl/src/mock_db/payment_intent.rs
@@ -41,6 +41,15 @@ impl PaymentIntentInterface for MockDb {
         Err(StorageError::MockDbError)?
     }
     #[cfg(feature = "olap")]
+    async fn get_intent_status_with_count(
+        &self,
+        _merchant_id: &common_utils::id_type::MerchantId,
+        _time_range: &api_models::payments::TimeRange,
+    ) -> CustomResult<Vec<(common_enums::IntentStatus, i64)>, StorageError> {
+        // [#172]: Implement function for `MockDb`
+        Err(StorageError::MockDbError)?
+    }
+    #[cfg(feature = "olap")]
     async fn get_filtered_active_attempt_ids_for_total_count(
         &self,
         _merchant_id: &common_utils::id_type::MerchantId,
diff --git a/crates/storage_impl/src/payments/payment_intent.rs b/crates/storage_impl/src/payments/payment_intent.rs
index c8dc011b814..f7127761afb 100644
--- a/crates/storage_impl/src/payments/payment_intent.rs
+++ b/crates/storage_impl/src/payments/payment_intent.rs
@@ -346,6 +346,16 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
             )
             .await
     }
+    #[cfg(feature = "olap")]
+    async fn get_intent_status_with_count(
+        &self,
+        merchant_id: &common_utils::id_type::MerchantId,
+        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)
+            .await
+    }
 
     #[cfg(feature = "olap")]
     async fn get_filtered_payment_intents_attempt(
@@ -655,6 +665,45 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
         .await
     }
 
+    #[cfg(feature = "olap")]
+    #[instrument(skip_all)]
+    async fn get_intent_status_with_count(
+        &self,
+        merchant_id: &common_utils::id_type::MerchantId,
+        time_range: &api_models::payments::TimeRange,
+    ) -> error_stack::Result<Vec<(common_enums::IntentStatus, i64)>, StorageError> {
+        let conn = connection::pg_connection_read(self).await.switch()?;
+        let conn = async_bb8_diesel::Connection::as_async_conn(&conn);
+
+        let mut query = <DieselPaymentIntent as HasTable>::table()
+            .group_by(pi_dsl::status)
+            .select((pi_dsl::status, diesel::dsl::count_star()))
+            .filter(pi_dsl::merchant_id.eq(merchant_id.to_owned()))
+            .into_boxed();
+
+        query = query.filter(pi_dsl::created_at.ge(time_range.start_time));
+
+        query = match time_range.end_time {
+            Some(ending_at) => query.filter(pi_dsl::created_at.le(ending_at)),
+            None => query,
+        };
+
+        logger::debug!(filter = %diesel::debug_query::<diesel::pg::Pg,_>(&query).to_string());
+
+        db_metrics::track_database_call::<<DieselPaymentIntent as HasTable>::Table, _, _>(
+            query.get_results_async::<(common_enums::IntentStatus, i64)>(conn),
+            db_metrics::DatabaseOperation::Filter,
+        )
+        .await
+        .map_err(|er| {
+            StorageError::DatabaseError(
+                error_stack::report!(diesel_models::errors::DatabaseError::from(er))
+                    .attach_printable("Error filtering payment records"),
+            )
+            .into()
+        })
+    }
+
     #[cfg(feature = "olap")]
     #[instrument(skip_all)]
     async fn get_filtered_payment_intents_attempt(