Issue:
Bug: fix(list): improve querying time for payments list



Quick Fix.
Currently payment list is giving intermittent 5xx. After analyzing logs we got to knwo that it is happening for the count query
That query need to be called always and we can skip this in most of the cases.

Also add the logs around count.

This can be a temporary help, till will move to more durable solution.

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [x] Bugfix
- [ ] New feature
- [x] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
- Skip count query if no filters are applied for attempts, in this we don't need to filter attempts as count of active attempt ids will be the total count.
- Profile id filter is already getting applied in intents, so removing it from last active attempts.
- Add logging around count query.

### Additional Changes

- [ ] 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 [#6330](https://github.com/juspay/hyperswitch/issues/6330)


## How did you test it?
For Tests Lists working is same, same request and response:
```
curl --location 'http://localhost:8080/payments/list' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'x-api-key: dev_VWtyXkyK1Ath1vBNRMM4dqm18J0PVW8SiML4IsKtuzpsR5ojI0FCDBIY0ZRggpFf' \
--header 'Authorization: Bearer JWT' \
--data '{
}'
``` 
Response will be something like
```
{
    "count": 0,
    "total_count": 0,
    "data": []
}
```

Log is getting printed sucessfully
`Completed count query in 292.71075m
`

Results.
After hitting some 100 concurrent request for a very large dataset, we can clearly see without count query api is taking less time, and there is significant diff.
WIth count query
<img width="464" alt="Screenshot 2024-10-16 at 3 59 08 AM" src="https://github.com/user-attachments/assets/7e689588-672d-4148-978d-262ef71e6f94">
Without count query
<img width="433" alt="Screenshot 2024-10-16 at 3 59 21 AM" src="https://github.com/user-attachments/assets/fb261141-b857-4a81-b4c0-3abe6fdd6a58">



## 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:
For Tests Lists working is same, same request and response:
```
curl --location 'http://localhost:8080/payments/list' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'x-api-key: dev_VWtyXkyK1Ath1vBNRMM4dqm18J0PVW8SiML4IsKtuzpsR5ojI0FCDBIY0ZRggpFf' \
--header 'Authorization: Bearer JWT' \
--data '{
}'
``` 
Response will be something like
```
{
    "count": 0,
    "total_count": 0,
    "data": []
}
```

Log is getting printed sucessfully
`Completed count query in 292.71075m
`

Results.
After hitting some 100 concurrent request for a very large dataset, we can clearly see without count query api is taking less time, and there is significant diff.
WIth count query
<img width="464" alt="Screenshot 2024-10-16 at 3 59 08 AM" src="https://github.com/user-attachments/assets/7e689588-672d-4148-978d-262ef71e6f94">
Without count query
<img width="433" alt="Screenshot 2024-10-16 at 3 59 21 AM" src="https://github.com/user-attachments/assets/fb261141-b857-4a81-b4c0-3abe6fdd6a58">

Patch:
diff --git a/crates/api_models/src/payments.rs b/crates/api_models/src/payments.rs
index 674a19f3ef8..039c99ef348 100644
--- a/crates/api_models/src/payments.rs
+++ b/crates/api_models/src/payments.rs
@@ -4449,6 +4449,17 @@ pub struct PaymentListFilterConstraints {
     /// The List of all the card networks to filter payments list
     pub card_network: Option<Vec<enums::CardNetwork>>,
 }
+
+impl PaymentListFilterConstraints {
+    pub fn has_no_attempt_filters(&self) -> bool {
+        self.connector.is_none()
+            && self.payment_method.is_none()
+            && self.payment_method_type.is_none()
+            && self.authentication_type.is_none()
+            && self.merchant_connector_id.is_none()
+    }
+}
+
 #[derive(Clone, Debug, serde::Serialize)]
 pub struct PaymentListFilters {
     /// The list of available connector filters
diff --git a/crates/diesel_models/src/query/payment_attempt.rs b/crates/diesel_models/src/query/payment_attempt.rs
index 67af7a3bef5..85c69824892 100644
--- a/crates/diesel_models/src/query/payment_attempt.rs
+++ b/crates/diesel_models/src/query/payment_attempt.rs
@@ -369,7 +369,6 @@ impl PaymentAttempt {
         payment_method: Option<Vec<enums::PaymentMethod>>,
         payment_method_type: Option<Vec<enums::PaymentMethodType>>,
         authentication_type: Option<Vec<enums::AuthenticationType>>,
-        profile_id_list: Option<Vec<common_utils::id_type::ProfileId>>,
         merchant_connector_id: Option<Vec<common_utils::id_type::MerchantConnectorAccountId>>,
     ) -> StorageResult<i64> {
         let mut filter = <Self as HasTable>::table()
@@ -394,17 +393,23 @@ impl PaymentAttempt {
         if let Some(merchant_connector_id) = merchant_connector_id {
             filter = filter.filter(dsl::merchant_connector_id.eq_any(merchant_connector_id))
         }
-        if let Some(profile_id_list) = profile_id_list {
-            filter = filter.filter(dsl::profile_id.eq_any(profile_id_list))
-        }
+
         router_env::logger::debug!(query = %debug_query::<Pg, _>(&filter).to_string());
 
-        db_metrics::track_database_call::<<Self as HasTable>::Table, _, _>(
+        // TODO: Remove these logs after debugging the issue for delay in count query
+        let start_time = std::time::Instant::now();
+        router_env::logger::debug!("Executing count query start_time: {:?}", start_time);
+        let result = db_metrics::track_database_call::<<Self as HasTable>::Table, _, _>(
             filter.get_result_async::<i64>(conn),
             db_metrics::DatabaseOperation::Filter,
         )
         .await
         .change_context(DatabaseError::Others)
-        .attach_printable("Error filtering count of payments")
+        .attach_printable("Error filtering count of payments");
+
+        let duration = start_time.elapsed();
+        router_env::logger::debug!("Completed count query in {:?}", duration);
+
+        result
     }
 }
diff --git a/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs b/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
index 5aba0616143..ea65cfff8b4 100644
--- a/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
+++ b/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
@@ -158,7 +158,6 @@ pub trait PaymentAttemptInterface {
         payment_method_type: Option<Vec<storage_enums::PaymentMethodType>>,
         authentication_type: Option<Vec<storage_enums::AuthenticationType>>,
         merchant_connector_id: Option<Vec<id_type::MerchantConnectorAccountId>>,
-        profile_id_list: Option<Vec<id_type::ProfileId>>,
         storage_scheme: storage_enums::MerchantStorageScheme,
     ) -> error_stack::Result<i64, errors::StorageError>;
 }
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index 68f318958a2..651a79a0c14 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -3737,54 +3737,69 @@ pub async fn apply_filters_on_payments(
     merchant_key_store: domain::MerchantKeyStore,
     constraints: api::PaymentListFilterConstraints,
 ) -> RouterResponse<api::PaymentListResponseV2> {
-    let limit = &constraints.limit;
-    helpers::validate_payment_list_request_for_joins(*limit)?;
-    let db: &dyn StorageInterface = state.store.as_ref();
-    let pi_fetch_constraints = (constraints.clone(), profile_id_list.clone()).try_into()?;
-    let list: Vec<(storage::PaymentIntent, storage::PaymentAttempt)> = db
-        .get_filtered_payment_intents_attempt(
-            &(&state).into(),
-            merchant.get_id(),
-            &pi_fetch_constraints,
-            &merchant_key_store,
-            merchant.storage_scheme,
-        )
-        .await
-        .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
-    let data: Vec<api::PaymentsResponse> =
-        list.into_iter().map(ForeignFrom::foreign_from).collect();
-
-    let active_attempt_ids = db
-        .get_filtered_active_attempt_ids_for_total_count(
-            merchant.get_id(),
-            &pi_fetch_constraints,
-            merchant.storage_scheme,
-        )
-        .await
-        .to_not_found_response(errors::ApiErrorResponse::InternalServerError)?;
+    common_utils::metrics::utils::record_operation_time(
+        async {
+            let limit = &constraints.limit;
+            helpers::validate_payment_list_request_for_joins(*limit)?;
+            let db: &dyn StorageInterface = state.store.as_ref();
+            let pi_fetch_constraints = (constraints.clone(), profile_id_list.clone()).try_into()?;
+            let list: Vec<(storage::PaymentIntent, storage::PaymentAttempt)> = db
+                .get_filtered_payment_intents_attempt(
+                    &(&state).into(),
+                    merchant.get_id(),
+                    &pi_fetch_constraints,
+                    &merchant_key_store,
+                    merchant.storage_scheme,
+                )
+                .await
+                .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
+            let data: Vec<api::PaymentsResponse> =
+                list.into_iter().map(ForeignFrom::foreign_from).collect();
+
+            let active_attempt_ids = db
+                .get_filtered_active_attempt_ids_for_total_count(
+                    merchant.get_id(),
+                    &pi_fetch_constraints,
+                    merchant.storage_scheme,
+                )
+                .await
+                .to_not_found_response(errors::ApiErrorResponse::InternalServerError)?;
 
-    let total_count = db
-        .get_total_count_of_filtered_payment_attempts(
-            merchant.get_id(),
-            &active_attempt_ids,
-            constraints.connector,
-            constraints.payment_method,
-            constraints.payment_method_type,
-            constraints.authentication_type,
-            constraints.merchant_connector_id,
-            pi_fetch_constraints.get_profile_id_list(),
-            merchant.storage_scheme,
-        )
-        .await
-        .change_context(errors::ApiErrorResponse::InternalServerError)?;
+            let total_count = if constraints.has_no_attempt_filters() {
+                i64::try_from(active_attempt_ids.len())
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable("Error while converting from usize to i64")
+            } else {
+                db.get_total_count_of_filtered_payment_attempts(
+                    merchant.get_id(),
+                    &active_attempt_ids,
+                    constraints.connector,
+                    constraints.payment_method,
+                    constraints.payment_method_type,
+                    constraints.authentication_type,
+                    constraints.merchant_connector_id,
+                    merchant.storage_scheme,
+                )
+                .await
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+            }?;
 
-    Ok(services::ApplicationResponse::Json(
-        api::PaymentListResponseV2 {
-            count: data.len(),
-            total_count,
-            data,
+            Ok(services::ApplicationResponse::Json(
+                api::PaymentListResponseV2 {
+                    count: data.len(),
+                    total_count,
+                    data,
+                },
+            ))
         },
-    ))
+        &metrics::PAYMENT_LIST_LATENCY,
+        &metrics::CONTEXT,
+        &[router_env::opentelemetry::KeyValue::new(
+            "merchant_id",
+            merchant.get_id().clone(),
+        )],
+    )
+    .await
 }
 
 #[cfg(all(feature = "olap", feature = "v1"))]
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 9f237c226b6..e8fbd7cb483 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -1601,7 +1601,6 @@ impl PaymentAttemptInterface for KafkaStore {
         payment_method_type: Option<Vec<common_enums::PaymentMethodType>>,
         authentication_type: Option<Vec<common_enums::AuthenticationType>>,
         merchant_connector_id: Option<Vec<id_type::MerchantConnectorAccountId>>,
-        profile_id_list: Option<Vec<id_type::ProfileId>>,
         storage_scheme: MerchantStorageScheme,
     ) -> CustomResult<i64, errors::DataStorageError> {
         self.diesel_store
@@ -1613,7 +1612,6 @@ impl PaymentAttemptInterface for KafkaStore {
                 payment_method_type,
                 authentication_type,
                 merchant_connector_id,
-                profile_id_list,
                 storage_scheme,
             )
             .await
diff --git a/crates/router/src/routes/metrics.rs b/crates/router/src/routes/metrics.rs
index f3f1cd7cda0..1b55c838c45 100644
--- a/crates/router/src/routes/metrics.rs
+++ b/crates/router/src/routes/metrics.rs
@@ -21,6 +21,8 @@ counter_metric!(PAYMENT_OPS_COUNT, GLOBAL_METER);
 
 counter_metric!(PAYMENT_COUNT, GLOBAL_METER);
 counter_metric!(SUCCESSFUL_PAYMENT, GLOBAL_METER);
+//TODO: This can be removed, added for payment list debugging
+histogram_metric!(PAYMENT_LIST_LATENCY, GLOBAL_METER);
 
 counter_metric!(REFUND_COUNT, GLOBAL_METER);
 counter_metric!(SUCCESSFUL_REFUND, GLOBAL_METER);
diff --git a/crates/router/src/routes/payments.rs b/crates/router/src/routes/payments.rs
index 7410c5ee363..77ba1da0fcd 100644
--- a/crates/router/src/routes/payments.rs
+++ b/crates/router/src/routes/payments.rs
@@ -1035,7 +1035,7 @@ pub async fn profile_payments_list_by_filter(
         |state, auth: auth::AuthenticationData, req, _| {
             payments::apply_filters_on_payments(
                 state,
-                auth.merchant_account,
+                auth.merchant_account.clone(),
                 auth.profile_id.map(|profile_id| vec![profile_id]),
                 auth.key_store,
                 req,
diff --git a/crates/storage_impl/src/mock_db/payment_attempt.rs b/crates/storage_impl/src/mock_db/payment_attempt.rs
index 13c75e005c4..59fb1155480 100644
--- a/crates/storage_impl/src/mock_db/payment_attempt.rs
+++ b/crates/storage_impl/src/mock_db/payment_attempt.rs
@@ -52,7 +52,6 @@ impl PaymentAttemptInterface for MockDb {
         _payment_method_type: Option<Vec<common_enums::PaymentMethodType>>,
         _authentication_type: Option<Vec<common_enums::AuthenticationType>>,
         _merchanat_connector_id: Option<Vec<common_utils::id_type::MerchantConnectorAccountId>>,
-        _profile_id_list: Option<Vec<common_utils::id_type::ProfileId>>,
         _storage_scheme: storage_enums::MerchantStorageScheme,
     ) -> CustomResult<i64, StorageError> {
         Err(StorageError::MockDbError)?
diff --git a/crates/storage_impl/src/payments/payment_attempt.rs b/crates/storage_impl/src/payments/payment_attempt.rs
index 9053b7561ad..20b46a8460d 100644
--- a/crates/storage_impl/src/payments/payment_attempt.rs
+++ b/crates/storage_impl/src/payments/payment_attempt.rs
@@ -402,7 +402,6 @@ impl<T: DatabaseStore> PaymentAttemptInterface for RouterStore<T> {
         payment_method_type: Option<Vec<common_enums::PaymentMethodType>>,
         authentication_type: Option<Vec<common_enums::AuthenticationType>>,
         merchant_connector_id: Option<Vec<common_utils::id_type::MerchantConnectorAccountId>>,
-        profile_id_list: Option<Vec<common_utils::id_type::ProfileId>>,
         _storage_scheme: MerchantStorageScheme,
     ) -> CustomResult<i64, errors::StorageError> {
         let conn = self
@@ -425,7 +424,6 @@ impl<T: DatabaseStore> PaymentAttemptInterface for RouterStore<T> {
             payment_method,
             payment_method_type,
             authentication_type,
-            profile_id_list,
             merchant_connector_id,
         )
         .await
@@ -1280,7 +1278,6 @@ impl<T: DatabaseStore> PaymentAttemptInterface for KVRouterStore<T> {
         payment_method_type: Option<Vec<common_enums::PaymentMethodType>>,
         authentication_type: Option<Vec<common_enums::AuthenticationType>>,
         merchant_connector_id: Option<Vec<common_utils::id_type::MerchantConnectorAccountId>>,
-        profile_id_list: Option<Vec<common_utils::id_type::ProfileId>>,
         storage_scheme: MerchantStorageScheme,
     ) -> CustomResult<i64, errors::StorageError> {
         self.router_store
@@ -1292,7 +1289,6 @@ impl<T: DatabaseStore> PaymentAttemptInterface for KVRouterStore<T> {
                 payment_method_type,
                 authentication_type,
                 merchant_connector_id,
-                profile_id_list,
                 storage_scheme,
             )
             .await