Bug: [BUG] Outgoing webhook retry scheduler tasks remain in pending state if merchant webhook URL is not configured



### Bug Description

In the outgoing webhook retry process tracker (scheduler) workflow introduced in #3842, the process tracker entries added before the webhook is attempted to be delivered remain in the `process_started` status and `Pending` business status, if either:

1. The `webhook_details` field is not provided in the business profile (or merchant account with default profile).
2. The `webhook_details` is provided but the `webhook_url` within the `webhook_details` object is not provided.
3. The business profile initially had a webhoook URL configured but the profile was updated later to remove the webhook URL (or webhook details).

Since in any of the above cases the URL to deliver the webhook to the merchant is unavailable, the task can be aborted. Another optimization is to not create the outgoing webhook event (in the `events` database table) at all, if the merchant webhook URL is not available.

### Expected Behavior

- The event (and corresponding process tracker task) should not be created in the `events` table if the merchant webhook URL has not been configured.
- In case the business profile initially had a webhook URL configured and was later removed from the business profile, the retry tasks for previously failed webhooks should be aborted.

### Actual Behavior

The process tracker task remains in the `process_started` status and `Pending` business status.

### Steps To Reproduce

1. Create a merchant account without the `webhook_details` object or without the `webhook_url` field, using the API.
2. Create an API key, and create a merchant connector account.
3. Make a successful payment.
4. Query the `process_tracker` table in the database to search for workflows with the runner `OUTGOING_WEBHOOK_RETRY_WORKFLOW` and having the payment ID in the name:

   ```sql
   SELECT * FROM process_tracker WHERE runner = 'OUTGOING_WEBHOOK_RETRY_WORKFLOW' AND id LIKE '%pay_1234%';
   ```

   The `status` is `process_started` and the `business_status` of the task is `Pending`.
diff --git a/crates/router/src/core/errors.rs b/crates/router/src/core/errors.rs
index 968181061a8..609337744d1 100644
--- a/crates/router/src/core/errors.rs
+++ b/crates/router/src/core/errors.rs
@@ -259,45 +259,44 @@ pub enum WebhooksFlowError {
     #[error("Webhook details for merchant not configured")]
     MerchantWebhookDetailsNotFound,
     #[error("Merchant does not have a webhook URL configured")]
-    MerchantWebhookURLNotConfigured,
-    #[error("Payments core flow failed")]
-    PaymentsCoreFailed,
-    #[error("Refunds core flow failed")]
-    RefundsCoreFailed,
-    #[error("Dispuste core flow failed")]
-    DisputeCoreFailed,
-    #[error("Webhook event creation failed")]
-    WebhookEventCreationFailed,
+    MerchantWebhookUrlNotConfigured,
     #[error("Webhook event updation failed")]
     WebhookEventUpdationFailed,
     #[error("Outgoing webhook body signing failed")]
     OutgoingWebhookSigningFailed,
-    #[error("Unable to fork webhooks flow for outgoing webhooks")]
-    ForkFlowFailed,
     #[error("Webhook api call to merchant failed")]
     CallToMerchantFailed,
     #[error("Webhook not received by merchant")]
     NotReceivedByMerchant,
-    #[error("Resource not found")]
-    ResourceNotFound,
-    #[error("Webhook source verification failed")]
-    WebhookSourceVerificationFailed,
-    #[error("Webhook event object creation failed")]
-    WebhookEventObjectCreationFailed,
-    #[error("Not implemented")]
-    NotImplemented,
     #[error("Dispute webhook status validation failed")]
     DisputeWebhookValidationFailed,
     #[error("Outgoing webhook body encoding failed")]
     OutgoingWebhookEncodingFailed,
-    #[error("Missing required field: {field_name}")]
-    MissingRequiredField { field_name: &'static str },
     #[error("Failed to update outgoing webhook process tracker task")]
     OutgoingWebhookProcessTrackerTaskUpdateFailed,
     #[error("Failed to schedule retry attempt for outgoing webhook")]
     OutgoingWebhookRetrySchedulingFailed,
 }
 
+impl WebhooksFlowError {
+    pub(crate) fn is_webhook_delivery_retryable_error(&self) -> bool {
+        match self {
+            Self::MerchantConfigNotFound
+            | Self::MerchantWebhookDetailsNotFound
+            | Self::MerchantWebhookUrlNotConfigured => false,
+
+            Self::WebhookEventUpdationFailed
+            | Self::OutgoingWebhookSigningFailed
+            | Self::CallToMerchantFailed
+            | Self::NotReceivedByMerchant
+            | Self::DisputeWebhookValidationFailed
+            | Self::OutgoingWebhookEncodingFailed
+            | Self::OutgoingWebhookProcessTrackerTaskUpdateFailed
+            | Self::OutgoingWebhookRetrySchedulingFailed => true,
+        }
+    }
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum ApplePayDecryptionError {
     #[error("Failed to base64 decode input data")]
diff --git a/crates/router/src/core/webhooks.rs b/crates/router/src/core/webhooks.rs
index b3e04e72f33..8d03bea06cb 100644
--- a/crates/router/src/core/webhooks.rs
+++ b/crates/router/src/core/webhooks.rs
@@ -659,8 +659,21 @@ pub(crate) async fn create_event_and_trigger_outgoing_webhook(
     primary_object_type: enums::EventObjectType,
     content: api::OutgoingWebhookContent,
 ) -> CustomResult<(), errors::ApiErrorResponse> {
-    let merchant_id = business_profile.merchant_id.clone();
     let event_id = format!("{primary_object_id}_{event_type}");
+
+    if !state.conf.webhooks.outgoing_enabled
+        || get_webhook_url_from_business_profile(&business_profile).is_err()
+    {
+        logger::debug!(
+            business_profile_id=%business_profile.profile_id,
+            %event_id,
+            "Outgoing webhooks are disabled in application configuration, or merchant webhook URL \
+             could not be obtained; skipping outgoing webhooks for event"
+        );
+        return Ok(());
+    }
+
+    let merchant_id = business_profile.merchant_id.clone();
     let new_event = storage::EventNew {
         event_id: event_id.clone(),
         event_type,
@@ -688,50 +701,48 @@ pub(crate) async fn create_event_and_trigger_outgoing_webhook(
         }
     }?;
 
-    if state.conf.webhooks.outgoing_enabled {
-        let outgoing_webhook = api::OutgoingWebhook {
-            merchant_id: merchant_id.clone(),
-            event_id: event.event_id.clone(),
-            event_type: event.event_type,
-            content: content.clone(),
-            timestamp: event.created_at,
-        };
-
-        let process_tracker = add_outgoing_webhook_retry_task_to_process_tracker(
-            &*state.store,
-            &business_profile,
-            &event,
-        )
-        .await
-        .map_err(|error| {
-            logger::error!(
-                ?error,
-                "Failed to add outgoing webhook retry task to process tracker"
-            );
-            error
-        })
-        .ok();
+    let outgoing_webhook = api::OutgoingWebhook {
+        merchant_id: merchant_id.clone(),
+        event_id: event.event_id.clone(),
+        event_type: event.event_type,
+        content: content.clone(),
+        timestamp: event.created_at,
+    };
 
-        // Using a tokio spawn here and not arbiter because not all caller of this function
-        // may have an actix arbiter
-        tokio::spawn(
-            async move {
-                trigger_appropriate_webhook_and_raise_event(
-                    state,
-                    merchant_account,
-                    business_profile,
-                    outgoing_webhook,
-                    types::WebhookDeliveryAttempt::InitialAttempt,
-                    content,
-                    event.event_id,
-                    event_type,
-                    process_tracker,
-                )
-                .await;
-            }
-            .in_current_span(),
+    let process_tracker = add_outgoing_webhook_retry_task_to_process_tracker(
+        &*state.store,
+        &business_profile,
+        &event,
+    )
+    .await
+    .map_err(|error| {
+        logger::error!(
+            ?error,
+            "Failed to add outgoing webhook retry task to process tracker"
         );
-    }
+        error
+    })
+    .ok();
+
+    // Using a tokio spawn here and not arbiter because not all caller of this function
+    // may have an actix arbiter
+    tokio::spawn(
+        async move {
+            trigger_appropriate_webhook_and_raise_event(
+                state,
+                merchant_account,
+                business_profile,
+                outgoing_webhook,
+                types::WebhookDeliveryAttempt::InitialAttempt,
+                content,
+                event.event_id,
+                event_type,
+                process_tracker,
+            )
+            .await;
+        }
+        .in_current_span(),
+    );
 
     Ok(())
 }
@@ -817,21 +828,30 @@ async fn trigger_webhook_to_merchant<W: types::OutgoingWebhookType>(
     delivery_attempt: types::WebhookDeliveryAttempt,
     process_tracker: Option<storage::ProcessTracker>,
 ) -> CustomResult<(), errors::WebhooksFlowError> {
-    let webhook_details_json = business_profile
-        .webhook_details
-        .get_required_value("webhook_details")
-        .change_context(errors::WebhooksFlowError::MerchantWebhookDetailsNotFound)?;
-
-    let webhook_details: api::WebhookDetails =
-        webhook_details_json
-            .parse_value("WebhookDetails")
-            .change_context(errors::WebhooksFlowError::MerchantWebhookDetailsNotFound)?;
-
-    let webhook_url = webhook_details
-        .webhook_url
-        .get_required_value("webhook_url")
-        .change_context(errors::WebhooksFlowError::MerchantWebhookURLNotConfigured)
-        .map(ExposeInterface::expose)?;
+    let webhook_url = match (
+        get_webhook_url_from_business_profile(&business_profile),
+        process_tracker.clone(),
+    ) {
+        (Ok(webhook_url), _) => Ok(webhook_url),
+        (Err(error), Some(process_tracker)) => {
+            if !error
+                .current_context()
+                .is_webhook_delivery_retryable_error()
+            {
+                logger::debug!("Failed to obtain merchant webhook URL, aborting retries");
+                state
+                    .store
+                    .as_scheduler()
+                    .finish_process_with_business_status(process_tracker, "FAILURE".into())
+                    .await
+                    .change_context(
+                        errors::WebhooksFlowError::OutgoingWebhookProcessTrackerTaskUpdateFailed,
+                    )?;
+            }
+            Err(error)
+        }
+        (Err(error), None) => Err(error),
+    }?;
 
     let outgoing_webhook_event_id = webhook.event_id.clone();
 
@@ -1579,7 +1599,7 @@ pub async fn add_outgoing_webhook_retry_task_to_process_tracker(
     let process_tracker_id = scheduler::utils::get_process_tracker_id(
         runner,
         task,
-        &event.primary_object_id,
+        &event.event_id,
         &business_profile.merchant_id,
     );
     let process_tracker_entry = storage::ProcessTrackerNew::new(
@@ -1611,3 +1631,24 @@ pub async fn add_outgoing_webhook_retry_task_to_process_tracker(
         }
     }
 }
+
+fn get_webhook_url_from_business_profile(
+    business_profile: &diesel_models::business_profile::BusinessProfile,
+) -> CustomResult<String, errors::WebhooksFlowError> {
+    let webhook_details_json = business_profile
+        .webhook_details
+        .clone()
+        .get_required_value("webhook_details")
+        .change_context(errors::WebhooksFlowError::MerchantWebhookDetailsNotFound)?;
+
+    let webhook_details: api::WebhookDetails =
+        webhook_details_json
+            .parse_value("WebhookDetails")
+            .change_context(errors::WebhooksFlowError::MerchantWebhookDetailsNotFound)?;
+
+    webhook_details
+        .webhook_url
+        .get_required_value("webhook_url")
+        .change_context(errors::WebhooksFlowError::MerchantWebhookUrlNotConfigured)
+        .map(ExposeInterface::expose)
+}

## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [x] Bugfix
- [ ] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
This PR fixes a bug in the outgoing webhook retry workflow where process tracker tasks remained in pending state if merchant webhook URL was not configured in the business profile. For more information, see: #3995.

In addition, this PR includes these changes:

1. Updates the process tracker task ID format to include the event ID instead of the primary resource ID.
2. Removes unused enum variants from the `WebhooksFlowError` error enum.

The PR can easily be reviewed one commit at a time.

## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->
Fixes #3995.

## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->

Locally, by running `router`, `producer` and `consumer`, and testing for these three cases:

1. For a business profile with webhook URL configured, the behavior should remain as before: webhooks should either be delivered successfully in the first attempt, or scheduled for later retries.
   1. Screenshot of a webhook delivered in the initial attempt (business status is `INITIAL_DELIVERY_ATTEMPT_SUCCESSFUL`):

      ![Screenshot of webhook delivery attempt successful in the first try](https://github.com/juspay/hyperswitch/assets/22217505/04426f89-4718-476c-970f-a78cfcc33422)

   2. Screenshot of a webhook delivered during a retry attempt (business status is `COMPLETED_BY_PT` and retry count is positive):

      ![Screenshot of webhook delivery attempt successful during retry](https://github.com/juspay/hyperswitch/assets/22217505/3ecbfe60-140a-4a01-a5ef-eb6ce6bd7ab0)

2. For a business profile without `webhook_url` (or `webhook_details`) configured, the event is not created in the `events` table (and no process tracker task), and the `router` service displays a log line about the same:

   ![Screenshot of log message indicating failure in retrieving merchant webhook URL](https://github.com/juspay/hyperswitch/assets/22217505/d9eca15e-5731-4fde-a418-cbfd7148d39d)

   The absence of the event in the `events` table and the process tracker task in the `process_tracker` table can be confirmed using the SQL queries, no rows should be returned in either case:

   ```sql
   SELECT * FROM events WHERE event_id = 'event_id';
   SELECT * FROM process_tracker WHERE runner = 'OUTGOING_WEBHOOK_RETRY_WORKFLOW' AND id LIKE '%event_id%';
   ```

3. For a business profile with a webhook URL configured initially but later updated to not include the `webhook_url` field  or the `webhook_details` object, the task is aborted by the consumer:

   1. The business status is `FAILURE`, and retry count is positive:

      ![Screenshot of task aborted during a retry attempt](https://github.com/juspay/hyperswitch/assets/22217505/2b968190-397e-4679-b98e-227a59d0ff23)

   2. In addition, the `consumer` service displays a log line about the task being aborted.

      ![Screenshot of log message indicating task being aborted](https://github.com/juspay/hyperswitch/assets/22217505/0cec108a-d37f-4b2c-9a36-b2425527f58c)

The process tracker tasks in the above cases can be queried by using the SQL command, substituting the resource ID: 

```sql
SELECT * FROM process_tracker WHERE runner = 'OUTGOING_WEBHOOK_RETRY_WORKFLOW' AND id LIKE '%pay_1234%';
```


## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
- [ ] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable
