fix(core): add fix for stopping multiple event locking idempotent logs

Author: Debarati Ghatak

crates/diesel_models/src/query/events.rs

@@ -32,6 +32,20 @@ impl Event {
         .await
     }
 
+    pub async fn find_by_merchant_id_idempotent_event_id(
+        conn: &PgPooledConn,
+        merchant_id: &common_utils::id_type::MerchantId,
+        idempotent_event_id: &str,
+    ) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::merchant_id
+                .eq(merchant_id.to_owned())
+                .and(dsl::idempotent_event_id.eq(idempotent_event_id.to_owned())),
+        )
+        .await
+    }
+
     pub async fn list_initial_attempts_by_merchant_id_primary_object_id(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,

crates/router/src/core/webhooks/outgoing.rs

@@ -157,10 +157,10 @@ pub(crate) async fn create_event_and_trigger_outgoing_webhook(
 
     if (state
         .store
-        .find_event_by_merchant_id_event_id(
+        .find_event_by_merchant_id_idempotent_event_id(
             key_manager_state,
             &merchant_id,
-            &event_id,
+            &idempotent_event_id,
             merchant_context.get_merchant_key_store(),
         )
         .await)

crates/router/src/db/events.rs

@@ -38,6 +38,14 @@ where
         merchant_key_store: &domain::MerchantKeyStore,
     ) -> CustomResult<domain::Event, errors::StorageError>;
 
+    async fn find_event_by_merchant_id_idempotent_event_id(
+        &self,
+        state: &KeyManagerState,
+        merchant_id: &common_utils::id_type::MerchantId,
+        idempotent_event_id: &str,
+        merchant_key_store: &domain::MerchantKeyStore,
+    ) -> CustomResult<domain::Event, errors::StorageError>;
+
     async fn list_initial_events_by_merchant_id_primary_object_id(
         &self,
         state: &KeyManagerState,
@@ -157,6 +165,31 @@ impl EventInterface for Store {
             .change_context(errors::StorageError::DecryptionError)
     }
 
+    #[instrument(skip_all)]
+    async fn find_event_by_merchant_id_idempotent_event_id(
+        &self,
+        state: &KeyManagerState,
+        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(
+            state,
+            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_merchant_id_primary_object_id(
         &self,
@@ -460,6 +493,40 @@ impl EventInterface for MockDb {
             )
     }
 
+    async fn find_event_by_merchant_id_idempotent_event_id(
+        &self,
+        state: &KeyManagerState,
+        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(
+                        state,
+                        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 list_initial_events_by_merchant_id_primary_object_id(
         &self,
         state: &KeyManagerState,

crates/router/src/db/kafka_store.rs

@@ -749,6 +749,23 @@ impl EventInterface for KafkaStore {
             .await
     }
 
+    async fn find_event_by_merchant_id_idempotent_event_id(
+        &self,
+        state: &KeyManagerState,
+        merchant_id: &id_type::MerchantId,
+        idempotent_event_id: &str,
+        merchant_key_store: &domain::MerchantKeyStore,
+    ) -> CustomResult<domain::Event, errors::StorageError> {
+        self.diesel_store
+            .find_event_by_merchant_id_idempotent_event_id(
+                state,
+                merchant_id,
+                idempotent_event_id,
+                merchant_key_store,
+            )
+            .await
+    }
+
     async fn list_initial_events_by_merchant_id_primary_object_id(
         &self,
         state: &KeyManagerState,