#[cfg(feature = "olap")]
use std::collections::{HashMap, HashSet};

#[cfg(feature = "olap")]
use common_utils::types::{ConnectorTransactionIdTrait, MinorUnit};
use diesel_models::{errors::DatabaseError, refund as diesel_refund};
use hyperswitch_domain_models::refunds;

use super::MockDb;
use crate::{
    core::errors::{self, CustomResult},
    types::storage::enums,
};

#[cfg(feature = "olap")]
const MAX_LIMIT: usize = 100;

#[async_trait::async_trait]
pub trait RefundInterface {
    #[cfg(feature = "v1")]
    async fn find_refund_by_internal_reference_id_processor_merchant_id(
        &self,
        internal_reference_id: &str,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<diesel_refund::Refund, errors::StorageError>;

    #[cfg(feature = "v1")]
    async fn find_refund_by_payment_id_processor_merchant_id(
        &self,
        payment_id: &common_utils::id_type::PaymentId,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<Vec<diesel_refund::Refund>, errors::StorageError>;

    #[cfg(feature = "v1")]
    async fn find_refund_by_processor_merchant_id_refund_id(
        &self,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        refund_id: &str,
        storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<diesel_refund::Refund, errors::StorageError>;

    #[cfg(feature = "v1")]
    async fn find_refund_by_processor_merchant_id_connector_refund_id_connector(
        &self,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        connector_refund_id: &str,
        connector: &str,
        storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<diesel_refund::Refund, errors::StorageError>;

    async fn update_refund(
        &self,
        this: diesel_refund::Refund,
        refund: diesel_refund::RefundUpdate,
        storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<diesel_refund::Refund, errors::StorageError>;

    async fn find_refund_by_processor_merchant_id_connector_transaction_id(
        &self,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        connector_transaction_id: &str,
        storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<Vec<diesel_refund::Refund>, errors::StorageError>;

    #[cfg(feature = "v2")]
    async fn find_refund_by_id(
        &self,
        id: &common_utils::id_type::GlobalRefundId,
        storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<diesel_refund::Refund, errors::StorageError>;

    async fn insert_refund(
        &self,
        new: diesel_refund::RefundNew,
        storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<diesel_refund::Refund, errors::StorageError>;

    #[cfg(all(feature = "v1", feature = "olap"))]
    async fn filter_refund_by_constraints(
        &self,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        refund_details: &refunds::RefundListConstraints,
        storage_scheme: enums::MerchantStorageScheme,
        limit: i64,
        offset: i64,
    ) -> CustomResult<Vec<diesel_models::refund::Refund>, errors::StorageError>;

    #[cfg(all(feature = "v2", feature = "olap"))]
    async fn filter_refund_by_constraints(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        refund_details: refunds::RefundListConstraints,
        storage_scheme: enums::MerchantStorageScheme,
        limit: i64,
        offset: i64,
    ) -> CustomResult<Vec<diesel_models::refund::Refund>, errors::StorageError>;

    #[cfg(all(feature = "v1", feature = "olap"))]
    async fn filter_refund_by_meta_constraints(
        &self,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        refund_details: &common_utils::types::TimeRange,
        storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<api_models::refunds::RefundListMetaData, errors::StorageError>;

    #[cfg(all(feature = "v1", feature = "olap"))]
    async fn get_refund_status_with_count(
        &self,
        processor_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>;

    #[cfg(all(feature = "v1", feature = "olap"))]
    async fn get_total_count_of_refunds(
        &self,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        refund_details: &refunds::RefundListConstraints,
        storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<i64, errors::StorageError>;

    #[cfg(all(feature = "v2", feature = "olap"))]
    async fn get_total_count_of_refunds(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        refund_details: refunds::RefundListConstraints,
        storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<i64, errors::StorageError>;
}

#[cfg(not(feature = "kv_store"))]
mod storage {
    use error_stack::report;
    use hyperswitch_domain_models::refunds;
    use router_env::{instrument, tracing};

    use super::RefundInterface;
    use crate::{
        connection,
        core::errors::{self, CustomResult},
        services::Store,
        types::storage::{self as storage_types, enums},
    };

    #[async_trait::async_trait]
    impl RefundInterface for Store {
        #[cfg(feature = "v1")]
        #[instrument(skip_all)]
        async fn find_refund_by_internal_reference_id_processor_merchant_id(
            &self,
            internal_reference_id: &str,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            // Stagger release fallback: first try processor_merchant_id, if not found fallback to merchant_id
            // For old records processor_merchant_id is NULL, so we use merchant_id (which has the same value)
            let result =
                diesel_refund::Refund::find_by_internal_reference_id_processor_merchant_id(
                    &conn,
                    internal_reference_id,
                    processor_merchant_id,
                )
                .await;

            match result {
                Ok(refund) => Ok(refund),
                Err(error) => {
                    if matches!(
                        error.current_context(),
                        diesel_models::errors::DatabaseError::NotFound
                    ) {
                        diesel_refund::Refund::find_by_internal_reference_id_merchant_id(
                            &conn,
                            internal_reference_id,
                            processor_merchant_id,
                        )
                        .await
                        .map_err(|error| report!(errors::StorageError::from(error)))
                    } else {
                        Err(report!(errors::StorageError::from(error)))
                    }
                }
            }
        }

        #[instrument(skip_all)]
        async fn insert_refund(
            &self,
            new: diesel_refund::RefundNew,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            let conn = connection::pg_connection_write(self).await?;
            new.insert(&conn)
                .await
                .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[instrument(skip_all)]
        async fn find_refund_by_processor_merchant_id_connector_transaction_id(
            &self,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            connector_transaction_id: &str,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<Vec<diesel_refund::Refund>, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            diesel_refund::Refund::find_by_processor_merchant_id_connector_transaction_id(
                &conn,
                processor_merchant_id,
                connector_transaction_id,
            )
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(feature = "v2")]
        async fn find_refund_by_id(
            &self,
            id: &common_utils::id_type::GlobalRefundId,
            storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            diesel_refund::Refund::find_by_global_id(&conn, id)
                .await
                .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(feature = "v1")]
        #[instrument(skip_all)]
        async fn update_refund(
            &self,
            this: diesel_refund::Refund,
            refund: diesel_refund::RefundUpdate,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            let conn = connection::pg_connection_write(self).await?;
            // Stagger release fallback: first try processor_merchant_id, if not found fallback to merchant_id
            // For old records processor_merchant_id is NULL, so we use merchant_id (which has the same value)
            let result = this.clone().update(&conn, refund.clone()).await;

            match result {
                Ok(refund) => Ok(refund),
                Err(error) => {
                    if matches!(
                        error.current_context(),
                        diesel_models::errors::DatabaseError::NotFound
                    ) {
                        this.update_by_merchant_id(&conn, refund)
                            .await
                            .map_err(|error| report!(errors::StorageError::from(error)))
                    } else {
                        Err(report!(errors::StorageError::from(error)))
                    }
                }
            }
        }

        #[cfg(feature = "v2")]
        #[instrument(skip_all)]
        async fn update_refund(
            &self,
            this: diesel_refund::Refund,
            refund: diesel_refund::RefundUpdate,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            let conn = connection::pg_connection_write(self).await?;
            this.update_with_id(&conn, refund)
                .await
                .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(feature = "v1")]
        #[instrument(skip_all)]
        async fn find_refund_by_processor_merchant_id_refund_id(
            &self,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            refund_id: &str,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            // Stagger release fallback: first try processor_merchant_id, if not found fallback to merchant_id
            // For old records processor_merchant_id is NULL, so we use merchant_id (which has the same value)
            let result = diesel_refund::Refund::find_by_processor_merchant_id_refund_id(
                &conn,
                processor_merchant_id,
                refund_id,
            )
            .await;

            match result {
                Ok(refund) => Ok(refund),
                Err(error) => {
                    if matches!(
                        error.current_context(),
                        diesel_models::errors::DatabaseError::NotFound
                    ) {
                        diesel_refund::Refund::find_by_merchant_id_refund_id(
                            &conn,
                            processor_merchant_id,
                            refund_id,
                        )
                        .await
                        .map_err(|error| report!(errors::StorageError::from(error)))
                    } else {
                        Err(report!(errors::StorageError::from(error)))
                    }
                }
            }
        }

        #[cfg(feature = "v1")]
        #[instrument(skip_all)]
        async fn find_refund_by_processor_merchant_id_connector_refund_id_connector(
            &self,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            connector_refund_id: &str,
            connector: &str,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            // Stagger release fallback: first try processor_merchant_id, if not found fallback to merchant_id
            // For old records processor_merchant_id is NULL, so we use merchant_id (which has the same value)
            let result =
                diesel_refund::Refund::find_by_processor_merchant_id_connector_refund_id_connector(
                    &conn,
                    processor_merchant_id,
                    connector_refund_id,
                    connector,
                )
                .await;

            match result {
                Ok(refund) => Ok(refund),
                Err(error) => {
                    if matches!(
                        error.current_context(),
                        diesel_models::errors::DatabaseError::NotFound
                    ) {
                        diesel_refund::Refund::find_by_merchant_id_connector_refund_id_connector(
                            &conn,
                            processor_merchant_id,
                            connector_refund_id,
                            connector,
                        )
                        .await
                        .map_err(|error| report!(errors::StorageError::from(error)))
                    } else {
                        Err(report!(errors::StorageError::from(error)))
                    }
                }
            }
        }

        #[cfg(feature = "v1")]
        #[instrument(skip_all)]
        async fn find_refund_by_payment_id_processor_merchant_id(
            &self,
            payment_id: &common_utils::id_type::PaymentId,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<Vec<diesel_refund::Refund>, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            diesel_refund::Refund::find_by_payment_id_processor_merchant_id(
                &conn,
                payment_id,
                processor_merchant_id,
            )
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(all(feature = "v1", feature = "olap"))]
        #[instrument(skip_all)]
        async fn filter_refund_by_constraints(
            &self,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            refund_details: &refunds::RefundListConstraints,
            _storage_scheme: enums::MerchantStorageScheme,
            limit: i64,
            offset: i64,
        ) -> CustomResult<Vec<diesel_models::refund::Refund>, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            <diesel_models::refund::Refund as storage_types::RefundDbExt>::filter_by_constraints(
                &conn,
                processor_merchant_id,
                refund_details,
                limit,
                offset,
            )
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(all(feature = "v2", feature = "olap"))]
        #[instrument(skip_all)]
        async fn filter_refund_by_constraints(
            &self,
            merchant_id: &common_utils::id_type::MerchantId,
            refund_details: refunds::RefundListConstraints,
            _storage_scheme: enums::MerchantStorageScheme,
            limit: i64,
            offset: i64,
        ) -> CustomResult<Vec<diesel_models::refund::Refund>, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            <diesel_models::refund::Refund as storage_types::RefundDbExt>::filter_by_constraints(
                &conn,
                merchant_id,
                refund_details,
                limit,
                offset,
            )
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(all(feature = "v1", feature = "olap"))]
        #[instrument(skip_all)]
        async fn filter_refund_by_meta_constraints(
            &self,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            refund_details: &api_models::payments::TimeRange,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<api_models::refunds::RefundListMetaData, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            <diesel_models::refund::Refund as storage_types::RefundDbExt>::filter_by_meta_constraints(
                &conn,
                processor_merchant_id,
                refund_details,
            )
            .await
            .map_err(|error|report!(errors::StorageError::from(error)))
        }

        #[cfg(all(feature = "v1", feature = "olap"))]
        #[instrument(skip_all)]
        async fn get_refund_status_with_count(
            &self,
            processor_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, processor_merchant_id,profile_id_list, time_range)
            .await
            .map_err(|error|report!(errors::StorageError::from(error)))
        }

        #[cfg(all(feature = "v1", feature = "olap"))]
        #[instrument(skip_all)]
        async fn get_total_count_of_refunds(
            &self,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            refund_details: &refunds::RefundListConstraints,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<i64, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            <diesel_models::refund::Refund as storage_types::RefundDbExt>::get_refunds_count(
                &conn,
                processor_merchant_id,
                refund_details,
            )
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(all(feature = "v2", feature = "olap"))]
        #[instrument(skip_all)]
        async fn get_total_count_of_refunds(
            &self,
            merchant_id: &common_utils::id_type::MerchantId,
            refund_details: refunds::RefundListConstraints,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<i64, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            <diesel_models::refund::Refund as storage_types::RefundDbExt>::get_refunds_count(
                &conn,
                merchant_id,
                refund_details,
            )
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))
        }
    }
}

#[cfg(feature = "kv_store")]
mod storage {
    use common_utils::{ext_traits::Encode, fallback_reverse_lookup_not_found};
    use diesel_models::refund as diesel_refund;
    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::{
        decide_storage_scheme, kv_wrapper, KvOperation, Op, PartitionKey,
    };

    use super::RefundInterface;
    use crate::{
        connection,
        core::errors::{self, utils::RedisErrorExt, CustomResult},
        db::reverse_lookup::ReverseLookupInterface,
        services::Store,
        types::storage::{self as storage_types, enums},
        utils::db_utils,
    };
    #[async_trait::async_trait]
    impl RefundInterface for Store {
        #[cfg(feature = "v1")]
        #[instrument(skip_all)]
        async fn find_refund_by_internal_reference_id_processor_merchant_id(
            &self,
            internal_reference_id: &str,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            // Stagger release fallback: first try processor_merchant_id, if not found fallback to merchant_id
            // For old records processor_merchant_id is NULL, so we use merchant_id (which has the same value)
            let database_call = || async {
                let conn = connection::pg_connection_read(self).await?;
                let result =
                    diesel_refund::Refund::find_by_internal_reference_id_processor_merchant_id(
                        &conn,
                        internal_reference_id,
                        processor_merchant_id,
                    )
                    .await;

                match result {
                    Ok(refund) => Ok(refund),
                    Err(error) => {
                        if matches!(
                            error.current_context(),
                            diesel_models::errors::DatabaseError::NotFound
                        ) {
                            diesel_refund::Refund::find_by_internal_reference_id_merchant_id(
                                &conn,
                                internal_reference_id,
                                processor_merchant_id,
                            )
                            .await
                            .map_err(|error| report!(errors::StorageError::from(error)))
                        } else {
                            Err(report!(errors::StorageError::from(error)))
                        }
                    }
                }
            };
            let storage_scheme = Box::pin(decide_storage_scheme::<_, diesel_refund::Refund>(
                self,
                storage_scheme,
                Op::Find,
            ))
            .await;
            match storage_scheme {
                enums::MerchantStorageScheme::PostgresOnly => database_call().await,
                enums::MerchantStorageScheme::RedisKv => {
                    let lookup_id = diesel_refund::Refund::construct_lookup_id_processor_merchant_id_internal_reference_id(processor_merchant_id, internal_reference_id);
                    let lookup = fallback_reverse_lookup_not_found!(
                        self.get_lookup_by_lookup_id(&lookup_id, storage_scheme)
                            .await,
                        database_call().await
                    );

                    let key = PartitionKey::CombinationKey {
                        combination: &lookup.pk_id,
                    };
                    Box::pin(db_utils::try_redis_get_else_try_database_get(
                        async {
                            Box::pin(kv_wrapper(
                                self,
                                KvOperation::<diesel_refund::Refund>::HGet(&lookup.sk_id),
                                key,
                            ))
                            .await?
                            .try_into_hget()
                        },
                        database_call,
                    ))
                    .await
                }
            }
        }

        #[cfg(feature = "v1")]
        #[instrument(skip_all)]
        async fn insert_refund(
            &self,
            new: diesel_refund::RefundNew,
            storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            let storage_scheme = Box::pin(decide_storage_scheme::<_, diesel_refund::Refund>(
                self,
                storage_scheme,
                Op::Insert,
            ))
            .await;
            match storage_scheme {
                enums::MerchantStorageScheme::PostgresOnly => {
                    let conn = connection::pg_connection_write(self).await?;
                    new.insert(&conn)
                        .await
                        .map_err(|error| report!(errors::StorageError::from(error)))
                }
                enums::MerchantStorageScheme::RedisKv => {
                    let merchant_id = new.merchant_id.clone();
                    let payment_id = new.payment_id.clone();
                    let key = PartitionKey::MerchantIdPaymentId {
                        merchant_id: &merchant_id,
                        payment_id: &payment_id,
                    };
                    let key_str = key.to_string();
                    // TODO: need to add an application generated payment attempt id to distinguish between multiple attempts for the same payment id
                    // Check for database presence as well Maybe use a read replica here ?
                    let created_refund = diesel_refund::Refund {
                        refund_id: new.refund_id.clone(),
                        merchant_id: new.merchant_id.clone(),
                        attempt_id: new.attempt_id.clone(),
                        internal_reference_id: new.internal_reference_id.clone(),
                        payment_id: new.payment_id.clone(),
                        connector_transaction_id: new.connector_transaction_id.clone(),
                        connector: new.connector.clone(),
                        connector_refund_id: new.connector_refund_id.clone(),
                        external_reference_id: new.external_reference_id.clone(),
                        refund_type: new.refund_type,
                        total_amount: new.total_amount,
                        currency: new.currency,
                        refund_amount: new.refund_amount,
                        refund_status: new.refund_status,
                        sent_to_gateway: new.sent_to_gateway,
                        refund_error_message: None,
                        refund_error_code: None,
                        metadata: new.metadata.clone(),
                        refund_arn: new.refund_arn.clone(),
                        created_at: new.created_at,
                        modified_at: new.created_at,
                        description: new.description.clone(),
                        refund_reason: new.refund_reason.clone(),
                        profile_id: new.profile_id.clone(),
                        updated_by: new.updated_by.clone(),
                        merchant_connector_id: new.merchant_connector_id.clone(),
                        charges: new.charges.clone(),
                        split_refunds: new.split_refunds.clone(),
                        organization_id: new.organization_id.clone(),
                        unified_code: None,
                        unified_message: None,
                        processor_refund_data: new.processor_refund_data.clone(),
                        processor_transaction_data: new.processor_transaction_data.clone(),
                        issuer_error_code: None,
                        issuer_error_message: None,
                        processor_merchant_id: new.processor_merchant_id.clone(),
                        created_by: new.created_by.clone(),
                        // Below fields are deprecated. Please add any new fields above this line.
                        connector_refund_data: None,
                        connector_transaction_data: None,
                    };

                    let field = format!(
                        "pa_{}_ref_{}",
                        created_refund.attempt_id, created_refund.refund_id
                    );

                    let mut reverse_lookups = vec![
                        storage_types::ReverseLookupNew {
                            sk_id: field.clone(),
                            lookup_id: diesel_refund::Refund::construct_lookup_id_processor_merchant_id_refund_id(
                                &created_refund.merchant_id,
                                &created_refund.refund_id
                            ),
                            pk_id: key_str.clone(),
                            source: "refund".to_string(),
                            updated_by: storage_scheme.to_string(),
                        },
                        // [#492]: A discussion is required on whether this is required?
                        storage_types::ReverseLookupNew {
                            sk_id: field.clone(),
                            lookup_id: diesel_refund::Refund::construct_lookup_id_processor_merchant_id_internal_reference_id(
                                &created_refund.merchant_id,
                                &created_refund.internal_reference_id
                            ),
                            pk_id: key_str.clone(),
                            source: "refund".to_string(),
                            updated_by: storage_scheme.to_string(),
                        },
                    ];
                    if let Some(connector_refund_id) = &created_refund.connector_refund_id {
                        let lookup_id = diesel_refund::Refund::construct_lookup_id_processor_merchant_id_connector_refund_id_connector(
                            &created_refund.merchant_id,
                            connector_refund_id.get_id(),
                            &created_refund.connector
                        );
                        reverse_lookups.push(storage_types::ReverseLookupNew {
                            sk_id: field.clone(),
                            lookup_id,
                            pk_id: key_str.clone(),
                            source: "refund".to_string(),
                            updated_by: storage_scheme.to_string(),
                        })
                    };
                    let rev_look = reverse_lookups
                        .into_iter()
                        .map(|rev| self.insert_reverse_lookup(rev, storage_scheme));

                    futures::future::try_join_all(rev_look).await?;

                    let mut query_gen_conn = connection::pg_connection_write(self).await?;
                    let drainer_query = new
                        .generate_drainer_insert_query(&mut query_gen_conn)
                        .await
                        .change_context(errors::StorageError::KVError)
                        .attach_printable("Failed to generate refund insert query")?;

                    match Box::pin(kv_wrapper::<diesel_refund::Refund, _, _>(
                        self,
                        KvOperation::<diesel_refund::Refund>::HSetNx(
                            &field,
                            &created_refund,
                            drainer_query,
                        ),
                        key,
                    ))
                    .await
                    .map_err(|err| err.to_redis_failed_response(&key_str))?
                    .try_into_hsetnx()
                    {
                        Ok(HsetnxReply::KeyNotSet) => Err(errors::StorageError::DuplicateValue {
                            entity: "refund",
                            key: Some(created_refund.refund_id),
                        }
                        .into()),
                        Ok(HsetnxReply::KeySet) => Ok(created_refund),
                        Err(er) => Err(er).change_context(errors::StorageError::KVError),
                    }
                }
            }
        }

        #[cfg(feature = "v2")]
        #[instrument(skip_all)]
        async fn insert_refund(
            &self,
            new: diesel_refund::RefundNew,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            let conn = connection::pg_connection_write(self).await?;
            new.insert(&conn)
                .await
                .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(feature = "v1")]
        #[instrument(skip_all)]
        async fn find_refund_by_processor_merchant_id_connector_transaction_id(
            &self,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            connector_transaction_id: &str,
            storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<Vec<diesel_refund::Refund>, errors::StorageError> {
            let database_call = || async {
                let conn = connection::pg_connection_read(self).await?;
                diesel_refund::Refund::find_by_processor_merchant_id_connector_transaction_id(
                    &conn,
                    processor_merchant_id,
                    connector_transaction_id,
                )
                .await
                .map_err(|error| report!(errors::StorageError::from(error)))
            };
            let storage_scheme = Box::pin(decide_storage_scheme::<_, diesel_refund::Refund>(
                self,
                storage_scheme,
                Op::Find,
            ))
            .await;
            match storage_scheme {
                enums::MerchantStorageScheme::PostgresOnly => database_call().await,
                enums::MerchantStorageScheme::RedisKv => {
                    let lookup_id = format!(
                        "pa_conn_trans_{}_{connector_transaction_id}",
                        processor_merchant_id.get_string_repr()
                    );
                    let lookup = fallback_reverse_lookup_not_found!(
                        self.get_lookup_by_lookup_id(&lookup_id, storage_scheme)
                            .await,
                        database_call().await
                    );

                    let key = PartitionKey::CombinationKey {
                        combination: &lookup.pk_id,
                    };

                    let pattern = db_utils::generate_hscan_pattern_for_refund(&lookup.sk_id);

                    Box::pin(db_utils::try_redis_get_else_try_database_get(
                        async {
                            Box::pin(kv_wrapper(
                                self,
                                KvOperation::<diesel_refund::Refund>::Scan(&pattern),
                                key,
                            ))
                            .await?
                            .try_into_scan()
                        },
                        database_call,
                    ))
                    .await
                }
            }
        }

        #[cfg(feature = "v2")]
        #[instrument(skip_all)]
        async fn find_refund_by_processor_merchant_id_connector_transaction_id(
            &self,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            connector_transaction_id: &str,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<Vec<diesel_refund::Refund>, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            diesel_refund::Refund::find_by_merchant_id_connector_transaction_id(
                &conn,
                processor_merchant_id,
                connector_transaction_id,
            )
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(feature = "v1")]
        #[instrument(skip_all)]
        async fn update_refund(
            &self,
            this: diesel_refund::Refund,
            refund: diesel_refund::RefundUpdate,
            storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            let merchant_id = this.merchant_id.clone();
            let payment_id = this.payment_id.clone();
            let key = PartitionKey::MerchantIdPaymentId {
                merchant_id: &merchant_id,
                payment_id: &payment_id,
            };
            let field = format!("pa_{}_ref_{}", this.attempt_id, this.refund_id);
            let storage_scheme = Box::pin(decide_storage_scheme::<_, diesel_refund::Refund>(
                self,
                storage_scheme,
                Op::Update(key.clone(), &field, Some(&this.updated_by)),
            ))
            .await;
            match storage_scheme {
                enums::MerchantStorageScheme::PostgresOnly => {
                    let conn = connection::pg_connection_write(self).await?;
                    // Stagger release fallback: first try processor_merchant_id, if not found fallback to merchant_id
                    // For old records processor_merchant_id is NULL, so we use merchant_id (which has the same value)
                    let result = this.clone().update(&conn, refund.clone()).await;

                    match result {
                        Ok(refund) => Ok(refund),
                        Err(error) => {
                            if matches!(
                                error.current_context(),
                                diesel_models::errors::DatabaseError::NotFound
                            ) {
                                this.update_by_merchant_id(&conn, refund)
                                    .await
                                    .map_err(|error| report!(errors::StorageError::from(error)))
                            } else {
                                Err(report!(errors::StorageError::from(error)))
                            }
                        }
                    }
                }
                enums::MerchantStorageScheme::RedisKv => {
                    let key_str = key.to_string();
                    let updated_refund = refund.clone().apply_changeset(this.clone());

                    let old_connector_refund_id_option = this.connector_refund_id.clone();
                    let new_connector_refund_id_option = updated_refund.connector_refund_id.clone();

                    match (
                        old_connector_refund_id_option,
                        new_connector_refund_id_option,
                    ) {
                        (None, Some(new_connector_refund_id)) => {
                            // No connector_refund_id existed before
                            let lookup_id = diesel_refund::Refund::construct_lookup_id_processor_merchant_id_connector_refund_id_connector(
                                &updated_refund.merchant_id,
                                new_connector_refund_id.get_id(),
                                &updated_refund.connector
                            );
                            let reverse_lookup = storage_types::ReverseLookupNew {
                                sk_id: field.clone(),
                                lookup_id,
                                pk_id: key_str.clone(),
                                source: "refund".to_string(),
                                updated_by: storage_scheme.to_string(),
                            };
                            self.insert_reverse_lookup(reverse_lookup, storage_scheme)
                                .await?;
                        }
                        (Some(old_connector_refund_id), Some(new_connector_refund_id))
                            if old_connector_refund_id.ne(&new_connector_refund_id) =>
                        {
                            // connector_refund_id existed before but it is being updated in this update call
                            let lookup_id = diesel_refund::Refund::construct_lookup_id_processor_merchant_id_connector_refund_id_connector(
                                &updated_refund.merchant_id,
                                new_connector_refund_id.get_id(),
                                &updated_refund.connector
                            );
                            let reverse_lookup = storage_types::ReverseLookupNew {
                                sk_id: field.clone(),
                                lookup_id,
                                pk_id: key_str.clone(),
                                source: "refund".to_string(),
                                updated_by: storage_scheme.to_string(),
                            };
                            self.insert_reverse_lookup(reverse_lookup, storage_scheme)
                                .await?;
                        }
                        (_, _) => {}
                    }

                    let redis_value = updated_refund
                        .encode_to_string_of_json()
                        .change_context(errors::StorageError::SerializationFailed)?;

                    let mut query_gen_conn = connection::pg_connection_write(self).await?;
                    let drainer_query = refund
                        .generate_drainer_update_query(
                            &mut query_gen_conn,
                            this.refund_id.clone(),
                            this.processor_merchant_id
                                .clone()
                                .unwrap_or_else(|| merchant_id.clone()),
                        )
                        .await
                        .change_context(errors::StorageError::KVError)
                        .attach_printable("Failed to generate refund update query")?;

                    Box::pin(kv_wrapper::<(), _, _>(
                        self,
                        KvOperation::Hset::<diesel_refund::Refund>(
                            (&field, redis_value),
                            drainer_query,
                        ),
                        key,
                    ))
                    .await
                    .map_err(|err| err.to_redis_failed_response(&key_str))?
                    .try_into_hset()
                    .change_context(errors::StorageError::KVError)?;

                    Ok(updated_refund)
                }
            }
        }

        #[cfg(feature = "v2")]
        #[instrument(skip_all)]
        async fn update_refund(
            &self,
            this: diesel_refund::Refund,
            refund: diesel_refund::RefundUpdate,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            let conn = connection::pg_connection_write(self).await?;
            this.update_with_id(&conn, refund)
                .await
                .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(feature = "v1")]
        #[instrument(skip_all)]
        async fn find_refund_by_processor_merchant_id_refund_id(
            &self,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            refund_id: &str,
            storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            // Stagger release fallback: first try processor_merchant_id, if not found fallback to merchant_id
            // For old records processor_merchant_id is NULL, so we use merchant_id (which has the same value)
            let database_call = || async {
                let conn = connection::pg_connection_read(self).await?;
                let result = diesel_refund::Refund::find_by_processor_merchant_id_refund_id(
                    &conn,
                    processor_merchant_id,
                    refund_id,
                )
                .await;

                match result {
                    Ok(refund) => Ok(refund),
                    Err(error) => {
                        if matches!(
                            error.current_context(),
                            diesel_models::errors::DatabaseError::NotFound
                        ) {
                            diesel_refund::Refund::find_by_merchant_id_refund_id(
                                &conn,
                                processor_merchant_id,
                                refund_id,
                            )
                            .await
                            .map_err(|error| report!(errors::StorageError::from(error)))
                        } else {
                            Err(report!(errors::StorageError::from(error)))
                        }
                    }
                }
            };
            let storage_scheme = Box::pin(decide_storage_scheme::<_, diesel_refund::Refund>(
                self,
                storage_scheme,
                Op::Find,
            ))
            .await;
            match storage_scheme {
                enums::MerchantStorageScheme::PostgresOnly => database_call().await,
                enums::MerchantStorageScheme::RedisKv => {
                    let lookup_id =
                        diesel_refund::Refund::construct_lookup_id_processor_merchant_id_refund_id(
                            processor_merchant_id,
                            refund_id,
                        );
                    let lookup = fallback_reverse_lookup_not_found!(
                        self.get_lookup_by_lookup_id(&lookup_id, storage_scheme)
                            .await,
                        database_call().await
                    );

                    let key = PartitionKey::CombinationKey {
                        combination: &lookup.pk_id,
                    };
                    Box::pin(db_utils::try_redis_get_else_try_database_get(
                        async {
                            Box::pin(kv_wrapper(
                                self,
                                KvOperation::<diesel_refund::Refund>::HGet(&lookup.sk_id),
                                key,
                            ))
                            .await?
                            .try_into_hget()
                        },
                        database_call,
                    ))
                    .await
                }
            }
        }

        #[cfg(feature = "v1")]
        #[instrument(skip_all)]
        async fn find_refund_by_processor_merchant_id_connector_refund_id_connector(
            &self,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            connector_refund_id: &str,
            connector: &str,
            storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            // Stagger release fallback: first try processor_merchant_id, if not found fallback to merchant_id
            // For old records processor_merchant_id is NULL, so we use merchant_id (which has the same value)
            let database_call = || async {
                let conn = connection::pg_connection_read(self).await?;
                let result =
                    diesel_refund::Refund::find_by_processor_merchant_id_connector_refund_id_connector(
                        &conn,
                        processor_merchant_id,
                        connector_refund_id,
                        connector,
                    )
                    .await;

                match result {
                    Ok(refund) => Ok(refund),
                    Err(error) => {
                        if matches!(
                            error.current_context(),
                            diesel_models::errors::DatabaseError::NotFound
                        ) {
                            diesel_refund::Refund::find_by_merchant_id_connector_refund_id_connector(
                                &conn,
                                processor_merchant_id,
                                connector_refund_id,
                                connector,
                            )
                            .await
                            .map_err(|error| report!(errors::StorageError::from(error)))
                        } else {
                            Err(report!(errors::StorageError::from(error)))
                        }
                    }
                }
            };
            let storage_scheme = Box::pin(decide_storage_scheme::<_, diesel_refund::Refund>(
                self,
                storage_scheme,
                Op::Find,
            ))
            .await;
            match storage_scheme {
                enums::MerchantStorageScheme::PostgresOnly => database_call().await,
                enums::MerchantStorageScheme::RedisKv => {
                    let lookup_id = diesel_refund::Refund::construct_lookup_id_processor_merchant_id_connector_refund_id_connector(
                        processor_merchant_id,
                        connector_refund_id,
                        connector,
                    );
                    let lookup = fallback_reverse_lookup_not_found!(
                        self.get_lookup_by_lookup_id(&lookup_id, storage_scheme)
                            .await,
                        database_call().await
                    );

                    let key = PartitionKey::CombinationKey {
                        combination: &lookup.pk_id,
                    };
                    Box::pin(db_utils::try_redis_get_else_try_database_get(
                        async {
                            Box::pin(kv_wrapper(
                                self,
                                KvOperation::<diesel_refund::Refund>::HGet(&lookup.sk_id),
                                key,
                            ))
                            .await?
                            .try_into_hget()
                        },
                        database_call,
                    ))
                    .await
                }
            }
        }

        #[cfg(feature = "v1")]
        #[instrument(skip_all)]
        async fn find_refund_by_payment_id_processor_merchant_id(
            &self,
            payment_id: &common_utils::id_type::PaymentId,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<Vec<diesel_refund::Refund>, errors::StorageError> {
            let database_call = || async {
                let conn = connection::pg_connection_read(self).await?;
                diesel_refund::Refund::find_by_payment_id_processor_merchant_id(
                    &conn,
                    payment_id,
                    processor_merchant_id,
                )
                .await
                .map_err(|error| report!(errors::StorageError::from(error)))
            };
            let storage_scheme = Box::pin(decide_storage_scheme::<_, diesel_refund::Refund>(
                self,
                storage_scheme,
                Op::Find,
            ))
            .await;
            match storage_scheme {
                enums::MerchantStorageScheme::PostgresOnly => database_call().await,
                enums::MerchantStorageScheme::RedisKv => {
                    let key = PartitionKey::MerchantIdPaymentId {
                        merchant_id: processor_merchant_id,
                        payment_id,
                    };
                    Box::pin(db_utils::try_redis_get_else_try_database_get(
                        async {
                            Box::pin(kv_wrapper(
                                self,
                                KvOperation::<diesel_refund::Refund>::Scan("pa_*_ref_*"),
                                key,
                            ))
                            .await?
                            .try_into_scan()
                        },
                        database_call,
                    ))
                    .await
                }
            }
        }

        #[cfg(feature = "v2")]
        #[instrument(skip_all)]
        async fn find_refund_by_id(
            &self,
            id: &common_utils::id_type::GlobalRefundId,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            diesel_refund::Refund::find_by_global_id(&conn, id)
                .await
                .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(all(feature = "v1", feature = "olap"))]
        #[instrument(skip_all)]
        async fn filter_refund_by_constraints(
            &self,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            refund_details: &refunds::RefundListConstraints,
            _storage_scheme: enums::MerchantStorageScheme,
            limit: i64,
            offset: i64,
        ) -> CustomResult<Vec<diesel_models::refund::Refund>, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            <diesel_models::refund::Refund as storage_types::RefundDbExt>::filter_by_constraints(
                &conn,
                processor_merchant_id,
                refund_details,
                limit,
                offset,
            )
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(all(feature = "v2", feature = "olap"))]
        #[instrument(skip_all)]
        async fn filter_refund_by_constraints(
            &self,
            merchant_id: &common_utils::id_type::MerchantId,
            refund_details: refunds::RefundListConstraints,
            _storage_scheme: enums::MerchantStorageScheme,
            limit: i64,
            offset: i64,
        ) -> CustomResult<Vec<diesel_models::refund::Refund>, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            <diesel_models::refund::Refund as storage_types::RefundDbExt>::filter_by_constraints(
                &conn,
                merchant_id,
                refund_details,
                limit,
                offset,
            )
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(all(feature = "v1", feature = "olap"))]
        #[instrument(skip_all)]
        async fn filter_refund_by_meta_constraints(
            &self,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            refund_details: &common_utils::types::TimeRange,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<api_models::refunds::RefundListMetaData, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            <diesel_models::refund::Refund as storage_types::RefundDbExt>::filter_by_meta_constraints(&conn, processor_merchant_id, refund_details)
                        .await
                        .map_err(|error|report!(errors::StorageError::from(error)))
        }

        #[cfg(all(feature = "v1", feature = "olap"))]
        #[instrument(skip_all)]
        async fn get_refund_status_with_count(
            &self,
            processor_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, processor_merchant_id,profile_id_list, constraints)
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(all(feature = "v1", feature = "olap"))]
        #[instrument(skip_all)]
        async fn get_total_count_of_refunds(
            &self,
            processor_merchant_id: &common_utils::id_type::MerchantId,
            refund_details: &refunds::RefundListConstraints,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<i64, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            <diesel_models::refund::Refund as storage_types::RefundDbExt>::get_refunds_count(
                &conn,
                processor_merchant_id,
                refund_details,
            )
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))
        }

        #[cfg(all(feature = "v2", feature = "olap"))]
        #[instrument(skip_all)]
        async fn get_total_count_of_refunds(
            &self,
            merchant_id: &common_utils::id_type::MerchantId,
            refund_details: refunds::RefundListConstraints,
            _storage_scheme: enums::MerchantStorageScheme,
        ) -> CustomResult<i64, errors::StorageError> {
            let conn = connection::pg_connection_read(self).await?;
            <diesel_models::refund::Refund as storage_types::RefundDbExt>::get_refunds_count(
                &conn,
                merchant_id,
                refund_details,
            )
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))
        }
    }
}

#[async_trait::async_trait]
impl RefundInterface for MockDb {
    #[cfg(feature = "v1")]
    async fn find_refund_by_internal_reference_id_processor_merchant_id(
        &self,
        internal_reference_id: &str,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        _storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
        let refunds = self.refunds.lock().await;
        // Stagger release fallback: first try processor_merchant_id, if not found fallback to merchant_id
        // For old records processor_merchant_id is NULL, so we use merchant_id (which has the same value)
        refunds
            .iter()
            .find(|refund| {
                refund.processor_merchant_id.as_ref() == Some(processor_merchant_id)
                    && refund.internal_reference_id == internal_reference_id
            })
            .or_else(|| {
                refunds.iter().find(|refund| {
                    refund.merchant_id == *processor_merchant_id
                        && refund.internal_reference_id == internal_reference_id
                })
            })
            .cloned()
            .ok_or_else(|| {
                errors::StorageError::DatabaseError(DatabaseError::NotFound.into()).into()
            })
    }

    #[cfg(feature = "v1")]
    async fn insert_refund(
        &self,
        new: diesel_refund::RefundNew,
        _storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
        let mut refunds = self.refunds.lock().await;
        let current_time = common_utils::date_time::now();

        let refund = diesel_refund::Refund {
            internal_reference_id: new.internal_reference_id,
            refund_id: new.refund_id,
            payment_id: new.payment_id,
            merchant_id: new.merchant_id,
            attempt_id: new.attempt_id,
            connector_transaction_id: new.connector_transaction_id,
            connector: new.connector,
            connector_refund_id: new.connector_refund_id,
            external_reference_id: new.external_reference_id,
            refund_type: new.refund_type,
            total_amount: new.total_amount,
            currency: new.currency,
            refund_amount: new.refund_amount,
            refund_status: new.refund_status,
            sent_to_gateway: new.sent_to_gateway,
            refund_error_message: None,
            refund_error_code: None,
            metadata: new.metadata,
            refund_arn: new.refund_arn.clone(),
            created_at: new.created_at,
            modified_at: current_time,
            description: new.description,
            refund_reason: new.refund_reason.clone(),
            profile_id: new.profile_id,
            updated_by: new.updated_by,
            merchant_connector_id: new.merchant_connector_id,
            charges: new.charges,
            split_refunds: new.split_refunds,
            organization_id: new.organization_id,
            unified_code: None,
            unified_message: None,
            processor_refund_data: new.processor_refund_data.clone(),
            processor_transaction_data: new.processor_transaction_data.clone(),
            issuer_error_code: None,
            issuer_error_message: None,
            processor_merchant_id: new.processor_merchant_id.clone(),
            created_by: new.created_by.clone(),
            // Below fields are deprecated. Please add any new fields above this line.
            connector_refund_data: None,
            connector_transaction_data: None,
        };
        refunds.push(refund.clone());
        Ok(refund)
    }

    #[cfg(feature = "v2")]
    async fn insert_refund(
        &self,
        new: diesel_refund::RefundNew,
        _storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
        let mut refunds = self.refunds.lock().await;
        let current_time = common_utils::date_time::now();

        let refund = diesel_refund::Refund {
            id: new.id,
            merchant_reference_id: new.merchant_reference_id,
            payment_id: new.payment_id,
            merchant_id: new.merchant_id,
            attempt_id: new.attempt_id,
            connector_transaction_id: new.connector_transaction_id,
            connector: new.connector,
            connector_refund_id: new.connector_refund_id,
            external_reference_id: new.external_reference_id,
            refund_type: new.refund_type,
            total_amount: new.total_amount,
            currency: new.currency,
            refund_amount: new.refund_amount,
            refund_status: new.refund_status,
            sent_to_gateway: new.sent_to_gateway,
            refund_error_message: None,
            refund_error_code: None,
            metadata: new.metadata,
            refund_arn: new.refund_arn.clone(),
            created_at: new.created_at,
            modified_at: current_time,
            description: new.description,
            refund_reason: new.refund_reason.clone(),
            profile_id: new.profile_id,
            updated_by: new.updated_by,
            connector_id: new.connector_id,
            charges: new.charges,
            split_refunds: new.split_refunds,
            organization_id: new.organization_id,
            unified_code: None,
            unified_message: None,
            processor_refund_data: new.processor_refund_data.clone(),
            processor_transaction_data: new.processor_transaction_data.clone(),
            processor_merchant_id: new.processor_merchant_id.clone(),
            created_by: new.created_by.clone(),
        };
        refunds.push(refund.clone());
        Ok(refund)
    }

    async fn find_refund_by_processor_merchant_id_connector_transaction_id(
        &self,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        connector_transaction_id: &str,
        _storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<Vec<diesel_refund::Refund>, errors::StorageError> {
        let refunds = self.refunds.lock().await;

        Ok(refunds
            .iter()
            .filter(|refund| {
                (refund.processor_merchant_id.as_ref() == Some(processor_merchant_id)
                    || (refund.processor_merchant_id.is_none()
                        && refund.merchant_id == *processor_merchant_id))
                    && refund.get_connector_transaction_id() == connector_transaction_id
            })
            .cloned()
            .collect::<Vec<_>>())
    }

    #[cfg(feature = "v1")]
    async fn update_refund(
        &self,
        this: diesel_refund::Refund,
        refund: diesel_refund::RefundUpdate,
        _storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
        let mut refunds = self.refunds.lock().await;
        // Stagger release fallback: first try processor_merchant_id, if not found fallback to merchant_id
        // For old records processor_merchant_id is NULL, so we use merchant_id (which has the same value)
        let processor_merchant_id = this
            .processor_merchant_id
            .clone()
            .unwrap_or_else(|| this.merchant_id.clone());
        let index = refunds
            .iter()
            .position(|r| {
                r.refund_id == this.refund_id
                    && r.processor_merchant_id.as_ref() == Some(&processor_merchant_id)
            })
            .or_else(|| {
                refunds.iter().position(|r| {
                    r.refund_id == this.refund_id && r.merchant_id == processor_merchant_id
                })
            });

        index
            .and_then(|idx| {
                refunds.get_mut(idx).map(|r| {
                    let refund_updated =
                        diesel_refund::RefundUpdateInternal::from(refund).create_refund(r.clone());
                    *r = refund_updated.clone();
                    refund_updated
                })
            })
            .ok_or_else(|| {
                errors::StorageError::ValueNotFound("cannot find refund to update".to_string())
                    .into()
            })
    }

    #[cfg(feature = "v2")]
    async fn update_refund(
        &self,
        this: diesel_refund::Refund,
        refund: diesel_refund::RefundUpdate,
        _storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
        self.refunds
            .lock()
            .await
            .iter_mut()
            .find(|refund| this.merchant_reference_id == refund.merchant_reference_id)
            .map(|r| {
                let refund_updated =
                    diesel_refund::RefundUpdateInternal::from(refund).create_refund(r.clone());
                *r = refund_updated.clone();
                refund_updated
            })
            .ok_or_else(|| {
                errors::StorageError::ValueNotFound("cannot find refund to update".to_string())
                    .into()
            })
    }

    #[cfg(feature = "v1")]
    async fn find_refund_by_processor_merchant_id_refund_id(
        &self,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        refund_id: &str,
        _storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
        let refunds = self.refunds.lock().await;
        // Stagger release fallback: first try processor_merchant_id, if not found fallback to merchant_id
        // For old records processor_merchant_id is NULL, so we use merchant_id (which has the same value)
        refunds
            .iter()
            .find(|refund| {
                refund.processor_merchant_id.as_ref() == Some(processor_merchant_id)
                    && refund.refund_id == refund_id
            })
            .or_else(|| {
                refunds.iter().find(|refund| {
                    refund.merchant_id == *processor_merchant_id && refund.refund_id == refund_id
                })
            })
            .cloned()
            .ok_or_else(|| {
                errors::StorageError::DatabaseError(DatabaseError::NotFound.into()).into()
            })
    }

    #[cfg(feature = "v1")]
    async fn find_refund_by_processor_merchant_id_connector_refund_id_connector(
        &self,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        connector_refund_id: &str,
        connector: &str,
        _storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
        let refunds = self.refunds.lock().await;
        // Stagger release fallback: first try processor_merchant_id, if not found fallback to merchant_id
        // For old records processor_merchant_id is NULL, so we use merchant_id (which has the same value)
        refunds
            .iter()
            .find(|refund| {
                refund.processor_merchant_id.as_ref() == Some(processor_merchant_id)
                    && refund
                        .get_optional_connector_refund_id()
                        .map(|refund_id| refund_id.as_str())
                        == Some(connector_refund_id)
                    && refund.connector == connector
            })
            .or_else(|| {
                refunds.iter().find(|refund| {
                    refund.merchant_id == *processor_merchant_id
                        && refund
                            .get_optional_connector_refund_id()
                            .map(|refund_id| refund_id.as_str())
                            == Some(connector_refund_id)
                        && refund.connector == connector
                })
            })
            .cloned()
            .ok_or_else(|| {
                errors::StorageError::DatabaseError(DatabaseError::NotFound.into()).into()
            })
    }

    #[cfg(feature = "v1")]
    async fn find_refund_by_payment_id_processor_merchant_id(
        &self,
        payment_id: &common_utils::id_type::PaymentId,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        _storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<Vec<diesel_refund::Refund>, errors::StorageError> {
        let refunds = self.refunds.lock().await;

        Ok(refunds
            .iter()
            .filter(|refund| {
                (refund.processor_merchant_id.as_ref() == Some(processor_merchant_id)
                    || (refund.processor_merchant_id.is_none()
                        && refund.merchant_id == *processor_merchant_id))
                    && refund.payment_id == *payment_id
            })
            .cloned()
            .collect::<Vec<_>>())
    }

    #[cfg(feature = "v2")]
    async fn find_refund_by_id(
        &self,
        id: &common_utils::id_type::GlobalRefundId,
        _storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<diesel_refund::Refund, errors::StorageError> {
        let refunds = self.refunds.lock().await;

        refunds
            .iter()
            .find(|refund| refund.id == *id)
            .cloned()
            .ok_or_else(|| {
                errors::StorageError::DatabaseError(DatabaseError::NotFound.into()).into()
            })
    }

    #[cfg(all(feature = "v1", feature = "olap"))]
    async fn filter_refund_by_constraints(
        &self,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        refund_details: &refunds::RefundListConstraints,
        _storage_scheme: enums::MerchantStorageScheme,
        limit: i64,
        offset: i64,
    ) -> CustomResult<Vec<diesel_models::refund::Refund>, 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 {
            connectors.iter().for_each(|connector| {
                unique_connectors.insert(connector);
            });
        }

        if let Some(merchant_connector_ids) = &refund_details.merchant_connector_id {
            merchant_connector_ids
                .iter()
                .for_each(|unique_merchant_connector_id| {
                    unique_merchant_connector_ids.insert(unique_merchant_connector_id);
                });
        }

        if let Some(currencies) = &refund_details.currency {
            currencies.iter().for_each(|currency| {
                unique_currencies.insert(currency);
            });
        }

        if let Some(refund_statuses) = &refund_details.refund_status {
            refund_statuses.iter().for_each(|refund_status| {
                unique_statuses.insert(refund_status);
            });
        }

        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()
            .filter(|refund| {
                refund.processor_merchant_id.as_ref() == Some(processor_merchant_id)
                    || (refund.processor_merchant_id.is_none()
                        && refund.merchant_id == *processor_merchant_id)
            })
            .filter(|refund| {
                refund_details
                    .payment_id
                    .clone()
                    .is_none_or(|id| id == refund.payment_id)
            })
            .filter(|refund| {
                refund_details
                    .refund_id
                    .clone()
                    .is_none_or(|id| id == refund.refund_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(
                        common_utils::date_time::now() - time::Duration::days(60),
                        |range| range.start_time,
                    )
                    && refund.created_at
                        <= refund_details
                            .time_range
                            .map_or(common_utils::date_time::now(), |range| {
                                range.end_time.unwrap_or_else(common_utils::date_time::now)
                            })
            })
            .filter(|refund| {
                refund_details.amount_filter.as_ref().is_none_or(|amount| {
                    refund.refund_amount >= MinorUnit::new(amount.start_amount.unwrap_or(i64::MIN))
                        && refund.refund_amount
                            <= MinorUnit::new(amount.end_amount.unwrap_or(i64::MAX))
                })
            })
            .filter(|refund| {
                unique_connectors.is_empty() || unique_connectors.contains(&refund.connector)
            })
            .filter(|refund| {
                unique_merchant_connector_ids.is_empty()
                    || refund
                        .merchant_connector_id
                        .as_ref()
                        .is_some_and(|id| unique_merchant_connector_ids.contains(id))
            })
            .filter(|refund| {
                unique_currencies.is_empty() || unique_currencies.contains(&refund.currency)
            })
            .filter(|refund| {
                unique_statuses.is_empty() || unique_statuses.contains(&refund.refund_status)
            })
            .skip(usize::try_from(offset).unwrap_or_default())
            .take(usize::try_from(limit).unwrap_or(MAX_LIMIT))
            .cloned()
            .collect::<Vec<_>>();

        Ok(filtered_refunds)
    }

    #[cfg(all(feature = "v2", feature = "olap"))]
    async fn filter_refund_by_constraints(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        refund_details: refunds::RefundListConstraints,
        _storage_scheme: enums::MerchantStorageScheme,
        limit: i64,
        offset: i64,
    ) -> CustomResult<Vec<diesel_models::refund::Refund>, errors::StorageError> {
        let mut unique_connectors = HashSet::new();
        let mut unique_connector_ids = HashSet::new();
        let mut unique_currencies = HashSet::new();
        let mut unique_statuses = HashSet::new();

        // Fill the hash sets with data from refund_details
        if let Some(connectors) = &refund_details.connector {
            connectors.iter().for_each(|connector| {
                unique_connectors.insert(connector);
            });
        }

        if let Some(connector_id_list) = &refund_details.connector_id_list {
            connector_id_list.iter().for_each(|unique_connector_id| {
                unique_connector_ids.insert(unique_connector_id);
            });
        }

        if let Some(currencies) = &refund_details.currency {
            currencies.iter().for_each(|currency| {
                unique_currencies.insert(currency);
            });
        }

        if let Some(refund_statuses) = &refund_details.refund_status {
            refund_statuses.iter().for_each(|refund_status| {
                unique_statuses.insert(refund_status);
            });
        }

        let refunds = self.refunds.lock().await;
        let filtered_refunds = refunds
            .iter()
            .filter(|refund| refund.merchant_id == *merchant_id)
            .filter(|refund| {
                refund_details
                    .payment_id
                    .clone()
                    .is_none_or(|id| id == refund.payment_id)
            })
            .filter(|refund| {
                refund_details
                    .refund_id
                    .clone()
                    .is_none_or(|id| id == refund.id)
            })
            .filter(|refund| {
                refund
                    .profile_id
                    .as_ref()
                    .is_some_and(|profile_id| profile_id == &refund_details.profile_id)
            })
            .filter(|refund| {
                refund.created_at
                    >= refund_details.time_range.map_or(
                        common_utils::date_time::now() - time::Duration::days(60),
                        |range| range.start_time,
                    )
                    && refund.created_at
                        <= refund_details
                            .time_range
                            .map_or(common_utils::date_time::now(), |range| {
                                range.end_time.unwrap_or_else(common_utils::date_time::now)
                            })
            })
            .filter(|refund| {
                refund_details.amount_filter.as_ref().is_none_or(|amount| {
                    refund.refund_amount >= MinorUnit::new(amount.start_amount.unwrap_or(i64::MIN))
                        && refund.refund_amount
                            <= MinorUnit::new(amount.end_amount.unwrap_or(i64::MAX))
                })
            })
            .filter(|refund| {
                unique_connectors.is_empty() || unique_connectors.contains(&refund.connector)
            })
            .filter(|refund| {
                unique_connector_ids.is_empty()
                    || refund
                        .connector_id
                        .as_ref()
                        .is_some_and(|id| unique_connector_ids.contains(id))
            })
            .filter(|refund| {
                unique_currencies.is_empty() || unique_currencies.contains(&refund.currency)
            })
            .filter(|refund| {
                unique_statuses.is_empty() || unique_statuses.contains(&refund.refund_status)
            })
            .skip(usize::try_from(offset).unwrap_or_default())
            .take(usize::try_from(limit).unwrap_or(MAX_LIMIT))
            .cloned()
            .collect::<Vec<_>>();

        Ok(filtered_refunds)
    }

    #[cfg(all(feature = "v1", feature = "olap"))]
    async fn filter_refund_by_meta_constraints(
        &self,
        _processor_merchant_id: &common_utils::id_type::MerchantId,
        refund_details: &common_utils::types::TimeRange,
        _storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<api_models::refunds::RefundListMetaData, errors::StorageError> {
        let refunds = self.refunds.lock().await;

        let start_time = refund_details.start_time;
        let end_time = refund_details
            .end_time
            .unwrap_or_else(common_utils::date_time::now);

        let filtered_refunds = refunds
            .iter()
            .filter(|refund| refund.created_at >= start_time && refund.created_at <= end_time)
            .cloned()
            .collect::<Vec<diesel_models::refund::Refund>>();

        let mut refund_meta_data = api_models::refunds::RefundListMetaData {
            connector: vec![],
            currency: vec![],
            refund_status: vec![],
        };

        let mut unique_connectors = HashSet::new();
        let mut unique_currencies = HashSet::new();
        let mut unique_statuses = HashSet::new();

        for refund in filtered_refunds.into_iter() {
            unique_connectors.insert(refund.connector);

            let currency: api_models::enums::Currency = refund.currency;
            unique_currencies.insert(currency);

            let status: api_models::enums::RefundStatus = refund.refund_status;
            unique_statuses.insert(status);
        }

        refund_meta_data.connector = unique_connectors.into_iter().collect();
        refund_meta_data.currency = unique_currencies.into_iter().collect();
        refund_meta_data.refund_status = unique_statuses.into_iter().collect();

        Ok(refund_meta_data)
    }

    #[cfg(all(feature = "v1", feature = "olap"))]
    async fn get_refund_status_with_count(
        &self,
        _processor_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> {
        let refunds = self.refunds.lock().await;

        let start_time = time_range.start_time;
        let end_time = time_range
            .end_time
            .unwrap_or_else(common_utils::date_time::now);

        let filtered_refunds = refunds
            .iter()
            .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>>();

        let mut refund_status_counts: HashMap<api_models::enums::RefundStatus, i64> =
            HashMap::new();

        for refund in filtered_refunds {
            *refund_status_counts
                .entry(refund.refund_status)
                .or_insert(0) += 1;
        }

        let result: Vec<(api_models::enums::RefundStatus, i64)> =
            refund_status_counts.into_iter().collect();

        Ok(result)
    }

    #[cfg(all(feature = "v1", feature = "olap"))]
    async fn get_total_count_of_refunds(
        &self,
        processor_merchant_id: &common_utils::id_type::MerchantId,
        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 {
            connectors.iter().for_each(|connector| {
                unique_connectors.insert(connector);
            });
        }

        if let Some(merchant_connector_ids) = &refund_details.merchant_connector_id {
            merchant_connector_ids
                .iter()
                .for_each(|unique_merchant_connector_id| {
                    unique_merchant_connector_ids.insert(unique_merchant_connector_id);
                });
        }

        if let Some(currencies) = &refund_details.currency {
            currencies.iter().for_each(|currency| {
                unique_currencies.insert(currency);
            });
        }

        if let Some(refund_statuses) = &refund_details.refund_status {
            refund_statuses.iter().for_each(|refund_status| {
                unique_statuses.insert(refund_status);
            });
        }

        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()
            .filter(|refund| {
                refund.processor_merchant_id.as_ref() == Some(processor_merchant_id)
                    || (refund.processor_merchant_id.is_none()
                        && refund.merchant_id == *processor_merchant_id)
            })
            .filter(|refund| {
                refund_details
                    .payment_id
                    .clone()
                    .is_none_or(|id| id == refund.payment_id)
            })
            .filter(|refund| {
                refund_details
                    .refund_id
                    .clone()
                    .is_none_or(|id| id == refund.refund_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(
                        common_utils::date_time::now() - time::Duration::days(60),
                        |range| range.start_time,
                    )
                    && refund.created_at
                        <= refund_details
                            .time_range
                            .map_or(common_utils::date_time::now(), |range| {
                                range.end_time.unwrap_or_else(common_utils::date_time::now)
                            })
            })
            .filter(|refund| {
                refund_details.amount_filter.as_ref().is_none_or(|amount| {
                    refund.refund_amount >= MinorUnit::new(amount.start_amount.unwrap_or(i64::MIN))
                        && refund.refund_amount
                            <= MinorUnit::new(amount.end_amount.unwrap_or(i64::MAX))
                })
            })
            .filter(|refund| {
                unique_connectors.is_empty() || unique_connectors.contains(&refund.connector)
            })
            .filter(|refund| {
                unique_merchant_connector_ids.is_empty()
                    || refund
                        .merchant_connector_id
                        .as_ref()
                        .is_some_and(|id| unique_merchant_connector_ids.contains(id))
            })
            .filter(|refund| {
                unique_currencies.is_empty() || unique_currencies.contains(&refund.currency)
            })
            .filter(|refund| {
                unique_statuses.is_empty() || unique_statuses.contains(&refund.refund_status)
            })
            .cloned()
            .collect::<Vec<_>>();

        let filtered_refunds_count = filtered_refunds.len().try_into().unwrap_or_default();

        Ok(filtered_refunds_count)
    }

    #[cfg(all(feature = "v2", feature = "olap"))]
    async fn get_total_count_of_refunds(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        refund_details: refunds::RefundListConstraints,
        _storage_scheme: enums::MerchantStorageScheme,
    ) -> CustomResult<i64, errors::StorageError> {
        let mut unique_connectors = HashSet::new();
        let mut unique_connector_ids = HashSet::new();
        let mut unique_currencies = HashSet::new();
        let mut unique_statuses = HashSet::new();

        // Fill the hash sets with data from refund_details
        if let Some(connectors) = &refund_details.connector {
            connectors.iter().for_each(|connector| {
                unique_connectors.insert(connector);
            });
        }

        if let Some(connector_id_list) = &refund_details.connector_id_list {
            connector_id_list.iter().for_each(|unique_connector_id| {
                unique_connector_ids.insert(unique_connector_id);
            });
        }

        if let Some(currencies) = &refund_details.currency {
            currencies.iter().for_each(|currency| {
                unique_currencies.insert(currency);
            });
        }

        if let Some(refund_statuses) = &refund_details.refund_status {
            refund_statuses.iter().for_each(|refund_status| {
                unique_statuses.insert(refund_status);
            });
        }

        let refunds = self.refunds.lock().await;
        let filtered_refunds = refunds
            .iter()
            .filter(|refund| refund.merchant_id == *merchant_id)
            .filter(|refund| {
                refund_details
                    .payment_id
                    .clone()
                    .is_none_or(|id| id == refund.payment_id)
            })
            .filter(|refund| {
                refund_details
                    .refund_id
                    .clone()
                    .is_none_or(|id| id == refund.id)
            })
            .filter(|refund| {
                refund
                    .profile_id
                    .as_ref()
                    .is_some_and(|profile_id| profile_id == &refund_details.profile_id)
            })
            .filter(|refund| {
                refund.created_at
                    >= refund_details.time_range.map_or(
                        common_utils::date_time::now() - time::Duration::days(60),
                        |range| range.start_time,
                    )
                    && refund.created_at
                        <= refund_details
                            .time_range
                            .map_or(common_utils::date_time::now(), |range| {
                                range.end_time.unwrap_or_else(common_utils::date_time::now)
                            })
            })
            .filter(|refund| {
                refund_details.amount_filter.as_ref().is_none_or(|amount| {
                    refund.refund_amount >= MinorUnit::new(amount.start_amount.unwrap_or(i64::MIN))
                        && refund.refund_amount
                            <= MinorUnit::new(amount.end_amount.unwrap_or(i64::MAX))
                })
            })
            .filter(|refund| {
                unique_connectors.is_empty() || unique_connectors.contains(&refund.connector)
            })
            .filter(|refund| {
                unique_connector_ids.is_empty()
                    || refund
                        .connector_id
                        .as_ref()
                        .is_some_and(|id| unique_connector_ids.contains(id))
            })
            .filter(|refund| {
                unique_currencies.is_empty() || unique_currencies.contains(&refund.currency)
            })
            .filter(|refund| {
                unique_statuses.is_empty() || unique_statuses.contains(&refund.refund_status)
            })
            .cloned()
            .collect::<Vec<_>>();

        let filtered_refunds_count = filtered_refunds.len().try_into().unwrap_or_default();

        Ok(filtered_refunds_count)
    }
}
