refactor(webhook): event filtering to support separate query paths for event_id and object_id filters

crates/api_models/src/webhook_events.rs

@@ -56,6 +56,8 @@ pub enum EventListConstraintsInternal {
     },
     ObjectIdFilter {
         object_id: String,
+    },
+    EventIdFilter {
         event_id: String,
     },
 }

crates/diesel_models/src/query/events.rs

@@ -46,30 +46,52 @@ impl Event {
         .await
     }
 
-    pub async fn list_initial_attempts_by_merchant_id_primary_object_id_or_initial_attempt_id(
+    pub async fn list_initial_attempts_by_merchant_id_primary_object_id(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
         primary_object_id: &str,
-        initial_attempt_id: &str,
     ) -> StorageResult<Vec<Self>> {
         generics::generic_filter::<<Self as HasTable>::Table, _, _, _>(
             conn,
             dsl::event_id
                 .nullable()
                 .eq(dsl::initial_attempt_id) // Filter initial attempts only
                 .and(dsl::merchant_id.eq(merchant_id.to_owned()))
-                .and(
-                    dsl::primary_object_id
-                        .eq(primary_object_id.to_owned())
-                        .or(dsl::initial_attempt_id.eq(initial_attempt_id.to_owned())),
-                ),
+                .and(dsl::primary_object_id.eq(primary_object_id.to_owned())),
             None,
             None,
             Some(dsl::created_at.desc()),
         )
         .await
     }
 
+    pub async fn find_initial_attempt_by_merchant_id_initial_attempt_id(
+        conn: &PgPooledConn,
+        merchant_id: &common_utils::id_type::MerchantId,
+        initial_attempt_id: &str,
+    ) -> StorageResult<Option<Self>> {
+        use diesel::{BoolExpressionMethods, ExpressionMethods};
+        use error_stack::ResultExt;
+
+        use super::generics;
+        use crate::errors::DatabaseError;
+
+        let predicate = dsl::merchant_id
+            .eq(merchant_id.to_owned())
+            .and(dsl::event_id.eq(initial_attempt_id.to_owned()));
+
+        let result =
+            generics::generic_find_one::<<Self as HasTable>::Table, _, _>(conn, predicate).await;
+
+        match result {
+            Ok(event) => Ok(Some(event)),
+            Err(err) => match err.current_context() {
+                DatabaseError::NotFound => Ok(None),
+                _ => Err(err).attach_printable("Error finding event by initial_attempt_id"),
+            },
+        }
+    }
+
     #[allow(clippy::too_many_arguments)]
     pub async fn list_initial_attempts_by_merchant_id_constraints(
         conn: &PgPooledConn,
@@ -134,30 +156,52 @@ impl Event {
         .await
     }
 
-    pub async fn list_initial_attempts_by_profile_id_primary_object_id_or_initial_attempt_id(
+    pub async fn list_initial_attempts_by_profile_id_primary_object_id(
         conn: &PgPooledConn,
         profile_id: &common_utils::id_type::ProfileId,
         primary_object_id: &str,
-        initial_attempt_id: &str,
     ) -> StorageResult<Vec<Self>> {
         generics::generic_filter::<<Self as HasTable>::Table, _, _, _>(
             conn,
             dsl::event_id
                 .nullable()
                 .eq(dsl::initial_attempt_id) // Filter initial attempts only
                 .and(dsl::business_profile_id.eq(profile_id.to_owned()))
-                .and(
-                    dsl::primary_object_id
-                        .eq(primary_object_id.to_owned())
-                        .or(dsl::initial_attempt_id.eq(initial_attempt_id.to_owned())),
-                ),
+                .and(dsl::primary_object_id.eq(primary_object_id.to_owned())),
             None,
             None,
             Some(dsl::created_at.desc()),
         )
         .await
     }
 
+    pub async fn find_initial_attempt_by_profile_id_initial_attempt_id(
+        conn: &PgPooledConn,
+        profile_id: &common_utils::id_type::ProfileId,
+        initial_attempt_id: &str,
+    ) -> StorageResult<Option<Self>> {
+        use diesel::{BoolExpressionMethods, ExpressionMethods};
+        use error_stack::ResultExt;
+
+        use super::generics;
+        use crate::errors::DatabaseError;
+
+        let predicate = dsl::business_profile_id
+            .eq(profile_id.to_owned())
+            .and(dsl::event_id.eq(initial_attempt_id.to_owned()));
+
+        let result =
+            generics::generic_find_one::<<Self as HasTable>::Table, _, _>(conn, predicate).await;
+
+        match result {
+            Ok(event) => Ok(Some(event)),
+            Err(err) => match err.current_context() {
+                DatabaseError::NotFound => Ok(None),
+                _ => Err(err).attach_printable("Error finding event by initial_attempt_id"),
+            },
+        }
+    }
+
     #[allow(clippy::too_many_arguments)]
     pub async fn list_initial_attempts_by_profile_id_constraints(
         conn: &PgPooledConn,

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

@@ -42,39 +42,65 @@ pub async fn list_initial_delivery_attempts(
         (now.date() - time::Duration::days(INITIAL_DELIVERY_ATTEMPTS_LIST_MAX_DAYS)).midnight();
 
     let (events, total_count) = match constraints {
-        api_models::webhook_events::EventListConstraintsInternal::ObjectIdFilter {
-            object_id,
-            event_id,
-        } => {
-            let events =
-                match account {
-                    MerchantAccountOrProfile::MerchantAccount(merchant_account) => store
-                        .list_initial_events_by_merchant_id_primary_object_or_initial_attempt_id(
+        api_models::webhook_events::EventListConstraintsInternal::ObjectIdFilter { object_id } => {
+            let events = match account {
+                MerchantAccountOrProfile::MerchantAccount(merchant_account) => {
+                    store
+                        .list_initial_events_by_merchant_id_primary_object_id(
                             merchant_account.get_id(),
-                            &object_id,
-                            &event_id,
+                            object_id.as_str(),
                             &key_store,
                         )
-                        .await,
-                    MerchantAccountOrProfile::Profile(business_profile) => {
-                        store
-                            .list_initial_events_by_profile_id_primary_object_or_initial_attempt_id(
-                                business_profile.get_id(),
-                                &object_id,
-                                &event_id,
-                                &key_store,
-                            )
-                            .await
-                    }
+                        .await
                 }
-                .change_context(errors::ApiErrorResponse::InternalServerError)
-                .attach_printable("Failed to list events with specified constraints")?;
+                MerchantAccountOrProfile::Profile(business_profile) => {
+                    store
+                        .list_initial_events_by_profile_id_primary_object_id(
+                            business_profile.get_id(),
+                            object_id.as_str(),
+                            &key_store,
+                        )
+                        .await
+                }
+            }
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed to list events with specified constraints")?;
 
             let total_count = i64::try_from(events.len())
                 .change_context(errors::ApiErrorResponse::InternalServerError)
                 .attach_printable("Error while converting from usize to i64")?;
             (events, total_count)
         }
+        api_models::webhook_events::EventListConstraintsInternal::EventIdFilter { event_id } => {
+            let event_opt = match account {
+                MerchantAccountOrProfile::MerchantAccount(merchant_account) => {
+                    store
+                        .find_initial_event_by_merchant_id_initial_attempt_id(
+                            merchant_account.get_id(),
+                            event_id.as_str(),
+                            &key_store,
+                        )
+                        .await
+                }
+                MerchantAccountOrProfile::Profile(business_profile) => {
+                    store
+                        .find_initial_event_by_profile_id_initial_attempt_id(
+                            business_profile.get_id(),
+                            event_id.as_str(),
+                            &key_store,
+                        )
+                        .await
+                }
+            }
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed to find event with specified event_id")?;
+
+            let (events, total_count) = match event_opt {
+                Some(event) => (vec![event], 1),
+                None => (vec![], 0),
+            };
+            (events, total_count)
+        }
         api_models::webhook_events::EventListConstraintsInternal::GenericFilter {
             created_after,
             created_before,