use std::collections::HashSet;

use common_utils::ext_traits::AsyncExt;
use error_stack::{report, ResultExt};
use futures::future::try_join_all;
use router_env::{instrument, tracing};

use super::{MockDb, Store};
use crate::{
    connection,
    core::errors::{self, CustomResult},
    types::{
        domain::{
            self,
            behaviour::{Conversion, ReverseConversion},
        },
        storage,
    },
};

#[async_trait::async_trait]
pub trait EventInterface
where
    domain::Event:
        Conversion<DstType = storage::events::Event, NewDstType = storage::events::EventNew>,
{
    async fn insert_event(
        &self,
        event: domain::Event,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError>;

    async fn find_event_by_event_id(
        &self,
        event_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError>;

    async fn find_event_by_merchant_id_event_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        event_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError>;

    async fn find_event_by_initiator_merchant_id_idempotent_event_id(
        &self,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        idempotent_event_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError>;

    async fn find_event_by_merchant_id_idempotent_event_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        idempotent_event_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError>;

    #[allow(clippy::too_many_arguments)]
    async fn list_initial_events_by_initiator_merchant_id_constraints(
        &self,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        limit: Option<i64>,
        offset: Option<i64>,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError>;

    #[allow(clippy::too_many_arguments)]
    async fn list_initial_events_by_merchant_id_constraints(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        limit: Option<i64>,
        offset: Option<i64>,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError>;

    async fn list_events_by_initiator_merchant_id_initial_attempt_id(
        &self,
        initial_attempt_id: &str,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError>;

    async fn list_events_by_merchant_id_initial_attempt_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        initial_attempt_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError>;

    async fn list_initial_events_by_initiator_merchant_id_primary_object_id(
        &self,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        primary_object_id: &str,
        profile_id: Option<common_utils::id_type::ProfileId>,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError>;

    async fn list_initial_events_by_merchant_id_primary_object_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        primary_object_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError>;

    async fn find_initial_event_by_merchant_id_initial_attempt_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        initial_attempt_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<Option<domain::Event>, errors::StorageError>;

    #[allow(clippy::too_many_arguments)]
    async fn list_initial_events_by_profile_id_constraints(
        &self,
        profile_id: &common_utils::id_type::ProfileId,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        limit: Option<i64>,
        offset: Option<i64>,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError>;

    async fn list_initial_events_by_profile_id_primary_object_id(
        &self,
        profile_id: &common_utils::id_type::ProfileId,
        primary_object_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError>;

    async fn find_initial_event_by_profile_id_initial_attempt_id(
        &self,
        profile_id: &common_utils::id_type::ProfileId,
        initial_attempt_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<Option<domain::Event>, errors::StorageError>;

    async fn update_event_by_event_id(
        &self,
        event_id: &str,
        event: domain::EventUpdate,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError>;

    async fn update_event_by_merchant_id_event_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        event_id: &str,
        event: domain::EventUpdate,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError>;

    #[allow(clippy::too_many_arguments)]
    async fn count_initial_events_by_initiator_merchant_id_constraints(
        &self,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        profile_id: Option<common_utils::id_type::ProfileId>,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<i64, errors::StorageError>;

    async fn count_initial_events_by_profile_id_constraints(
        &self,
        profile_id: &common_utils::id_type::ProfileId,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<i64, errors::StorageError>;

    #[allow(clippy::too_many_arguments)]
    async fn count_initial_events_by_constraints(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        profile_id: Option<common_utils::id_type::ProfileId>,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<i64, errors::StorageError>;
}

#[async_trait::async_trait]
impl EventInterface for Store {
    #[instrument(skip_all)]
    async fn insert_event(
        &self,
        event: domain::Event,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError> {
        let conn = connection::pg_connection_write(self).await?;
        event
            .construct_new()
            .await
            .change_context(errors::StorageError::EncryptionError)?
            .insert(&conn)
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))?
            .convert(
                self.get_keymanager_state()
                    .attach_printable("Missing KeyManagerState")?,
                merchant_key_store.key.get_inner(),
                merchant_key_store.merchant_id.clone().into(),
            )
            .await
            .change_context(errors::StorageError::DecryptionError)
    }

    #[instrument(skip_all)]
    async fn find_event_by_event_id(
        &self,
        event_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::find_by_event_id(&conn, event_id)
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))?
            .convert(
                self.get_keymanager_state()
                    .attach_printable("Missing KeyManagerState")?,
                merchant_key_store.key.get_inner(),
                merchant_key_store.merchant_id.clone().into(),
            )
            .await
            .change_context(errors::StorageError::DecryptionError)
    }

    #[instrument(skip_all)]
    async fn find_event_by_merchant_id_event_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        event_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::find_by_merchant_id_event_id(&conn, merchant_id, event_id)
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))?
            .convert(
                self.get_keymanager_state()
                    .attach_printable("Missing KeyManagerState")?,
                merchant_key_store.key.get_inner(),
                merchant_key_store.merchant_id.clone().into(),
            )
            .await
            .change_context(errors::StorageError::DecryptionError)
    }

    #[instrument(skip_all)]
    async fn find_event_by_merchant_id_idempotent_event_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        idempotent_event_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::find_by_merchant_id_idempotent_event_id(
            &conn,
            merchant_id,
            idempotent_event_id,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))?
        .convert(
            self.get_keymanager_state()
                .attach_printable("Missing KeyManagerState")?,
            merchant_key_store.key.get_inner(),
            merchant_key_store.merchant_id.clone().into(),
        )
        .await
        .change_context(errors::StorageError::DecryptionError)
    }

    #[instrument(skip_all)]
    async fn find_event_by_initiator_merchant_id_idempotent_event_id(
        &self,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        idempotent_event_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::find_by_initiator_merchant_id_idempotent_event_id(
            &conn,
            initiator_merchant_id,
            idempotent_event_id,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))?
        .convert(
            self.get_keymanager_state()
                .attach_printable("Missing KeyManagerState")?,
            merchant_key_store.key.get_inner(),
            merchant_key_store.merchant_id.clone().into(),
        )
        .await
        .change_context(errors::StorageError::DecryptionError)
    }

    #[instrument(skip_all)]
    async fn list_initial_events_by_initiator_merchant_id_constraints(
        &self,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        limit: Option<i64>,
        offset: Option<i64>,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::list_initial_attempts_by_initiator_merchant_id_constraints(
            &conn,
            initiator_merchant_id,
            created_after,
            created_before,
            limit,
            offset,
            event_types,
            is_delivered,
            event_recipient,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))
        .async_and_then(|events| {
            try_join_all(events.into_iter().map(|event| async move {
                event
                    .convert(
                        self.get_keymanager_state()
                            .attach_printable("Missing KeyManagerState")?,
                        merchant_key_store.key.get_inner(),
                        merchant_key_store.merchant_id.clone().into(),
                    )
                    .await
                    .change_context(errors::StorageError::DecryptionError)
            }))
        })
        .await
    }

    #[instrument(skip_all)]
    async fn list_initial_events_by_merchant_id_constraints(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        limit: Option<i64>,
        offset: Option<i64>,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::list_initial_attempts_by_merchant_id_constraints(
            &conn,
            merchant_id,
            created_after,
            created_before,
            limit,
            offset,
            event_types,
            is_delivered,
            event_recipient,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))
        .async_and_then(|events| {
            try_join_all(events.into_iter().map(|event| async move {
                event
                    .convert(
                        self.get_keymanager_state()
                            .attach_printable("Missing KeyManagerState")?,
                        merchant_key_store.key.get_inner(),
                        merchant_key_store.merchant_id.clone().into(),
                    )
                    .await
                    .change_context(errors::StorageError::DecryptionError)
            }))
        })
        .await
    }

    #[instrument(skip_all)]
    async fn list_events_by_initiator_merchant_id_initial_attempt_id(
        &self,
        initial_attempt_id: &str,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::list_by_initiator_merchant_id_initial_attempt_id(
            &conn,
            initial_attempt_id,
            initiator_merchant_id,
            event_recipient,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))
        .async_and_then(|events| {
            try_join_all(events.into_iter().map(|event| async move {
                event
                    .convert(
                        self.get_keymanager_state()
                            .attach_printable("Missing KeyManagerState")?,
                        merchant_key_store.key.get_inner(),
                        merchant_key_store.merchant_id.clone().into(),
                    )
                    .await
                    .change_context(errors::StorageError::DecryptionError)
            }))
        })
        .await
    }

    #[instrument(skip_all)]
    async fn list_events_by_merchant_id_initial_attempt_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        initial_attempt_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::list_by_merchant_id_initial_attempt_id(
            &conn,
            merchant_id,
            initial_attempt_id,
            event_recipient,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))
        .async_and_then(|events| {
            try_join_all(events.into_iter().map(|event| async move {
                event
                    .convert(
                        self.get_keymanager_state()
                            .attach_printable("Missing KeyManagerState")?,
                        merchant_key_store.key.get_inner(),
                        merchant_key_store.merchant_id.clone().into(),
                    )
                    .await
                    .change_context(errors::StorageError::DecryptionError)
            }))
        })
        .await
    }

    #[instrument(skip_all)]
    async fn list_initial_events_by_initiator_merchant_id_primary_object_id(
        &self,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        primary_object_id: &str,
        profile_id: Option<common_utils::id_type::ProfileId>,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::list_initial_attempts_by_initiator_merchant_id_primary_object_id(
            &conn,
            initiator_merchant_id,
            primary_object_id,
            profile_id,
            event_recipient,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))
        .async_and_then(|events| {
            try_join_all(events.into_iter().map(|event| async move {
                event
                    .convert(
                        self.get_keymanager_state()
                            .attach_printable("Missing KeyManagerState")?,
                        merchant_key_store.key.get_inner(),
                        merchant_key_store.merchant_id.clone().into(),
                    )
                    .await
                    .change_context(errors::StorageError::DecryptionError)
            }))
        })
        .await
    }

    #[instrument(skip_all)]
    async fn list_initial_events_by_merchant_id_primary_object_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        primary_object_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::list_initial_attempts_by_merchant_id_primary_object_id(
            &conn,
            merchant_id,
            primary_object_id,
            event_recipient,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))
        .async_and_then(|events| {
            try_join_all(events.into_iter().map(|event| async move {
                event
                    .convert(
                        self.get_keymanager_state()
                            .attach_printable("Missing KeyManagerState")?,
                        merchant_key_store.key.get_inner(),
                        merchant_key_store.merchant_id.clone().into(),
                    )
                    .await
                    .change_context(errors::StorageError::DecryptionError)
            }))
        })
        .await
    }

    #[instrument(skip_all)]
    async fn find_initial_event_by_merchant_id_initial_attempt_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        initial_attempt_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<Option<domain::Event>, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::find_initial_attempt_by_merchant_id_initial_attempt_id(
            &conn,
            merchant_id,
            initial_attempt_id,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))
        .async_and_then(|event_opt| async {
            match event_opt {
                Some(event) => Ok(Some(
                    event
                        .convert(
                            self.get_keymanager_state()
                                .attach_printable("Missing KeyManagerState")?,
                            merchant_key_store.key.get_inner(),
                            merchant_key_store.merchant_id.clone().into(),
                        )
                        .await
                        .change_context(errors::StorageError::DecryptionError)?,
                )),
                None => Ok(None),
            }
        })
        .await
    }

    #[instrument(skip_all)]
    async fn list_initial_events_by_profile_id_constraints(
        &self,
        profile_id: &common_utils::id_type::ProfileId,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        limit: Option<i64>,
        offset: Option<i64>,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::list_initial_attempts_by_profile_id_constraints(
            &conn,
            profile_id,
            created_after,
            created_before,
            limit,
            offset,
            event_types,
            is_delivered,
            event_recipient,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))
        .async_and_then(|events| {
            try_join_all(events.into_iter().map(|event| async move {
                event
                    .convert(
                        self.get_keymanager_state()
                            .attach_printable("Missing KeyManagerState")?,
                        merchant_key_store.key.get_inner(),
                        common_utils::types::keymanager::Identifier::Merchant(
                            merchant_key_store.merchant_id.clone(),
                        ),
                    )
                    .await
                    .change_context(errors::StorageError::DecryptionError)
            }))
        })
        .await
    }

    #[instrument(skip_all)]
    async fn list_initial_events_by_profile_id_primary_object_id(
        &self,
        profile_id: &common_utils::id_type::ProfileId,
        primary_object_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::list_initial_attempts_by_profile_id_primary_object_id(
            &conn,
            profile_id,
            primary_object_id,
            event_recipient,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))
        .async_and_then(|events| {
            try_join_all(events.into_iter().map(|event| async {
                event
                    .convert(
                        self.get_keymanager_state()
                            .attach_printable("Missing KeyManagerState")?,
                        merchant_key_store.key.get_inner(),
                        merchant_key_store.merchant_id.clone().into(),
                    )
                    .await
                    .change_context(errors::StorageError::DecryptionError)
            }))
        })
        .await
    }

    #[instrument(skip_all)]
    async fn find_initial_event_by_profile_id_initial_attempt_id(
        &self,
        profile_id: &common_utils::id_type::ProfileId,
        initial_attempt_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<Option<domain::Event>, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::find_initial_attempt_by_profile_id_initial_attempt_id(
            &conn,
            profile_id,
            initial_attempt_id,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))
        .async_and_then(|event_opt| async {
            match event_opt {
                Some(event) => Ok(Some(
                    event
                        .convert(
                            self.get_keymanager_state()
                                .attach_printable("Missing KeyManagerState")?,
                            merchant_key_store.key.get_inner(),
                            common_utils::types::keymanager::Identifier::Merchant(
                                merchant_key_store.merchant_id.clone(),
                            ),
                        )
                        .await
                        .change_context(errors::StorageError::DecryptionError)?,
                )),
                None => Ok(None),
            }
        })
        .await
    }

    #[instrument(skip_all)]
    async fn update_event_by_event_id(
        &self,
        event_id: &str,
        event: domain::EventUpdate,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError> {
        let conn = connection::pg_connection_write(self).await?;
        storage::Event::update_by_event_id(&conn, event_id, event.into())
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))?
            .convert(
                self.get_keymanager_state()
                    .attach_printable("Missing KeyManagerState")?,
                merchant_key_store.key.get_inner(),
                merchant_key_store.merchant_id.clone().into(),
            )
            .await
            .change_context(errors::StorageError::DecryptionError)
    }

    #[instrument(skip_all)]
    async fn update_event_by_merchant_id_event_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        event_id: &str,
        event: domain::EventUpdate,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError> {
        let conn = connection::pg_connection_write(self).await?;
        storage::Event::update_by_merchant_id_event_id(&conn, merchant_id, event_id, event.into())
            .await
            .map_err(|error| report!(errors::StorageError::from(error)))?
            .convert(
                self.get_keymanager_state()
                    .attach_printable("Missing KeyManagerState")?,
                merchant_key_store.key.get_inner(),
                merchant_key_store.merchant_id.clone().into(),
            )
            .await
            .change_context(errors::StorageError::DecryptionError)
    }

    async fn count_initial_events_by_profile_id_constraints(
        &self,
        profile_id: &common_utils::id_type::ProfileId,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<i64, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::count_initial_attempts_by_profile_id_constraints(
            &conn,
            profile_id,
            created_after,
            created_before,
            event_types,
            is_delivered,
            event_recipient,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))
    }

    #[allow(clippy::too_many_arguments)]
    async fn count_initial_events_by_initiator_merchant_id_constraints(
        &self,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        profile_id: Option<common_utils::id_type::ProfileId>,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<i64, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::count_initial_attempts_by_initiator_merchant_id_constraints(
            &conn,
            initiator_merchant_id,
            profile_id,
            created_after,
            created_before,
            event_types,
            is_delivered,
            event_recipient,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))
    }

    async fn count_initial_events_by_constraints(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        profile_id: Option<common_utils::id_type::ProfileId>,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<i64, errors::StorageError> {
        let conn = connection::pg_connection_read(self).await?;
        storage::Event::count_initial_attempts_by_constraints(
            &conn,
            merchant_id,
            profile_id,
            created_after,
            created_before,
            event_types,
            is_delivered,
            event_recipient,
        )
        .await
        .map_err(|error| report!(errors::StorageError::from(error)))
    }
}

#[async_trait::async_trait]
impl EventInterface for MockDb {
    async fn insert_event(
        &self,
        event: domain::Event,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError> {
        let mut locked_events = self.events.lock().await;

        let stored_event = Conversion::convert(event)
            .await
            .change_context(errors::StorageError::EncryptionError)?;

        locked_events.push(stored_event.clone());

        stored_event
            .convert(
                self.get_keymanager_state()
                    .attach_printable("Missing KeyManagerState")?,
                merchant_key_store.key.get_inner(),
                merchant_key_store.merchant_id.clone().into(),
            )
            .await
            .change_context(errors::StorageError::DecryptionError)
    }

    async fn find_event_by_event_id(
        &self,
        event_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError> {
        let locked_events = self.events.lock().await;
        locked_events
            .iter()
            .find(|event| event.event_id == event_id)
            .cloned()
            .async_map(|event| async {
                event
                    .convert(
                        self.get_keymanager_state()
                            .attach_printable("Missing KeyManagerState")?,
                        merchant_key_store.key.get_inner(),
                        merchant_key_store.merchant_id.clone().into(),
                    )
                    .await
                    .change_context(errors::StorageError::DecryptionError)
            })
            .await
            .transpose()?
            .ok_or(
                errors::StorageError::ValueNotFound(format!(
                    "No event available with event_id = {event_id}"
                ))
                .into(),
            )
    }

    async fn find_event_by_merchant_id_event_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        event_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError> {
        let locked_events = self.events.lock().await;
        locked_events
            .iter()
            .find(|event| {
                event.merchant_id == Some(merchant_id.to_owned()) && event.event_id == event_id
            })
            .cloned()
            .async_map(|event| async {
                event
                    .convert(self.get_keymanager_state().attach_printable("Missing KeyManagerState")?,
                        merchant_key_store.key.get_inner(),
                        merchant_key_store.merchant_id.clone().into(),
                    )
                    .await
                    .change_context(errors::StorageError::DecryptionError)
            })
            .await
            .transpose()?
            .ok_or(
                errors::StorageError::ValueNotFound(format!(
                    "No event available with merchant_id = {merchant_id:?} and event_id  = {event_id}"
                ))
                .into(),
            )
    }

    async fn find_event_by_merchant_id_idempotent_event_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        idempotent_event_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError> {
        let locked_events = self.events.lock().await;
        locked_events
            .iter()
            .find(|event| {
                event.merchant_id == Some(merchant_id.to_owned()) && event.idempotent_event_id == Some(idempotent_event_id.to_string())
            })
            .cloned()
            .async_map(|event| async {
                event
                    .convert(self.get_keymanager_state().attach_printable("Missing KeyManagerState")?,
                        merchant_key_store.key.get_inner(),
                        merchant_key_store.merchant_id.clone().into(),
                    )
                    .await
                    .change_context(errors::StorageError::DecryptionError)
            })
            .await
            .transpose()?
            .ok_or(
                errors::StorageError::ValueNotFound(format!(
                    "No event available with merchant_id = {merchant_id:?} and idempotent_event_id  = {idempotent_event_id}"
                ))
                .into(),
            )
    }

    async fn find_event_by_initiator_merchant_id_idempotent_event_id(
        &self,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        idempotent_event_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError> {
        let locked_events = self.events.lock().await;
        locked_events
            .iter()
            .find(|event| {
                let matches_initiator = event.initiator_merchant_id.as_ref()
                    == Some(initiator_merchant_id)
                    || (event.initiator_merchant_id.is_none()
                        && event.merchant_id.as_ref() == Some(initiator_merchant_id));
                matches_initiator
                    && event.idempotent_event_id == Some(idempotent_event_id.to_string())
            })
            .cloned()
            .async_map(|event| async {
                event
                    .convert(
                        self.get_keymanager_state()
                            .attach_printable("Missing KeyManagerState")?,
                        merchant_key_store.key.get_inner(),
                        merchant_key_store.merchant_id.clone().into(),
                    )
                    .await
                    .change_context(errors::StorageError::DecryptionError)
            })
            .await
            .transpose()?
            .ok_or(
                errors::StorageError::ValueNotFound(format!(
                    "No event available with initiator_merchant_id = {initiator_merchant_id:?} and idempotent_event_id = {idempotent_event_id}"
                ))
                .into(),
            )
    }

    async fn list_initial_events_by_merchant_id_constraints(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        limit: Option<i64>,
        offset: Option<i64>,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let locked_events = self.events.lock().await;
        let events_iter = locked_events.iter().filter(|event| {
            let check = event.merchant_id == Some(merchant_id.to_owned())
                && event.initial_attempt_id.as_ref() == Some(&event.event_id)
                && (event.created_at >= created_after)
                && (event.created_at <= created_before)
                && (event_types.is_empty() || event_types.contains(&event.event_type))
                && (event.is_overall_delivery_successful == is_delivered)
                && (event_recipient.is_none() || event_recipient == event.recipient);
            check
        });

        let offset: usize = if let Some(offset) = offset {
            if offset < 0 {
                Err(errors::StorageError::MockDbError)?;
            }
            offset
                .try_into()
                .map_err(|_| errors::StorageError::MockDbError)?
        } else {
            0
        };

        let limit: usize = if let Some(limit) = limit {
            if limit < 0 {
                Err(errors::StorageError::MockDbError)?;
            }
            limit
                .try_into()
                .map_err(|_| errors::StorageError::MockDbError)?
        } else {
            usize::MAX
        };

        let events = events_iter
            .skip(offset)
            .take(limit)
            .cloned()
            .collect::<Vec<_>>();
        let mut domain_events = Vec::with_capacity(events.len());

        for event in events {
            let domain_event = event
                .convert(
                    self.get_keymanager_state()
                        .attach_printable("Missing KeyManagerState")?,
                    merchant_key_store.key.get_inner(),
                    merchant_key_store.merchant_id.clone().into(),
                )
                .await
                .change_context(errors::StorageError::DecryptionError)?;
            domain_events.push(domain_event);
        }

        Ok(domain_events)
    }

    async fn list_events_by_merchant_id_initial_attempt_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        initial_attempt_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let locked_events = self.events.lock().await;
        let events = locked_events
            .iter()
            .filter(|event| {
                event.merchant_id == Some(merchant_id.to_owned())
                    && event.initial_attempt_id == Some(initial_attempt_id.to_owned())
                    && (event_recipient.is_none() || event_recipient == event.recipient)
            })
            .cloned()
            .collect::<Vec<_>>();
        let mut domain_events = Vec::with_capacity(events.len());

        for event in events {
            let domain_event = event
                .convert(
                    self.get_keymanager_state()
                        .attach_printable("Missing KeyManagerState")?,
                    merchant_key_store.key.get_inner(),
                    merchant_key_store.merchant_id.clone().into(),
                )
                .await
                .change_context(errors::StorageError::DecryptionError)?;
            domain_events.push(domain_event);
        }

        Ok(domain_events)
    }

    async fn list_initial_events_by_merchant_id_primary_object_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        primary_object_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let locked_events = self.events.lock().await;
        let events = locked_events
            .iter()
            .filter(|event| {
                event.merchant_id == Some(merchant_id.to_owned())
                    && event.initial_attempt_id.as_deref() == Some(event.event_id.as_str())
                    && event.primary_object_id.as_str() == primary_object_id
                    && (event_recipient.is_none() || event_recipient == event.recipient)
            })
            .cloned()
            .collect::<Vec<_>>();

        let mut domain_events = Vec::with_capacity(events.len());

        for event in events {
            let domain_event = event
                .convert(
                    self.get_keymanager_state()
                        .attach_printable("Missing KeyManagerState")?,
                    merchant_key_store.key.get_inner(),
                    merchant_key_store.merchant_id.clone().into(),
                )
                .await
                .change_context(errors::StorageError::DecryptionError)?;
            domain_events.push(domain_event);
        }

        Ok(domain_events)
    }

    async fn find_initial_event_by_merchant_id_initial_attempt_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        initial_attempt_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<Option<domain::Event>, errors::StorageError> {
        let locked_events = self.events.lock().await;

        let event_opt = locked_events.iter().find(|event| {
            event.merchant_id == Some(merchant_id.to_owned())
                && event.event_id == initial_attempt_id
        });

        match event_opt {
            Some(event) => {
                let domain_event = event
                    .clone()
                    .convert(
                        self.get_keymanager_state()
                            .attach_printable("Missing KeyManagerState")?,
                        merchant_key_store.key.get_inner(),
                        merchant_key_store.merchant_id.clone().into(),
                    )
                    .await
                    .change_context(errors::StorageError::DecryptionError)?;
                Ok(Some(domain_event))
            }
            None => Ok(None),
        }
    }

    async fn list_initial_events_by_profile_id_constraints(
        &self,
        profile_id: &common_utils::id_type::ProfileId,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        limit: Option<i64>,
        offset: Option<i64>,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let locked_events = self.events.lock().await;
        let events_iter = locked_events.iter().filter(|event| {
            let check = event.business_profile_id == Some(profile_id.to_owned())
                && event.initial_attempt_id.as_ref() == Some(&event.event_id)
                && (event.created_at >= created_after)
                && (event.created_at <= created_before)
                && (event_types.is_empty() || event_types.contains(&event.event_type))
                && (event.is_overall_delivery_successful == is_delivered)
                && (event_recipient.is_none() || event_recipient == event.recipient);

            check
        });

        let offset: usize = if let Some(offset) = offset {
            if offset < 0 {
                Err(errors::StorageError::MockDbError)?;
            }
            offset
                .try_into()
                .map_err(|_| errors::StorageError::MockDbError)?
        } else {
            0
        };

        let limit: usize = if let Some(limit) = limit {
            if limit < 0 {
                Err(errors::StorageError::MockDbError)?;
            }
            limit
                .try_into()
                .map_err(|_| errors::StorageError::MockDbError)?
        } else {
            usize::MAX
        };

        let events = events_iter
            .skip(offset)
            .take(limit)
            .cloned()
            .collect::<Vec<_>>();
        let mut domain_events = Vec::with_capacity(events.len());

        for event in events {
            let domain_event = event
                .convert(
                    self.get_keymanager_state()
                        .attach_printable("Missing KeyManagerState")?,
                    merchant_key_store.key.get_inner(),
                    merchant_key_store.merchant_id.clone().into(),
                )
                .await
                .change_context(errors::StorageError::DecryptionError)?;
            domain_events.push(domain_event);
        }

        Ok(domain_events)
    }

    async fn list_initial_events_by_profile_id_primary_object_id(
        &self,
        profile_id: &common_utils::id_type::ProfileId,
        primary_object_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let locked_events = self.events.lock().await;
        let events = locked_events
            .iter()
            .filter(|event| {
                event.business_profile_id == Some(profile_id.to_owned())
                    && event.initial_attempt_id.as_deref() == Some(event.event_id.as_str())
                    && event.primary_object_id.as_str() == primary_object_id
                    && (event_recipient.is_none() || event_recipient == event.recipient)
            })
            .cloned()
            .collect::<Vec<_>>();

        let mut domain_events = Vec::with_capacity(events.len());

        for event in events {
            let domain_event = event
                .convert(
                    self.get_keymanager_state()
                        .attach_printable("Missing KeyManagerState")?,
                    merchant_key_store.key.get_inner(),
                    merchant_key_store.merchant_id.clone().into(),
                )
                .await
                .change_context(errors::StorageError::DecryptionError)?;
            domain_events.push(domain_event);
        }

        Ok(domain_events)
    }

    async fn find_initial_event_by_profile_id_initial_attempt_id(
        &self,
        profile_id: &common_utils::id_type::ProfileId,
        initial_attempt_id: &str,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<Option<domain::Event>, errors::StorageError> {
        let locked_events = self.events.lock().await;

        let event_opt = locked_events.iter().find(|event| {
            event.business_profile_id == Some(profile_id.to_owned())
                && event.event_id == initial_attempt_id
        });

        match event_opt {
            Some(event) => {
                let domain_event = event
                    .clone()
                    .convert(
                        self.get_keymanager_state()
                            .attach_printable("Missing KeyManagerState")?,
                        merchant_key_store.key.get_inner(),
                        merchant_key_store.merchant_id.clone().into(),
                    )
                    .await
                    .change_context(errors::StorageError::DecryptionError)?;
                Ok(Some(domain_event))
            }
            None => Ok(None),
        }
    }

    async fn update_event_by_merchant_id_event_id(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        event_id: &str,
        event: domain::EventUpdate,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError> {
        let mut locked_events = self.events.lock().await;
        let event_to_update = locked_events
            .iter_mut()
            .find(|event| {
                event.merchant_id == Some(merchant_id.to_owned()) && event.event_id == event_id
            })
            .ok_or(errors::StorageError::MockDbError)?;

        match event {
            domain::EventUpdate::UpdateResponse {
                is_webhook_notified,
                response,
            } => {
                event_to_update.is_webhook_notified = is_webhook_notified;
                event_to_update.response = response.map(Into::into);
            }
            domain::EventUpdate::OverallDeliveryStatusUpdate {
                is_overall_delivery_successful,
            } => {
                event_to_update.is_overall_delivery_successful =
                    Some(is_overall_delivery_successful)
            }
        }

        event_to_update
            .clone()
            .convert(
                self.get_keymanager_state()
                    .attach_printable("Missing KeyManagerState")?,
                merchant_key_store.key.get_inner(),
                merchant_key_store.merchant_id.clone().into(),
            )
            .await
            .change_context(errors::StorageError::DecryptionError)
    }

    async fn count_initial_events_by_constraints(
        &self,
        merchant_id: &common_utils::id_type::MerchantId,
        profile_id: Option<common_utils::id_type::ProfileId>,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<i64, errors::StorageError> {
        let locked_events = self.events.lock().await;

        let iter_events = locked_events.iter().filter(|event| {
            let check = event.initial_attempt_id.as_ref() == Some(&event.event_id)
                && (event.merchant_id == Some(merchant_id.to_owned()))
                && (event.business_profile_id == profile_id)
                && (event.created_at >= created_after)
                && (event.created_at <= created_before)
                && (event_types.is_empty() || event_types.contains(&event.event_type))
                && (event.is_overall_delivery_successful == is_delivered)
                && (event_recipient.is_none() || event_recipient == event.recipient);
            check
        });

        let events = iter_events.cloned().collect::<Vec<_>>();

        i64::try_from(events.len())
            .change_context(errors::StorageError::MockDbError)
            .attach_printable("Failed to convert usize to i64")
    }

    async fn list_initial_events_by_initiator_merchant_id_constraints(
        &self,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        limit: Option<i64>,
        offset: Option<i64>,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let locked_events = self.events.lock().await;
        let events_iter = locked_events.iter().filter(|event| {
            // Fallback to merchant_id for NULL initiator_merchant_id to handle
            // events created during staggered rollout by older code.
            let matches_initiator = event.initiator_merchant_id.as_ref()
                == Some(initiator_merchant_id)
                || (event.initiator_merchant_id.is_none()
                    && event.merchant_id.as_ref() == Some(initiator_merchant_id));
            let check = matches_initiator
                && event.initial_attempt_id.as_ref() == Some(&event.event_id)
                && (event.created_at >= created_after)
                && (event.created_at <= created_before)
                && (event_types.is_empty() || event_types.contains(&event.event_type))
                && (event.is_overall_delivery_successful == is_delivered)
                && (event_recipient.is_none() || event_recipient == event.recipient);

            check
        });

        let offset: usize = if let Some(offset) = offset {
            if offset < 0 {
                Err(errors::StorageError::MockDbError)?;
            }
            offset
                .try_into()
                .map_err(|_| errors::StorageError::MockDbError)?
        } else {
            0
        };

        let limit: usize = if let Some(limit) = limit {
            if limit < 0 {
                Err(errors::StorageError::MockDbError)?;
            }
            limit
                .try_into()
                .map_err(|_| errors::StorageError::MockDbError)?
        } else {
            usize::MAX
        };

        let events = events_iter
            .skip(offset)
            .take(limit)
            .cloned()
            .collect::<Vec<_>>();
        let mut domain_events = Vec::with_capacity(events.len());

        for event in events {
            let domain_event = event
                .convert(
                    self.get_keymanager_state()
                        .attach_printable("Missing KeyManagerState")?,
                    merchant_key_store.key.get_inner(),
                    merchant_key_store.merchant_id.clone().into(),
                )
                .await
                .change_context(errors::StorageError::DecryptionError)?;
            domain_events.push(domain_event);
        }

        Ok(domain_events)
    }

    async fn list_events_by_initiator_merchant_id_initial_attempt_id(
        &self,
        initial_attempt_id: &str,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let locked_events = self.events.lock().await;
        let events = locked_events
            .iter()
            .filter(|event| {
                let matches_initiator = event.initiator_merchant_id.as_ref()
                    == Some(initiator_merchant_id)
                    || (event.initiator_merchant_id.is_none()
                        && event.merchant_id.as_ref() == Some(initiator_merchant_id));
                event.initial_attempt_id.as_deref() == Some(initial_attempt_id)
                    && matches_initiator
                    && (event_recipient.is_none() || event_recipient == event.recipient)
            })
            .cloned()
            .collect::<Vec<_>>();
        let mut domain_events = Vec::with_capacity(events.len());

        for event in events {
            let domain_event = event
                .convert(
                    self.get_keymanager_state()
                        .attach_printable("Missing KeyManagerState")?,
                    merchant_key_store.key.get_inner(),
                    merchant_key_store.merchant_id.clone().into(),
                )
                .await
                .change_context(errors::StorageError::DecryptionError)?;
            domain_events.push(domain_event);
        }

        Ok(domain_events)
    }

    async fn list_initial_events_by_initiator_merchant_id_primary_object_id(
        &self,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        primary_object_id: &str,
        profile_id: Option<common_utils::id_type::ProfileId>,
        merchant_key_store: &domain::MerchantKeyStore,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<Vec<domain::Event>, errors::StorageError> {
        let locked_events = self.events.lock().await;
        let events = locked_events
            .iter()
            .filter(|event| {
                // Fallback to merchant_id for NULL initiator_merchant_id.
                let matches_initiator = event.initiator_merchant_id.as_ref()
                    == Some(initiator_merchant_id)
                    || (event.initiator_merchant_id.is_none()
                        && event.merchant_id.as_ref() == Some(initiator_merchant_id));
                let matches_profile = profile_id
                    .as_ref()
                    .is_none_or(|pid| event.business_profile_id.as_ref() == Some(pid));
                matches_initiator
                    && matches_profile
                    && event.initial_attempt_id.as_deref() == Some(event.event_id.as_str())
                    && event.primary_object_id.as_str() == primary_object_id
                    && (event_recipient.is_none() || event_recipient == event.recipient)
            })
            .cloned()
            .collect::<Vec<_>>();

        let mut domain_events = Vec::with_capacity(events.len());

        for event in events {
            let domain_event = event
                .convert(
                    self.get_keymanager_state()
                        .attach_printable("Missing KeyManagerState")?,
                    merchant_key_store.key.get_inner(),
                    merchant_key_store.merchant_id.clone().into(),
                )
                .await
                .change_context(errors::StorageError::DecryptionError)?;
            domain_events.push(domain_event);
        }

        Ok(domain_events)
    }

    async fn update_event_by_event_id(
        &self,
        event_id: &str,
        event: domain::EventUpdate,
        merchant_key_store: &domain::MerchantKeyStore,
    ) -> CustomResult<domain::Event, errors::StorageError> {
        let mut locked_events = self.events.lock().await;
        let event_to_update = locked_events
            .iter_mut()
            .find(|event| event.event_id == event_id)
            .ok_or(errors::StorageError::MockDbError)?;

        match event {
            domain::EventUpdate::UpdateResponse {
                is_webhook_notified,
                response,
            } => {
                event_to_update.is_webhook_notified = is_webhook_notified;
                event_to_update.response = response.map(Into::into);
            }
            domain::EventUpdate::OverallDeliveryStatusUpdate {
                is_overall_delivery_successful,
            } => {
                event_to_update.is_overall_delivery_successful =
                    Some(is_overall_delivery_successful)
            }
        }

        event_to_update
            .clone()
            .convert(
                self.get_keymanager_state()
                    .attach_printable("Missing KeyManagerState")?,
                merchant_key_store.key.get_inner(),
                merchant_key_store.merchant_id.clone().into(),
            )
            .await
            .change_context(errors::StorageError::DecryptionError)
    }

    async fn count_initial_events_by_profile_id_constraints(
        &self,
        profile_id: &common_utils::id_type::ProfileId,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<i64, errors::StorageError> {
        let locked_events = self.events.lock().await;

        let iter_events = locked_events.iter().filter(|event| {
            let check = event.initial_attempt_id.as_ref() == Some(&event.event_id)
                && event.business_profile_id.as_ref() == Some(profile_id)
                && (event.created_at >= created_after)
                && (event.created_at <= created_before)
                && (event_types.is_empty() || event_types.contains(&event.event_type))
                && (event.is_overall_delivery_successful == is_delivered)
                && (event_recipient.is_none() || event_recipient == event.recipient);

            check
        });

        let events = iter_events.cloned().collect::<Vec<_>>();

        i64::try_from(events.len())
            .change_context(errors::StorageError::MockDbError)
            .attach_printable("Failed to convert usize to i64")
    }

    #[allow(clippy::too_many_arguments)]
    async fn count_initial_events_by_initiator_merchant_id_constraints(
        &self,
        initiator_merchant_id: &common_utils::id_type::MerchantId,
        profile_id: Option<common_utils::id_type::ProfileId>,
        created_after: time::PrimitiveDateTime,
        created_before: time::PrimitiveDateTime,
        event_types: HashSet<common_enums::EventType>,
        is_delivered: Option<bool>,
        event_recipient: Option<common_enums::EventRecipient>,
    ) -> CustomResult<i64, errors::StorageError> {
        let locked_events = self.events.lock().await;

        let iter_events = locked_events.iter().filter(|event| {
            // Fallback to merchant_id for NULL initiator_merchant_id.
            let matches_initiator = event.initiator_merchant_id.as_ref()
                == Some(initiator_merchant_id)
                || (event.initiator_merchant_id.is_none()
                    && event.merchant_id.as_ref() == Some(initiator_merchant_id));
            let check = event.initial_attempt_id.as_ref() == Some(&event.event_id)
                && matches_initiator
                && (event.business_profile_id == profile_id)
                && (event.created_at >= created_after)
                && (event.created_at <= created_before)
                && (event_types.is_empty() || event_types.contains(&event.event_type))
                && (event.is_overall_delivery_successful == is_delivered)
                && (event_recipient.is_none() || event_recipient == event.recipient);

            check
        });

        let events = iter_events.cloned().collect::<Vec<_>>();

        i64::try_from(events.len())
            .change_context(errors::StorageError::MockDbError)
            .attach_printable("Failed to convert usize to i64")
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use api_models::webhooks as api_webhooks;
    use common_enums::IntentStatus;
    use common_utils::{
        generate_organization_id_of_default_length, type_name,
        types::{
            keymanager::{Identifier, KeyManagerState},
            MinorUnit,
        },
    };
    use diesel_models::{
        business_profile::WebhookDetails,
        enums::{self},
        events::EventMetadata,
    };
    use futures::future::join_all;
    use hyperswitch_domain_models::{
        master_key::MasterKeyInterface, merchant_account::MerchantAccountSetter,
    };
    use time::macros::datetime;
    use tokio::time::{timeout, Duration};

    use crate::{
        core::webhooks as webhooks_core,
        db::{events::EventInterface, merchant_key_store::MerchantKeyStoreInterface, MockDb},
        routes::{
            self,
            app::{settings::Settings, StorageImpl},
        },
        services,
        types::{
            api,
            domain::{self, MerchantAccount},
        },
    };

    #[tokio::test]
    #[cfg(feature = "v1")]
    #[allow(clippy::panic_in_result_fn)]
    async fn test_mockdb_event_interface() -> Result<(), Box<dyn std::error::Error>> {
        let mockdb = MockDb::new(
            &redis_interface::RedisSettings::default(),
            KeyManagerState::mock(),
        )
        .await
        .expect("Failed to create Mock store");
        let event_id = "test_event_id";
        let (tx, _) = tokio::sync::oneshot::channel();
        let app_state = Box::pin(routes::AppState::with_storage(
            Settings::new()?,
            StorageImpl::PostgresqlTest,
            tx,
            Box::new(services::MockApiClient),
            env!("CARGO_PKG_NAME"),
        ))
        .await;
        let app_state_arc = Arc::new(app_state);
        let state = app_state_arc
            .get_session_state(
                &common_utils::id_type::TenantId::try_from_string("public".to_string()).unwrap(),
                None,
                || {},
            )
            .unwrap();
        let merchant_id =
            common_utils::id_type::MerchantId::try_from(std::borrow::Cow::from("merchant_1"))
                .unwrap();
        let business_profile_id =
            common_utils::id_type::ProfileId::try_from(std::borrow::Cow::from("profile1")).unwrap();
        let payment_id = "test_payment_id";
        let key_manager_state = &(&state).into();
        let master_key = mockdb.get_master_key();
        mockdb
            .insert_merchant_key_store(
                domain::MerchantKeyStore {
                    merchant_id: merchant_id.clone(),
                    key: domain::types::crypto_operation(
                        key_manager_state,
                        type_name!(domain::MerchantKeyStore),
                        domain::types::CryptoOperation::EncryptLocally(
                            services::generate_aes256_key().unwrap().to_vec().into(),
                        ),
                        Identifier::Merchant(merchant_id.to_owned()),
                        master_key,
                    )
                    .await
                    .and_then(|val| val.try_into_operation())
                    .unwrap(),
                    created_at: datetime!(2023-02-01 0:00),
                },
                &master_key.to_vec().into(),
            )
            .await
            .unwrap();
        let merchant_key_store = mockdb
            .get_merchant_key_store_by_merchant_id(&merchant_id, &master_key.to_vec().into())
            .await
            .unwrap();

        let event1 = mockdb
            .insert_event(
                domain::Event {
                    event_id: event_id.into(),
                    event_type: enums::EventType::PaymentSucceeded,
                    event_class: enums::EventClass::Payments,
                    is_webhook_notified: false,
                    primary_object_id: payment_id.into(),
                    primary_object_type: enums::EventObjectType::PaymentDetails,
                    created_at: common_utils::date_time::now(),
                    merchant_id: Some(merchant_id.to_owned()),
                    business_profile_id: Some(business_profile_id.to_owned()),
                    primary_object_created_at: Some(common_utils::date_time::now()),
                    idempotent_event_id: Some(event_id.into()),
                    initial_attempt_id: Some(event_id.into()),
                    request: None,
                    response: None,
                    delivery_attempt: Some(enums::WebhookDeliveryAttempt::InitialAttempt),
                    metadata: Some(EventMetadata::Payment {
                        payment_id: common_utils::id_type::PaymentId::try_from(
                            std::borrow::Cow::Borrowed(payment_id),
                        )
                        .unwrap(),
                    }),
                    is_overall_delivery_successful: Some(false),
                    processor_merchant_id: Some(merchant_id.to_owned()),
                    initiator_merchant_id: Some(merchant_id.to_owned()),
                    recipient: Some(enums::EventRecipient::Merchant),
                },
                &merchant_key_store,
            )
            .await
            .unwrap();

        assert_eq!(event1.event_id, event_id);

        let updated_event = mockdb
            .update_event_by_merchant_id_event_id(
                &merchant_id,
                event_id,
                domain::EventUpdate::UpdateResponse {
                    is_webhook_notified: true,
                    response: None,
                },
                &merchant_key_store,
            )
            .await
            .unwrap();

        assert!(updated_event.is_webhook_notified);
        assert_eq!(updated_event.primary_object_id, payment_id);
        assert_eq!(updated_event.event_id, event_id);
        Ok(())
    }

    #[tokio::test]
    #[cfg(feature = "v2")]
    #[allow(clippy::panic_in_result_fn)]
    async fn test_mockdb_event_interface() -> Result<(), Box<dyn std::error::Error>> {
        let mockdb = MockDb::new(
            &redis_interface::RedisSettings::default(),
            KeyManagerState::mock(),
        )
        .await
        .expect("Failed to create Mock store");
        let event_id = "test_event_id";
        let (tx, _) = tokio::sync::oneshot::channel();
        let app_state = Box::pin(routes::AppState::with_storage(
            Settings::new()?,
            StorageImpl::PostgresqlTest,
            tx,
            Box::new(services::MockApiClient),
            env!("CARGO_PKG_NAME"),
        ))
        .await;
        let state = &Arc::new(app_state)
            .get_session_state(
                &common_utils::id_type::TenantId::try_from_string("public".to_string()).unwrap(),
                None,
                || {},
            )
            .unwrap();
        let merchant_id =
            common_utils::id_type::MerchantId::try_from(std::borrow::Cow::from("merchant_1"))
                .unwrap();
        let business_profile_id =
            common_utils::id_type::ProfileId::try_from(std::borrow::Cow::from("profile1")).unwrap();
        let payment_id = "test_payment_id";
        let key_manager_state = &state.into();
        let master_key = mockdb.get_master_key();
        mockdb
            .insert_merchant_key_store(
                domain::MerchantKeyStore {
                    merchant_id: merchant_id.clone(),
                    key: domain::types::crypto_operation(
                        key_manager_state,
                        type_name!(domain::MerchantKeyStore),
                        domain::types::CryptoOperation::EncryptLocally(
                            services::generate_aes256_key().unwrap().to_vec().into(),
                        ),
                        Identifier::Merchant(merchant_id.to_owned()),
                        master_key,
                    )
                    .await
                    .and_then(|val| val.try_into_operation())
                    .unwrap(),
                    created_at: datetime!(2023-02-01 0:00),
                },
                &master_key.to_vec().into(),
            )
            .await
            .unwrap();
        let merchant_key_store = mockdb
            .get_merchant_key_store_by_merchant_id(&merchant_id, &master_key.to_vec().into())
            .await
            .unwrap();

        let event1 = mockdb
            .insert_event(
                domain::Event {
                    event_id: event_id.into(),
                    event_type: enums::EventType::PaymentSucceeded,
                    event_class: enums::EventClass::Payments,
                    is_webhook_notified: false,
                    primary_object_id: payment_id.into(),
                    primary_object_type: enums::EventObjectType::PaymentDetails,
                    created_at: common_utils::date_time::now(),
                    merchant_id: Some(merchant_id.to_owned()),
                    business_profile_id: Some(business_profile_id.to_owned()),
                    primary_object_created_at: Some(common_utils::date_time::now()),
                    idempotent_event_id: Some(event_id.into()),
                    initial_attempt_id: Some(event_id.into()),
                    request: None,
                    response: None,
                    delivery_attempt: Some(enums::WebhookDeliveryAttempt::InitialAttempt),
                    metadata: Some(EventMetadata::Payment {
                        payment_id: common_utils::id_type::GlobalPaymentId::try_from(
                            std::borrow::Cow::Borrowed(payment_id),
                        )
                        .unwrap(),
                    }),
                    is_overall_delivery_successful: Some(false),
                    processor_merchant_id: Some(merchant_id.to_owned()),
                    initiator_merchant_id: Some(merchant_id.to_owned()),
                    recipient: Some(enums::EventRecipient::Merchant),
                },
                &merchant_key_store,
            )
            .await
            .unwrap();

        assert_eq!(event1.event_id, event_id);

        let updated_event = mockdb
            .update_event_by_merchant_id_event_id(
                &merchant_id,
                event_id,
                domain::EventUpdate::UpdateResponse {
                    is_webhook_notified: true,
                    response: None,
                },
                &merchant_key_store,
            )
            .await
            .unwrap();

        assert!(updated_event.is_webhook_notified);
        assert_eq!(updated_event.primary_object_id, payment_id);
        assert_eq!(updated_event.event_id, event_id);
        Ok(())
    }

    #[cfg(feature = "v1")]
    #[allow(clippy::panic_in_result_fn)]
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_concurrent_webhook_insertion_with_redis_lock(
    ) -> Result<(), Box<dyn std::error::Error>> {
        // Test concurrent webhook insertion with a Redis lock to prevent race conditions
        let conf = Settings::new()?;
        let tx: tokio::sync::oneshot::Sender<()> = tokio::sync::oneshot::channel().0;

        let app_state = Box::pin(routes::AppState::with_storage(
            conf,
            StorageImpl::PostgresqlTest,
            tx,
            Box::new(services::MockApiClient),
            env!("CARGO_PKG_NAME"),
        ))
        .await;

        let tenant_id = common_utils::id_type::TenantId::try_from_string("public".to_string())?;

        let state = Arc::new(app_state)
            .get_session_state(&tenant_id, None, || {})
            .map_err(|_| "failed to get session state")?;

        let merchant_id =
            common_utils::id_type::MerchantId::try_from(std::borrow::Cow::from("juspay_merchant"))?;
        let business_profile_id =
            common_utils::id_type::ProfileId::try_from(std::borrow::Cow::from("profile1"))?;
        let key_manager_state = &(&state).into();
        let master_key = state.store.get_master_key();

        let aes_key = services::generate_aes256_key()?;

        let merchant_key_store = state
            .store
            .insert_merchant_key_store(
                domain::MerchantKeyStore {
                    merchant_id: merchant_id.clone(),
                    key: domain::types::crypto_operation(
                        key_manager_state,
                        type_name!(domain::MerchantKeyStore),
                        domain::types::CryptoOperation::EncryptLocally(aes_key.to_vec().into()),
                        Identifier::Merchant(merchant_id.to_owned()),
                        master_key,
                    )
                    .await?
                    .try_into_operation()?,
                    created_at: datetime!(2023-02-01 0:00),
                },
                &master_key.to_vec().into(),
            )
            .await?;

        let merchant_account_to_insert = MerchantAccount::from(MerchantAccountSetter {
            merchant_id: merchant_id.clone(),
            merchant_name: None,
            merchant_details: None,
            return_url: None,
            webhook_details: Some(WebhookDetails {
                webhook_version: None,
                webhook_username: None,
                webhook_password: None,
                webhook_url: Some(hyperswitch_masking::Secret::new(
                    "https://example.com/webhooks".to_string(),
                )),
                payment_created_enabled: None,
                payment_succeeded_enabled: Some(true),
                payment_failed_enabled: None,
                payment_statuses_enabled: None,
                refund_statuses_enabled: None,
                payout_statuses_enabled: None,
                multiple_webhooks_list: None,
            }),
            sub_merchants_enabled: None,
            parent_merchant_id: None,
            enable_payment_response_hash: true,
            payment_response_hash_key: None,
            redirect_to_merchant_with_http_post: false,
            publishable_key: "pk_test_11DviC2G2fb3lAJoes1q3A2222233327".to_string(),
            locker_id: None,
            storage_scheme: enums::MerchantStorageScheme::PostgresOnly,
            metadata: None,
            routing_algorithm: None,
            primary_business_details: serde_json::json!({ "country": "US", "business": "default" }),
            intent_fulfillment_time: Some(1),
            created_at: common_utils::date_time::now(),
            modified_at: common_utils::date_time::now(),
            frm_routing_algorithm: None,
            payout_routing_algorithm: None,
            organization_id: generate_organization_id_of_default_length(),
            is_recon_enabled: true,
            default_profile: None,
            recon_status: enums::ReconStatus::NotRequested,
            payment_link_config: None,
            pm_collect_link_config: None,
            is_platform_account: false,
            merchant_account_type: common_enums::MerchantAccountType::Standard,
            product_type: None,
            version: common_enums::ApiVersion::V1,
            network_tokenization_credentials: None,
            fingerprint_secret: None,
        });
        let merchant_account = state
            .store
            .insert_merchant(merchant_account_to_insert, &merchant_key_store)
            .await?;

        let platform = domain::Platform::new(
            merchant_account.clone(),
            merchant_key_store.clone(),
            merchant_account.clone(),
            merchant_key_store.clone(),
            None,
        );
        let merchant_id = merchant_id.clone(); // Clone merchant_id to avoid move

        let business_profile_to_insert = domain::Profile::from(domain::ProfileSetter {
            merchant_country_code: None,
            profile_id: business_profile_id.clone(),
            merchant_id: merchant_id.clone(),
            profile_name: "test_concurrent_profile".to_string(),
            created_at: common_utils::date_time::now(),
            modified_at: common_utils::date_time::now(),
            return_url: None,
            enable_payment_response_hash: true,
            payment_response_hash_key: None,
            redirect_to_merchant_with_http_post: false,
            webhook_details: Some(WebhookDetails {
                webhook_version: None,
                webhook_username: None,
                webhook_password: None,
                webhook_url: Some(hyperswitch_masking::Secret::new(
                    "https://example.com/webhooks".to_string(),
                )),
                payment_created_enabled: None,
                payment_succeeded_enabled: Some(true),
                payment_failed_enabled: None,
                payment_statuses_enabled: None,
                refund_statuses_enabled: None,
                payout_statuses_enabled: None,
                multiple_webhooks_list: None,
            }),
            metadata: None,
            routing_algorithm: None,
            intent_fulfillment_time: None,
            frm_routing_algorithm: None,
            payout_routing_algorithm: None,
            is_recon_enabled: false,
            applepay_verified_domains: None,
            payment_link_config: None,
            session_expiry: None,
            authentication_connector_details: None,
            payout_link_config: None,
            is_extended_card_info_enabled: None,
            extended_card_info_config: None,
            is_connector_agnostic_mit_enabled: None,
            use_billing_as_payment_method_billing: None,
            collect_shipping_details_from_wallet_connector: None,
            collect_billing_details_from_wallet_connector: None,
            outgoing_webhook_custom_http_headers: None,
            always_collect_billing_details_from_wallet_connector: None,
            always_collect_shipping_details_from_wallet_connector: None,
            tax_connector_id: None,
            is_tax_connector_enabled: false,
            dynamic_routing_algorithm: None,
            is_network_tokenization_enabled: false,
            is_auto_retries_enabled: false,
            max_auto_retries_enabled: None,
            always_request_extended_authorization: None,
            is_click_to_pay_enabled: false,
            authentication_product_ids: None,
            card_testing_guard_config: None,
            card_testing_secret_key: None,
            surcharge_connector_details: None,
            is_clear_pan_retries_enabled: false,
            force_3ds_challenge: false,
            is_debit_routing_enabled: false,
            merchant_business_country: None,
            is_iframe_redirection_enabled: None,
            is_pre_network_tokenization_enabled: false,
            merchant_category_code: None,
            dispute_polling_interval: None,
            is_manual_retry_enabled: None,
            always_enable_overcapture: None,
            external_vault_details: domain::ExternalVaultDetails::Skip,
            billing_processor_id: None,
            is_l2_l3_enabled: false,
            network_tokenization_credentials: None,
            payment_method_blocking: None,
            default_fallback_routing: None,
        });

        let business_profile = state
            .store
            .insert_business_profile(&merchant_key_store.clone(), business_profile_to_insert)
            .await?;

        // Same inputs for all threads
        let event_type = enums::EventType::PaymentSucceeded;
        let event_class = enums::EventClass::Payments;
        let primary_object_id = Arc::new("concurrent_payment_id".to_string());
        let primary_object_type = enums::EventObjectType::PaymentDetails;
        let payment_id = common_utils::id_type::PaymentId::try_from(std::borrow::Cow::Borrowed(
            "pay_mbabizu24mvu3mela5njyhpit10",
        ))?;

        let primary_object_created_at = Some(common_utils::date_time::now());
        let expected_response = api::PaymentsResponse {
            payment_id,
            status: IntentStatus::Succeeded,
            amount: MinorUnit::new(6540),
            amount_capturable: MinorUnit::new(0),
            amount_received: None,
            client_secret: None,
            created: None,
            modified_at: None,
            currency: "USD".to_string(),
            customer_id: None,
            description: Some("Its my first payment request".to_string()),
            refunds: None,
            mandate_id: None,
            merchant_id: merchant_id.clone(),
            net_amount: MinorUnit::new(6540),
            processor_merchant_id: merchant_id,
            initiator: None,
            sdk_authorization: None,
            connector: None,
            customer: None,
            disputes: None,
            attempts: None,
            captures: None,
            mandate_data: None,
            setup_future_usage: None,
            off_session: None,
            capture_on: None,
            capture_method: None,
            payment_method: None,
            payment_method_data: None,
            payment_token: None,
            shipping: None,
            billing: None,
            order_details: None,
            email: None,
            name: None,
            phone: None,
            return_url: None,
            authentication_type: None,
            statement_descriptor_name: None,
            statement_descriptor_suffix: None,
            next_action: None,
            cancellation_reason: None,
            error_code: None,
            error_message: None,
            unified_code: None,
            unified_message: None,
            payment_experience: None,
            payment_method_type: None,
            connector_label: None,
            business_country: None,
            business_label: None,
            business_sub_label: None,
            allowed_payment_method_types: None,
            manual_retry_allowed: None,
            connector_transaction_id: None,
            frm_message: None,
            metadata: None,
            connector_metadata: None,
            feature_metadata: None,
            reference_id: None,
            payment_link: None,
            profile_id: None,
            surcharge_details: None,
            attempt_count: 1,
            merchant_decision: None,
            merchant_connector_id: None,
            incremental_authorization_allowed: None,
            authorization_count: None,
            incremental_authorizations: None,
            external_authentication_details: None,
            external_3ds_authentication_attempted: None,
            expires_on: None,
            fingerprint: None,
            browser_info: None,
            payment_method_id: None,
            payment_method_status: None,
            updated: None,
            split_payments: None,
            frm_metadata: None,
            merchant_order_reference_id: None,
            capture_before: None,
            extended_authorization_applied: None,
            extended_authorization_last_applied_at: None,
            order_tax_amount: None,
            connector_mandate_id: None,
            shipping_cost: None,
            card_discovery: None,
            mit_category: None,
            tokenization: None,
            force_3ds_challenge: None,
            force_3ds_challenge_trigger: None,
            issuer_error_code: None,
            issuer_error_message: None,
            is_iframe_redirection_enabled: None,
            whole_connector_response: None,
            payment_channel: None,
            network_transaction_id: None,
            network_transaction_link_id: None,
            enable_partial_authorization: None,
            is_overcapture_enabled: None,
            enable_overcapture: None,
            network_details: None,
            is_stored_credential: None,
            request_extended_authorization: None,
            billing_descriptor: None,
            partner_merchant_identifier_details: None,
            payment_method_tokenization_details: None,
            error_details: None,
            installment_options: None,
            installment_data: None,
            state_metadata: None,
            connector_response_metadata: None,
            connector_customer_id: None,
            sender_payment_instrument_id: None,
        };
        let content =
            api_webhooks::OutgoingWebhookContent::PaymentDetails(Box::new(expected_response));

        // Run 10 concurrent webhook creations
        let mut handles = vec![];
        for _ in 0..10 {
            let state_clone = state.clone();
            let cloned_platform = platform.clone();
            let business_profile_clone = business_profile.clone();
            let content_clone = content.clone();
            let primary_object_id_clone = primary_object_id.clone();

            let handle = tokio::spawn(async move {
                let webhook_recipient =
                    webhooks_core::utils::resolve_webhook_recipient_from_created_by(
                        &state_clone,
                        &cloned_platform,
                        &business_profile_clone,
                        None,
                    )
                    .await
                    .map_err(|e| format!("resolve_webhook_recipient failed: {e}"))?;
                Box::pin(webhooks_core::create_event_and_trigger_outgoing_webhook(
                    state_clone,
                    cloned_platform,
                    event_type,
                    event_class,
                    (*primary_object_id_clone).to_string(),
                    primary_object_type,
                    content_clone,
                    primary_object_created_at,
                    webhook_recipient,
                    None,
                    business_profile_clone,
                ))
                .await
                .map_err(|e| format!("create_event_and_trigger_outgoing_webhook failed: {e}"))
            });

            handles.push(handle);
        }

        // Await all tasks
        // We give the whole batch 20 s; if they don't finish something is wrong.
        let results = timeout(Duration::from_secs(20), join_all(handles))
            .await
            .map_err(|_| "tasks hung for >20 s – possible dead-lock / endless retry")?;

        for res in results {
            // Any task that panicked or returned Err will make the test fail here.
            let _ = res.map_err(|e| format!("task panicked: {e}"))?;
        }

        // Collect all initial-attempt events for this payment
        let events_by_object_id = state
            .store
            .list_initial_events_by_merchant_id_primary_object_id(
                &business_profile.merchant_id,
                primary_object_id.as_str(),
                &merchant_key_store,
                Some(enums::EventRecipient::Merchant),
            )
            .await?;

        assert_eq!(
            events_by_object_id.len(),
            1,
            "Expected exactly 1 row in events table when querying by primary_object_id, found {}",
            events_by_object_id.len()
        );

        let first_event = events_by_object_id
            .first()
            .ok_or("Expected at least one event in the result")?;
        let event_id = first_event.event_id.clone();
        let event_by_event_id = state
            .store
            .find_initial_event_by_merchant_id_initial_attempt_id(
                &business_profile.merchant_id,
                &event_id,
                &merchant_key_store,
            )
            .await?;

        assert!(
            event_by_event_id.is_some(),
            "Expected to find an event when querying by initial_attempt_id (event_id)"
        );

        let found_event = event_by_event_id.unwrap();
        assert_eq!(
            first_event.event_id, found_event.event_id,
            "Event IDs should match between queries"
        );
        Ok(())
    }
}
