feat: enable payment and refund filter at DB query level (#5827)

crates/diesel_models/src/query/payment_attempt.rs

@@ -322,6 +322,7 @@ 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()
@@ -346,6 +347,9 @@ 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, _, _>(

crates/hyperswitch_domain_models/src/lib.rs

@@ -14,6 +14,7 @@ pub mod payment_methods;
 pub mod payments;
 #[cfg(feature = "payouts")]
 pub mod payouts;
+pub mod refunds;
 pub mod router_data;
 pub mod router_data_v2;
 pub mod router_flow_types;

crates/hyperswitch_domain_models/src/payments/payment_attempt.rs

@@ -112,6 +112,7 @@ 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>;
 }

crates/hyperswitch_domain_models/src/payments/payment_intent.rs

@@ -771,6 +771,16 @@ pub enum PaymentIntentFetchConstraints {
     List(Box<PaymentIntentListParams>),
 }
 
+impl PaymentIntentFetchConstraints {
+    pub fn get_profile_id_list(&self) -> Option<Vec<id_type::ProfileId>> {
+        if let Self::List(pi_list_params) = self {
+            pi_list_params.profile_id.clone()
+        } else {
+            None
+        }
+    }
+}
+
 pub struct PaymentIntentListParams {
     pub offset: u32,
     pub starting_at: Option<PrimitiveDateTime>,
@@ -783,7 +793,7 @@ pub struct PaymentIntentListParams {
     pub payment_method_type: Option<Vec<storage_enums::PaymentMethodType>>,
     pub authentication_type: Option<Vec<storage_enums::AuthenticationType>>,
     pub merchant_connector_id: Option<Vec<id_type::MerchantConnectorAccountId>>,
-    pub profile_id: Option<id_type::ProfileId>,
+    pub profile_id: Option<Vec<id_type::ProfileId>>,
     pub customer_id: Option<id_type::CustomerId>,
     pub starting_after_id: Option<id_type::PaymentId>,
     pub ending_before_id: Option<id_type::PaymentId>,
@@ -793,10 +803,21 @@ pub struct PaymentIntentListParams {
 
 impl From<api_models::payments::PaymentListConstraints> for PaymentIntentFetchConstraints {
     fn from(value: api_models::payments::PaymentListConstraints) -> Self {
+        let api_models::payments::PaymentListConstraints {
+            customer_id,
+            starting_after,
+            ending_before,
+            limit,
+            created,
+            created_lt,
+            created_gt,
+            created_lte,
+            created_gte,
+        } = value;
         Self::List(Box::new(PaymentIntentListParams {
             offset: 0,
-            starting_at: value.created_gte.or(value.created_gt).or(value.created),
+            starting_at: created_gte.or(created_gt).or(created),
-            ending_at: value.created_lte.or(value.created_lt).or(value.created),
+            ending_at: created_lte.or(created_lt).or(created),
             amount_filter: None,
             connector: None,
             currency: None,
@@ -806,10 +827,10 @@ impl From<api_models::payments::PaymentListConstraints> for PaymentIntentFetchCo
             authentication_type: None,
             merchant_connector_id: None,
             profile_id: None,
-            customer_id: value.customer_id,
+            customer_id,
-            starting_after_id: value.starting_after,
+            starting_after_id: starting_after,
-            ending_before_id: value.ending_before,
+            ending_before_id: ending_before,
-            limit: Some(std::cmp::min(value.limit, PAYMENTS_LIST_MAX_LIMIT_V1)),
+            limit: Some(std::cmp::min(limit, PAYMENTS_LIST_MAX_LIMIT_V1)),
             order: Default::default(),
         }))
     }
@@ -841,28 +862,96 @@ impl From<api_models::payments::TimeRange> for PaymentIntentFetchConstraints {
 
 impl From<api_models::payments::PaymentListFilterConstraints> for PaymentIntentFetchConstraints {
     fn from(value: api_models::payments::PaymentListFilterConstraints) -> Self {
-        if let Some(payment_intent_id) = value.payment_id {
+        let api_models::payments::PaymentListFilterConstraints {
+            payment_id,
+            profile_id,
+            customer_id,
+            limit,
+            offset,
+            amount_filter,
+            time_range,
+            connector,
+            currency,
+            status,
+            payment_method,
+            payment_method_type,
+            authentication_type,
+            merchant_connector_id,
+            order,
+        } = value;
+        if let Some(payment_intent_id) = payment_id {
             Self::Single { payment_intent_id }
         } else {
             Self::List(Box::new(PaymentIntentListParams {
-                offset: value.offset.unwrap_or_default(),
+                offset: offset.unwrap_or_default(),
-                starting_at: value.time_range.map(|t| t.start_time),
+                starting_at: time_range.map(|t| t.start_time),
-                ending_at: value.time_range.and_then(|t| t.end_time),
+                ending_at: time_range.and_then(|t| t.end_time),
-                amount_filter: value.amount_filter,
+                amount_filter,
-                connector: value.connector,
+                connector,
-                currency: value.currency,
+                currency,
-                status: value.status,
+                status,
-                payment_method: value.payment_method,
+                payment_method,
-                payment_method_type: value.payment_method_type,
+                payment_method_type,
-                authentication_type: value.authentication_type,
+                authentication_type,
-                merchant_connector_id: value.merchant_connector_id,
+                merchant_connector_id,
-                profile_id: value.profile_id,
+                profile_id: profile_id.map(|profile_id| vec![profile_id]),
-                customer_id: value.customer_id,
+                customer_id,
                 starting_after_id: None,
                 ending_before_id: None,
-                limit: Some(std::cmp::min(value.limit, PAYMENTS_LIST_MAX_LIMIT_V2)),
+                limit: Some(std::cmp::min(limit, PAYMENTS_LIST_MAX_LIMIT_V2)),
-                order: value.order,
+                order,
             }))
         }
     }
 }
+
+impl<T> TryFrom<(T, Option<Vec<id_type::ProfileId>>)> for PaymentIntentFetchConstraints
+where
+    Self: From<T>,
+{
+    type Error = error_stack::Report<errors::api_error_response::ApiErrorResponse>;
+    fn try_from(
+        (constraints, auth_profile_id_list): (T, Option<Vec<id_type::ProfileId>>),
+    ) -> Result<Self, Self::Error> {
+        let payment_intent_constraints = Self::from(constraints);
+        if let Self::List(mut pi_list_params) = payment_intent_constraints {
+            let profile_id_from_request_body = pi_list_params.profile_id;
+            match (profile_id_from_request_body, auth_profile_id_list) {
+                (None, None) => pi_list_params.profile_id = None,
+                (None, Some(auth_profile_id_list)) => {
+                    pi_list_params.profile_id = Some(auth_profile_id_list)
+                }
+                (Some(profile_id_from_request_body), None) => {
+                    pi_list_params.profile_id = Some(profile_id_from_request_body)
+                }
+                (Some(profile_id_from_request_body), Some(auth_profile_id_list)) => {
+                    let profile_id_from_request_body_is_available_in_auth_profile_id_list =
+                        profile_id_from_request_body
+                            .iter()
+                            .all(|profile_id| auth_profile_id_list.contains(profile_id));
+
+                    if profile_id_from_request_body_is_available_in_auth_profile_id_list {
+                        pi_list_params.profile_id = Some(profile_id_from_request_body)
+                    } else {
+                        // This scenario is very unlikely to happen
+                        let inaccessible_profile_ids: Vec<_> = profile_id_from_request_body
+                            .iter()
+                            .filter(|profile_id| !auth_profile_id_list.contains(profile_id))
+                            .collect();
+                        return Err(error_stack::Report::new(
+                            errors::api_error_response::ApiErrorResponse::PreconditionFailed {
+                                message: format!(
+                                    "Access not available for the given profile_id {:?}",
+                                    inaccessible_profile_ids
+                                ),
+                            },
+                        ));
+                    }
+                }
+            }
+            Ok(Self::List(pi_list_params))
+        } else {
+            Ok(payment_intent_constraints)
+        }
+    }
+}

crates/hyperswitch_domain_models/src/refunds.rs

@@ -0,0 +1,82 @@
+use crate::errors;
+
+pub struct RefundListConstraints {
+    pub payment_id: Option<common_utils::id_type::PaymentId>,
+    pub refund_id: Option<String>,
+    pub profile_id: Option<Vec<common_utils::id_type::ProfileId>>,
+    pub limit: Option<i64>,
+    pub offset: Option<i64>,
+    pub time_range: Option<api_models::payments::TimeRange>,
+    pub amount_filter: Option<api_models::payments::AmountFilter>,
+    pub connector: Option<Vec<String>>,
+    pub merchant_connector_id: Option<Vec<common_utils::id_type::MerchantConnectorAccountId>>,
+    pub currency: Option<Vec<common_enums::Currency>>,
+    pub refund_status: Option<Vec<common_enums::RefundStatus>>,
+}
+
+impl
+    TryFrom<(
+        api_models::refunds::RefundListRequest,
+        Option<Vec<common_utils::id_type::ProfileId>>,
+    )> for RefundListConstraints
+{
+    type Error = error_stack::Report<errors::api_error_response::ApiErrorResponse>;
+
+    fn try_from(
+        (value, auth_profile_id_list): (
+            api_models::refunds::RefundListRequest,
+            Option<Vec<common_utils::id_type::ProfileId>>,
+        ),
+    ) -> Result<Self, Self::Error> {
+        let api_models::refunds::RefundListRequest {
+            connector,
+            currency,
+            refund_status,
+            payment_id,
+            refund_id,
+            profile_id,
+            limit,
+            offset,
+            time_range,
+            amount_filter,
+            merchant_connector_id,
+        } = value;
+        let profile_id_from_request_body = profile_id;
+        let profile_id_list = match (profile_id_from_request_body, auth_profile_id_list) {
+            (None, None) => None,
+            (None, Some(auth_profile_id_list)) => Some(auth_profile_id_list),
+            (Some(profile_id_from_request_body), None) => Some(vec![profile_id_from_request_body]),
+            (Some(profile_id_from_request_body), Some(auth_profile_id_list)) => {
+                let profile_id_from_request_body_is_available_in_auth_profile_id_list =
+                    auth_profile_id_list.contains(&profile_id_from_request_body);
+
+                if profile_id_from_request_body_is_available_in_auth_profile_id_list {
+                    Some(vec![profile_id_from_request_body])
+                } else {
+                    // This scenario is very unlikely to happen
+                    return Err(error_stack::Report::new(
+                        errors::api_error_response::ApiErrorResponse::PreconditionFailed {
+                            message: format!(
+                                "Access not available for the given profile_id {:?}",
+                                profile_id_from_request_body
+                            ),
+                        },
+                    ));
+                }
+            }
+        };
+        Ok(Self {
+            payment_id,
+            refund_id,
+            profile_id: profile_id_list,
+            limit,
+            offset,
+            time_range,
+            amount_filter,
+            connector,
+            merchant_connector_id,
+            currency,
+            refund_status,
+        })
+    }
+}

crates/router/src/core/payments.rs

@@ -2940,15 +2940,13 @@ pub async fn list_payments(
     let db = state.store.as_ref();
     let payment_intents = helpers::filter_by_constraints(
         &state,
-        &constraints,
+        &(constraints, profile_id_list).try_into()?,
         merchant_id,
         &key_store,
         merchant.storage_scheme,
     )
     .await
     .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
-    let payment_intents =
-        utils::filter_objects_based_on_profile_id_list(profile_id_list, payment_intents);
 
     let collected_futures = payment_intents.into_iter().map(|pi| {
         async {
@@ -3011,25 +3009,25 @@ pub async fn apply_filters_on_payments(
 ) -> RouterResponse<api::PaymentListResponseV2> {
     let limit = &constraints.limit;
     helpers::validate_payment_list_request_for_joins(*limit)?;
-    let db = state.store.as_ref();
+    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(),
-            &constraints.clone().into(),
+            &pi_fetch_constraints,
             &merchant_key_store,
             merchant.storage_scheme,
         )
         .await
         .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
-    let list = utils::filter_objects_based_on_profile_id_list(profile_id_list, list);
     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(),
-            &constraints.clone().into(),
+            &pi_fetch_constraints,
             merchant.storage_scheme,
         )
         .await
@@ -3044,6 +3042,7 @@ pub async fn apply_filters_on_payments(
             constraints.payment_method_type,
             constraints.authentication_type,
             constraints.merchant_connector_id,
+            pi_fetch_constraints.get_profile_id_list(),
             merchant.storage_scheme,
         )
         .await

crates/router/src/core/payments/helpers.rs

@@ -26,7 +26,10 @@ use hyperswitch_domain_models::payments::payment_intent::CustomerData;
 use hyperswitch_domain_models::{
     mandates::MandateData,
     payment_method_data::GetPaymentMethodType,
-    payments::{payment_attempt::PaymentAttempt, PaymentIntent},
+    payments::{
+        payment_attempt::PaymentAttempt, payment_intent::PaymentIntentFetchConstraints,
+        PaymentIntent,
+    },
     router_data::KlarnaSdkResponse,
 };
 use hyperswitch_interfaces::integrity::{CheckIntegrity, FlowIntegrity, GetIntegrityObject};
@@ -2538,7 +2541,7 @@ where
 #[cfg(feature = "olap")]
 pub(super) async fn filter_by_constraints(
     state: &SessionState,
-    constraints: &api::PaymentListConstraints,
+    constraints: &PaymentIntentFetchConstraints,
     merchant_id: &id_type::MerchantId,
     key_store: &domain::MerchantKeyStore,
     storage_scheme: storage_enums::MerchantStorageScheme,
@@ -2548,7 +2551,7 @@ pub(super) async fn filter_by_constraints(
         .filter_payment_intent_by_constraints(
             &(state).into(),
             merchant_id,
-            &constraints.clone().into(),
+            constraints,
             key_store,
             storage_scheme,
         )

crates/router/src/core/refunds.rs

@@ -869,7 +869,7 @@ pub async fn validate_and_create_refund(
 pub async fn refund_list(
     state: SessionState,
     merchant_account: domain::MerchantAccount,
-    _profile_id_list: Option<Vec<common_utils::id_type::ProfileId>>,
+    profile_id_list: Option<Vec<common_utils::id_type::ProfileId>>,
     req: api_models::refunds::RefundListRequest,
 ) -> RouterResponse<api_models::refunds::RefundListResponse> {
     let db = state.store;
@@ -879,7 +879,7 @@ pub async fn refund_list(
     let refund_list = db
         .filter_refund_by_constraints(
             merchant_account.get_id(),
-            &req,
+            &(req.clone(), profile_id_list.clone()).try_into()?,
             merchant_account.storage_scheme,
             limit,
             offset,
@@ -895,7 +895,7 @@ pub async fn refund_list(
     let total_count = db
         .get_total_count_of_refunds(
             merchant_account.get_id(),
-            &req,
+            &(req, profile_id_list).try_into()?,
             merchant_account.storage_scheme,
         )
         .await

crates/router/src/db/kafka_store.rs

@@ -9,13 +9,14 @@ use diesel_models::{
     reverse_lookup::{ReverseLookup, ReverseLookupNew},
     user_role as user_storage,
 };
-use hyperswitch_domain_models::payments::{
-    payment_attempt::PaymentAttemptInterface, payment_intent::PaymentIntentInterface,
-};
 #[cfg(feature = "payouts")]
 use hyperswitch_domain_models::payouts::{
     payout_attempt::PayoutAttemptInterface, payouts::PayoutsInterface,
 };
+use hyperswitch_domain_models::{
+    payments::{payment_attempt::PaymentAttemptInterface, payment_intent::PaymentIntentInterface},
+    refunds,
+};
 #[cfg(not(feature = "payouts"))]
 use hyperswitch_domain_models::{PayoutAttemptInterface, PayoutsInterface};
 use masking::Secret;
@@ -1490,6 +1491,7 @@ 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
@@ -1501,6 +1503,7 @@ impl PaymentAttemptInterface for KafkaStore {
                 payment_method_type,
                 authentication_type,
                 merchant_connector_id,
+                profile_id_list,
                 storage_scheme,
             )
             .await
@@ -2361,7 +2364,7 @@ impl RefundInterface for KafkaStore {
     async fn filter_refund_by_constraints(
         &self,
         merchant_id: &id_type::MerchantId,
-        refund_details: &api_models::refunds::RefundListRequest,
+        refund_details: &refunds::RefundListConstraints,
         storage_scheme: MerchantStorageScheme,
         limit: i64,
         offset: i64,
@@ -2393,7 +2396,7 @@ impl RefundInterface for KafkaStore {
     async fn get_total_count_of_refunds(
         &self,
         merchant_id: &id_type::MerchantId,
-        refund_details: &api_models::refunds::RefundListRequest,
+        refund_details: &refunds::RefundListConstraints,
         storage_scheme: MerchantStorageScheme,
     ) -> CustomResult<i64, errors::StorageError> {
         self.diesel_store

crates/router/src/db/refund.rs

@@ -4,6 +4,7 @@ use std::collections::HashSet;
 #[cfg(feature = "olap")]
 use common_utils::types::MinorUnit;
 use diesel_models::{errors::DatabaseError, refund::RefundUpdateInternal};
+use hyperswitch_domain_models::refunds;
 
 use super::MockDb;
 use crate::{
@@ -69,7 +70,7 @@ pub trait RefundInterface {
     async fn filter_refund_by_constraints(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
-        refund_details: &api_models::refunds::RefundListRequest,
+        refund_details: &refunds::RefundListConstraints,
         storage_scheme: enums::MerchantStorageScheme,
         limit: i64,
         offset: i64,
@@ -87,7 +88,7 @@ pub trait RefundInterface {
     async fn get_total_count_of_refunds(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
-        refund_details: &api_models::refunds::RefundListRequest,
+        refund_details: &refunds::RefundListConstraints,
         storage_scheme: enums::MerchantStorageScheme,
     ) -> CustomResult<i64, errors::StorageError>;
 }
@@ -216,7 +217,7 @@ mod storage {
         async fn filter_refund_by_constraints(
             &self,
             merchant_id: &common_utils::id_type::MerchantId,
-            refund_details: &api_models::refunds::RefundListRequest,
+            refund_details: &refunds::RefundListConstraints,
             _storage_scheme: enums::MerchantStorageScheme,
             limit: i64,
             offset: i64,
@@ -255,7 +256,7 @@ mod storage {
         async fn get_total_count_of_refunds(
             &self,
             merchant_id: &common_utils::id_type::MerchantId,
-            refund_details: &api_models::refunds::RefundListRequest,
+            refund_details: &refunds::RefundListConstraints,
             _storage_scheme: enums::MerchantStorageScheme,
         ) -> CustomResult<i64, errors::StorageError> {
             let conn = connection::pg_connection_read(self).await?;
@@ -274,6 +275,7 @@ mod storage {
 mod storage {
     use common_utils::{ext_traits::Encode, fallback_reverse_lookup_not_found};
     use error_stack::{report, ResultExt};
+    use hyperswitch_domain_models::refunds;
     use redis_interface::HsetnxReply;
     use router_env::{instrument, tracing};
     use storage_impl::redis::kv_store::{
@@ -752,7 +754,7 @@ mod storage {
         async fn filter_refund_by_constraints(
             &self,
             merchant_id: &common_utils::id_type::MerchantId,
-            refund_details: &api_models::refunds::RefundListRequest,
+            refund_details: &refunds::RefundListConstraints,
             _storage_scheme: enums::MerchantStorageScheme,
             limit: i64,
             offset: i64,
@@ -788,7 +790,7 @@ mod storage {
         async fn get_total_count_of_refunds(
             &self,
             merchant_id: &common_utils::id_type::MerchantId,
-            refund_details: &api_models::refunds::RefundListRequest,
+            refund_details: &refunds::RefundListConstraints,
             _storage_scheme: enums::MerchantStorageScheme,
         ) -> CustomResult<i64, errors::StorageError> {
             let conn = connection::pg_connection_read(self).await?;
@@ -963,7 +965,7 @@ impl RefundInterface for MockDb {
     async fn filter_refund_by_constraints(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
-        refund_details: &api_models::refunds::RefundListRequest,
+        refund_details: &refunds::RefundListConstraints,
         _storage_scheme: enums::MerchantStorageScheme,
         limit: i64,
         offset: i64,
@@ -972,6 +974,7 @@ impl RefundInterface for MockDb {
         let mut unique_merchant_connector_ids = HashSet::new();
         let mut unique_currencies = HashSet::new();
         let mut unique_statuses = HashSet::new();
+        let mut unique_profile_ids = HashSet::new();
 
         // Fill the hash sets with data from refund_details
         if let Some(connectors) = &refund_details.connector {
@@ -1000,6 +1003,10 @@ impl RefundInterface for MockDb {
             });
         }
 
+        if let Some(profile_id_list) = &refund_details.profile_id {
+            unique_profile_ids = profile_id_list.iter().collect();
+        }
+
         let refunds = self.refunds.lock().await;
         let filtered_refunds = refunds
             .iter()
@@ -1016,7 +1023,11 @@ impl RefundInterface for MockDb {
                     .clone()
                     .map_or(true, |id| id == refund.refund_id)
             })
-            .filter(|refund| refund_details.profile_id == refund.profile_id)
+            .filter(|refund| {
+                refund.profile_id.as_ref().is_some_and(|profile_id| {
+                    unique_profile_ids.is_empty() || unique_profile_ids.contains(profile_id)
+                })
+            })
             .filter(|refund| {
                 refund.created_at
                     >= refund_details.time_range.map_or(
@@ -1116,13 +1127,14 @@ impl RefundInterface for MockDb {
     async fn get_total_count_of_refunds(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
-        refund_details: &api_models::refunds::RefundListRequest,
+        refund_details: &refunds::RefundListConstraints,
         _storage_scheme: enums::MerchantStorageScheme,
     ) -> CustomResult<i64, errors::StorageError> {
         let mut unique_connectors = HashSet::new();
         let mut unique_merchant_connector_ids = HashSet::new();
         let mut unique_currencies = HashSet::new();
         let mut unique_statuses = HashSet::new();
+        let mut unique_profile_ids = HashSet::new();
 
         // Fill the hash sets with data from refund_details
         if let Some(connectors) = &refund_details.connector {
@@ -1151,6 +1163,10 @@ impl RefundInterface for MockDb {
             });
         }
 
+        if let Some(profile_id_list) = &refund_details.profile_id {
+            unique_profile_ids = profile_id_list.iter().collect();
+        }
+
         let refunds = self.refunds.lock().await;
         let filtered_refunds = refunds
             .iter()
@@ -1167,7 +1183,11 @@ impl RefundInterface for MockDb {
                     .clone()
                     .map_or(true, |id| id == refund.refund_id)
             })
-            .filter(|refund| refund_details.profile_id == refund.profile_id)
+            .filter(|refund| {
+                refund.profile_id.as_ref().is_some_and(|profile_id| {
+                    unique_profile_ids.is_empty() || unique_profile_ids.contains(profile_id)
+                })
+            })
             .filter(|refund| {
                 refund.created_at
                     >= refund_details.time_range.map_or(

crates/router/src/types/storage/refund.rs

@@ -12,6 +12,7 @@ use diesel_models::{
     schema::refund::dsl,
 };
 use error_stack::ResultExt;
+use hyperswitch_domain_models::refunds;
 
 use crate::{connection::PgPooledConn, logger};
 
@@ -20,7 +21,7 @@ pub trait RefundDbExt: Sized {
     async fn filter_by_constraints(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
-        refund_list_details: &api_models::refunds::RefundListRequest,
+        refund_list_details: &refunds::RefundListConstraints,
         limit: i64,
         offset: i64,
     ) -> CustomResult<Vec<Self>, errors::DatabaseError>;
@@ -34,7 +35,7 @@ pub trait RefundDbExt: Sized {
     async fn get_refunds_count(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
-        refund_list_details: &api_models::refunds::RefundListRequest,
+        refund_list_details: &refunds::RefundListConstraints,
     ) -> CustomResult<i64, errors::DatabaseError>;
 }
 
@@ -43,7 +44,7 @@ impl RefundDbExt for Refund {
     async fn filter_by_constraints(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
-        refund_list_details: &api_models::refunds::RefundListRequest,
+        refund_list_details: &refunds::RefundListConstraints,
         limit: i64,
         offset: i64,
     ) -> CustomResult<Vec<Self>, errors::DatabaseError> {
@@ -88,7 +89,7 @@ impl RefundDbExt for Refund {
         match &refund_list_details.profile_id {
             Some(profile_id) => {
                 filter = filter
-                    .filter(dsl::profile_id.eq(profile_id.to_owned()))
+                    .filter(dsl::profile_id.eq_any(profile_id.to_owned()))
                     .limit(limit)
                     .offset(offset);
             }
@@ -206,7 +207,7 @@ impl RefundDbExt for Refund {
     async fn get_refunds_count(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
-        refund_list_details: &api_models::refunds::RefundListRequest,
+        refund_list_details: &refunds::RefundListConstraints,
     ) -> CustomResult<i64, errors::DatabaseError> {
         let mut filter = <Self as HasTable>::table()
             .count()
@@ -237,7 +238,7 @@ impl RefundDbExt for Refund {
             }
         }
         if let Some(profile_id) = &refund_list_details.profile_id {
-            filter = filter.filter(dsl::profile_id.eq(profile_id.to_owned()));
+            filter = filter.filter(dsl::profile_id.eq_any(profile_id.to_owned()));
         }
 
         if let Some(time_range) = refund_list_details.time_range {

crates/storage_impl/src/mock_db/payment_attempt.rs

@@ -45,6 +45,7 @@ impl PaymentAttemptInterface for MockDb {
         _payment_method_type: Option<Vec<PaymentMethodType>>,
         _authentication_type: Option<Vec<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)?

crates/storage_impl/src/payments/payment_attempt.rs

@@ -296,6 +296,7 @@ impl<T: DatabaseStore> PaymentAttemptInterface for RouterStore<T> {
         payment_method_type: Option<Vec<PaymentMethodType>>,
         authentication_type: Option<Vec<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
@@ -318,6 +319,7 @@ impl<T: DatabaseStore> PaymentAttemptInterface for RouterStore<T> {
             payment_method,
             payment_method_type,
             authentication_type,
+            profile_id_list,
             merchant_connector_id,
         )
         .await
@@ -1067,6 +1069,7 @@ impl<T: DatabaseStore> PaymentAttemptInterface for KVRouterStore<T> {
         payment_method_type: Option<Vec<PaymentMethodType>>,
         authentication_type: Option<Vec<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
@@ -1078,6 +1081,7 @@ impl<T: DatabaseStore> PaymentAttemptInterface for KVRouterStore<T> {
                 payment_method_type,
                 authentication_type,
                 merchant_connector_id,
+                profile_id_list,
                 storage_scheme,
             )
             .await

crates/storage_impl/src/payments/payment_intent.rs

@@ -553,7 +553,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
                     query = query.filter(pi_dsl::customer_id.eq(customer_id.clone()));
                 }
                 if let Some(profile_id) = &params.profile_id {
-                    query = query.filter(pi_dsl::profile_id.eq(profile_id.clone()));
+                    query = query.filter(pi_dsl::profile_id.eq_any(profile_id.clone()));
                 }
 
                 query = match (params.starting_at, &params.starting_after_id) {
@@ -758,7 +758,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
                 }
 
                 if let Some(profile_id) = &params.profile_id {
-                    query = query.filter(pi_dsl::profile_id.eq(profile_id.clone()));
+                    query = query.filter(pi_dsl::profile_id.eq_any(profile_id.clone()));
                 }
 
                 query = match (params.starting_at, &params.starting_after_id) {
@@ -922,7 +922,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
                     query = query.filter(pi_dsl::customer_id.eq(customer_id.clone()));
                 }
                 if let Some(profile_id) = &params.profile_id {
-                    query = query.filter(pi_dsl::profile_id.eq(profile_id.clone()));
+                    query = query.filter(pi_dsl::profile_id.eq_any(profile_id.clone()));
                 }
 
                 query = match params.starting_at {