feat(events): allow listing webhook events and webhook delivery attempts by business profile (#4159)

crates/api_models/src/webhook_events.rs

@@ -132,14 +132,28 @@ pub struct OutgoingWebhookResponseContent {
 
 #[derive(Debug, serde::Serialize)]
 pub struct EventListRequestInternal {
-    pub merchant_id: String,
+    pub merchant_id_or_profile_id: String,
     pub constraints: EventListConstraints,
 }
 
 impl common_utils::events::ApiEventMetric for EventListRequestInternal {
     fn get_api_event_type(&self) -> Option<common_utils::events::ApiEventsType> {
         Some(common_utils::events::ApiEventsType::Events {
-            merchant_id: self.merchant_id.clone(),
+            merchant_id_or_profile_id: self.merchant_id_or_profile_id.clone(),
+        })
+    }
+}
+
+#[derive(Debug, serde::Serialize)]
+pub struct WebhookDeliveryAttemptListRequestInternal {
+    pub merchant_id_or_profile_id: String,
+    pub initial_attempt_id: String,
+}
+
+impl common_utils::events::ApiEventMetric for WebhookDeliveryAttemptListRequestInternal {
+    fn get_api_event_type(&self) -> Option<common_utils::events::ApiEventsType> {
+        Some(common_utils::events::ApiEventsType::Events {
+            merchant_id_or_profile_id: self.merchant_id_or_profile_id.clone(),
         })
     }
 }

crates/common_utils/src/events.rs

@@ -54,7 +54,7 @@ pub enum ApiEventsType {
         dispute_id: String,
     },
     Events {
-        merchant_id: String,
+        merchant_id_or_profile_id: String,
     },
 }
 

crates/openapi/src/routes/webhook_events.rs

@@ -1,14 +1,14 @@
 /// Events - List
 ///
-/// List all Events associated with a Merchant Account.
+/// List all Events associated with a Merchant Account or Business Profile.
 #[utoipa::path(
     get,
-    path = "/events/{merchant_id}",
+    path = "/events/{merchant_id_or_profile_id}",
     params(
         (
-            "merchant_id" = String,
+            "merchant_id_or_profile_id" = String,
             Path,
-            description = "The unique identifier for the Merchant Account"
+            description = "The unique identifier for the Merchant Account or Business Profile"
         ),
         (
             "created_after" = Option<PrimitiveDateTime>,
@@ -45,7 +45,7 @@
         (status = 200, description = "List of Events retrieved successfully", body = Vec<EventListItemResponse>),
     ),
     tag = "Event",
-    operation_id = "List all Events associated with a Merchant Account",
+    operation_id = "List all Events associated with a Merchant Account or Business Profile",
     security(("admin_api_key" = []))
 )]
 pub fn list_initial_webhook_delivery_attempts() {}
@@ -55,9 +55,9 @@ pub fn list_initial_webhook_delivery_attempts() {}
 /// List all delivery attempts for the specified Event.
 #[utoipa::path(
     get,
-    path = "/events/{merchant_id}/{event_id}/attempts",
+    path = "/events/{merchant_id_or_profile_id}/{event_id}/attempts",
     params(
-        ("merchant_id" = String, Path, description = "The unique identifier for the Merchant Account"),
+        ("merchant_id_or_profile_id" = String, Path, description = "The unique identifier for the Merchant Account or Business Profile"),
         ("event_id" = String, Path, description = "The unique identifier for the Event"),
     ),
     responses(

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

@@ -5,15 +5,21 @@ use crate::{
     core::errors::{self, RouterResponse, StorageErrorExt},
     routes::AppState,
     services::ApplicationResponse,
-    types::{api, transformers::ForeignTryFrom},
+    types::{api, domain, transformers::ForeignTryFrom},
 };
 
 const INITIAL_DELIVERY_ATTEMPTS_LIST_MAX_LIMIT: i64 = 100;
 
+#[derive(Debug)]
+enum MerchantIdOrProfileId {
+    MerchantId(String),
+    ProfileId(String),
+}
+
 #[instrument(skip(state))]
 pub async fn list_initial_delivery_attempts(
     state: AppState,
-    merchant_id: String,
+    merchant_id_or_profile_id: String,
     constraints: api::webhook_events::EventListConstraints,
 ) -> RouterResponse<Vec<api::webhook_events::EventListItemResponse>> {
     let constraints =
@@ -21,24 +27,27 @@ pub async fn list_initial_delivery_attempts(
 
     let store = state.store.as_ref();
 
-    // This would handle verifying that the merchant ID actually exists
+    let (identifier, key_store) =
-    let key_store = store
+        determine_identifier_and_get_key_store(state.clone(), merchant_id_or_profile_id).await?;
-        .get_merchant_key_store_by_merchant_id(
-            &merchant_id,
-            &store.get_master_key().to_vec().into(),
-        )
-        .await
-        .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
 
     let events = match constraints {
         api_models::webhook_events::EventListConstraintsInternal::ObjectIdFilter { object_id } => {
-            store
+            match identifier {
+                MerchantIdOrProfileId::MerchantId(merchant_id) => store
                 .list_initial_events_by_merchant_id_primary_object_id(
                     &merchant_id,
                     &object_id,
                     &key_store,
                 )
-                .await
+                .await,
+                MerchantIdOrProfileId::ProfileId(profile_id) => store
+                .list_initial_events_by_profile_id_primary_object_id(
+                    &profile_id,
+                    &object_id,
+                    &key_store,
+                )
+                .await,
+            }
         }
         api_models::webhook_events::EventListConstraintsInternal::GenericFilter {
             created_after,
@@ -60,7 +69,8 @@ pub async fn list_initial_delivery_attempts(
                 _ => None,
             };
 
-            store
+            match identifier {
+                MerchantIdOrProfileId::MerchantId(merchant_id) => store
                 .list_initial_events_by_merchant_id_constraints(
                     &merchant_id,
                     created_after,
@@ -69,7 +79,18 @@ pub async fn list_initial_delivery_attempts(
                     offset,
                     &key_store,
                 )
-                .await
+                .await,
+                MerchantIdOrProfileId::ProfileId(profile_id) => store
+                .list_initial_events_by_profile_id_constraints(
+                    &profile_id,
+                    created_after,
+                    created_before,
+                    limit,
+                    offset,
+                    &key_store,
+                )
+                .await,
+            }
         }
     }
     .change_context(errors::ApiErrorResponse::InternalServerError)
@@ -86,22 +107,36 @@ pub async fn list_initial_delivery_attempts(
 #[instrument(skip(state))]
 pub async fn list_delivery_attempts(
     state: AppState,
-    merchant_id: &str,
+    merchant_id_or_profile_id: String,
-    initial_attempt_id: &str,
+    initial_attempt_id: String,
 ) -> RouterResponse<Vec<api::webhook_events::EventRetrieveResponse>> {
     let store = state.store.as_ref();
 
-    // This would handle verifying that the merchant ID actually exists
+    let (identifier, key_store) =
-    let key_store = store
+        determine_identifier_and_get_key_store(state.clone(), merchant_id_or_profile_id).await?;
-        .get_merchant_key_store_by_merchant_id(merchant_id, &store.get_master_key().to_vec().into())
-        .await
-        .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
 
-    let events = store
+    let events = match identifier {
-        .list_events_by_merchant_id_initial_attempt_id(merchant_id, initial_attempt_id, &key_store)
+        MerchantIdOrProfileId::MerchantId(merchant_id) => {
-        .await
+            store
-        .change_context(errors::ApiErrorResponse::InternalServerError)
+                .list_events_by_merchant_id_initial_attempt_id(
-        .attach_printable("Failed to list delivery attempts for initial event")?;
+                    &merchant_id,
+                    &initial_attempt_id,
+                    &key_store,
+                )
+                .await
+        }
+        MerchantIdOrProfileId::ProfileId(profile_id) => {
+            store
+                .list_events_by_profile_id_initial_attempt_id(
+                    &profile_id,
+                    &initial_attempt_id,
+                    &key_store,
+                )
+                .await
+        }
+    }
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .attach_printable("Failed to list delivery attempts for initial event")?;
 
     if events.is_empty() {
         Err(error_stack::report!(
@@ -117,3 +152,56 @@ pub async fn list_delivery_attempts(
         ))
     }
 }
+
+async fn determine_identifier_and_get_key_store(
+    state: AppState,
+    merchant_id_or_profile_id: String,
+) -> errors::RouterResult<(MerchantIdOrProfileId, domain::MerchantKeyStore)> {
+    let store = state.store.as_ref();
+    match store
+        .get_merchant_key_store_by_merchant_id(
+            &merchant_id_or_profile_id,
+            &store.get_master_key().to_vec().into(),
+        )
+        .await
+    {
+        // Valid merchant ID
+        Ok(key_store) => Ok((
+            MerchantIdOrProfileId::MerchantId(merchant_id_or_profile_id),
+            key_store,
+        )),
+
+        // Invalid merchant ID, check if we can find a business profile with the identifier
+        Err(error) if error.current_context().is_db_not_found() => {
+            router_env::logger::debug!(
+                ?error,
+                %merchant_id_or_profile_id,
+                "Failed to find merchant key store for the specified merchant ID or business profile ID"
+            );
+
+            let business_profile = store
+                .find_business_profile_by_profile_id(&merchant_id_or_profile_id)
+                .await
+                .to_not_found_response(errors::ApiErrorResponse::BusinessProfileNotFound {
+                    id: merchant_id_or_profile_id,
+                })?;
+
+            let key_store = store
+                .get_merchant_key_store_by_merchant_id(
+                    &business_profile.merchant_id,
+                    &store.get_master_key().to_vec().into(),
+                )
+                .await
+                .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
+
+            Ok((
+                MerchantIdOrProfileId::ProfileId(business_profile.profile_id),
+                key_store,
+            ))
+        }
+
+        Err(error) => Err(error)
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed to find merchant key store by merchant ID"),
+    }
+}

crates/router/src/routes/app.rs

@@ -1224,7 +1224,7 @@ pub struct WebhookEvents;
 #[cfg(feature = "olap")]
 impl WebhookEvents {
     pub fn server(config: AppState) -> Scope {
-        web::scope("/events/{merchant_id}")
+        web::scope("/events/{merchant_id_or_profile_id}")
             .app_data(web::Data::new(config))
             .service(web::resource("").route(web::get().to(list_initial_webhook_delivery_attempts)))
             .service(

crates/router/src/routes/webhook_events.rs

@@ -5,7 +5,9 @@ use crate::{
     core::{api_locking, webhooks::webhook_events},
     routes::AppState,
     services::{api, authentication as auth, authorization::permissions::Permission},
-    types::api::webhook_events::{EventListConstraints, EventListRequestInternal},
+    types::api::webhook_events::{
+        EventListConstraints, EventListRequestInternal, WebhookDeliveryAttemptListRequestInternal,
+    },
 };
 
 #[instrument(skip_all, fields(flow = ?Flow::WebhookEventInitialDeliveryAttemptList))]
@@ -16,11 +18,11 @@ pub async fn list_initial_webhook_delivery_attempts(
     query: web::Query<EventListConstraints>,
 ) -> impl Responder {
     let flow = Flow::WebhookEventInitialDeliveryAttemptList;
-    let merchant_id = path.into_inner();
+    let merchant_id_or_profile_id = path.into_inner();
     let constraints = query.into_inner();
 
     let request_internal = EventListRequestInternal {
-        merchant_id: merchant_id.clone(),
+        merchant_id_or_profile_id: merchant_id_or_profile_id.clone(),
         constraints,
     };
 
@@ -32,14 +34,14 @@ pub async fn list_initial_webhook_delivery_attempts(
         |state, _, request_internal| {
             webhook_events::list_initial_delivery_attempts(
                 state,
-                request_internal.merchant_id,
+                request_internal.merchant_id_or_profile_id,
                 request_internal.constraints,
             )
         },
         auth::auth_type(
             &auth::AdminApiAuth,
-            &auth::JWTAuthMerchantFromRoute {
+            &auth::JWTAuthMerchantOrProfileFromRoute {
-                merchant_id,
+                merchant_id_or_profile_id,
                 required_permission: Permission::WebhookEventRead,
             },
             req.headers(),
@@ -56,20 +58,29 @@ pub async fn list_webhook_delivery_attempts(
     path: web::Path<(String, String)>,
 ) -> impl Responder {
     let flow = Flow::WebhookEventDeliveryAttemptList;
-    let (merchant_id, initial_event_id) = path.into_inner();
+    let (merchant_id_or_profile_id, initial_attempt_id) = path.into_inner();
+
+    let request_internal = WebhookDeliveryAttemptListRequestInternal {
+        merchant_id_or_profile_id: merchant_id_or_profile_id.clone(),
+        initial_attempt_id,
+    };
 
     api::server_wrap(
         flow,
         state,
         &req,
-        (&merchant_id, &initial_event_id),
+        request_internal,
-        |state, _, (merchant_id, initial_event_id)| {
+        |state, _, request_internal| {
-            webhook_events::list_delivery_attempts(state, merchant_id, initial_event_id)
+            webhook_events::list_delivery_attempts(
+                state,
+                request_internal.merchant_id_or_profile_id,
+                request_internal.initial_attempt_id,
+            )
         },
         auth::auth_type(
             &auth::AdminApiAuth,
-            &auth::JWTAuthMerchantFromRoute {
+            &auth::JWTAuthMerchantOrProfileFromRoute {
-                merchant_id: merchant_id.clone(),
+                merchant_id_or_profile_id,
                 required_permission: Permission::WebhookEventRead,
             },
             req.headers(),

crates/router/src/services/authentication.rs

@@ -531,6 +531,68 @@ where
     }
 }
 
+pub struct JWTAuthMerchantOrProfileFromRoute {
+    pub merchant_id_or_profile_id: String,
+    pub required_permission: Permission,
+}
+
+#[async_trait]
+impl<A> AuthenticateAndFetch<(), A> for JWTAuthMerchantOrProfileFromRoute
+where
+    A: AppStateInfo + Sync,
+{
+    async fn authenticate_and_fetch(
+        &self,
+        request_headers: &HeaderMap,
+        state: &A,
+    ) -> RouterResult<((), AuthenticationType)> {
+        let payload = parse_jwt_payload::<A, AuthToken>(request_headers, state).await?;
+        if payload.check_in_blacklist(state).await? {
+            return Err(errors::ApiErrorResponse::InvalidJwtToken.into());
+        }
+
+        let permissions = authorization::get_permissions(state, &payload).await?;
+        authorization::check_authorization(&self.required_permission, &permissions)?;
+
+        // Check if token has access to MerchantId that has been requested through path or query param
+        if payload.merchant_id == self.merchant_id_or_profile_id {
+            return Ok((
+                (),
+                AuthenticationType::MerchantJwt {
+                    merchant_id: payload.merchant_id,
+                    user_id: Some(payload.user_id),
+                },
+            ));
+        }
+
+        // Route did not contain the merchant ID in present JWT, check if it corresponds to a
+        // business profile
+        let business_profile = state
+            .store()
+            .find_business_profile_by_profile_id(&self.merchant_id_or_profile_id)
+            .await
+            // Return access forbidden if business profile not found
+            .to_not_found_response(errors::ApiErrorResponse::AccessForbidden {
+                resource: self.merchant_id_or_profile_id.clone(),
+            })
+            .attach_printable("Could not find business profile specified in route")?;
+
+        // Check if merchant (from JWT) has access to business profile that has been requested
+        // through path or query param
+        if payload.merchant_id == business_profile.merchant_id {
+            Ok((
+                (),
+                AuthenticationType::MerchantJwt {
+                    merchant_id: payload.merchant_id,
+                    user_id: Some(payload.user_id),
+                },
+            ))
+        } else {
+            Err(report!(errors::ApiErrorResponse::InvalidJwtToken))
+        }
+    }
+}
+
 pub async fn parse_jwt_payload<A, T>(headers: &HeaderMap, state: &A) -> RouterResult<T>
 where
     T: serde::de::DeserializeOwned,

crates/router/src/types/api/webhook_events.rs

@@ -1,5 +1,5 @@
 pub use api_models::webhook_events::{
     EventListConstraints, EventListConstraintsInternal, EventListItemResponse,
     EventListRequestInternal, EventRetrieveResponse, OutgoingWebhookRequestContent,
-    OutgoingWebhookResponseContent,
+    OutgoingWebhookResponseContent, WebhookDeliveryAttemptListRequestInternal,
 };

openapi/openapi_spec.json

@@ -4325,19 +4325,19 @@
         ]
       }
     },
-    "/events/{merchant_id}": {
+    "/events/{merchant_id_or_profile_id}": {
       "get": {
         "tags": [
           "Event"
         ],
         "summary": "Events - List",
-        "description": "Events - List\n\nList all Events associated with a Merchant Account.",
+        "description": "Events - List\n\nList all Events associated with a Merchant Account or Business Profile.",
-        "operationId": "List all Events associated with a Merchant Account",
+        "operationId": "List all Events associated with a Merchant Account or Business Profile",
         "parameters": [
           {
-            "name": "merchant_id",
+            "name": "merchant_id_or_profile_id",
             "in": "path",
-            "description": "The unique identifier for the Merchant Account",
+            "description": "The unique identifier for the Merchant Account or Business Profile",
             "required": true,
             "schema": {
               "type": "string"
@@ -4420,7 +4420,7 @@
         ]
       }
     },
-    "/events/{merchant_id}/{event_id}/attempts": {
+    "/events/{merchant_id_or_profile_id}/{event_id}/attempts": {
       "get": {
         "tags": [
           "Event"
@@ -4430,9 +4430,9 @@
         "operationId": "List all delivery attempts for an Event",
         "parameters": [
           {
-            "name": "merchant_id",
+            "name": "merchant_id_or_profile_id",
             "in": "path",
-            "description": "The unique identifier for the Merchant Account",
+            "description": "The unique identifier for the Merchant Account or Business Profile",
             "required": true,
             "schema": {
               "type": "string"