Issue:
Bug: [FEATURE] Schedule webhook for retry



## Description

Refactor webhooks core to add retry logic in webhooks core.


Currently, we store information about whether we have sent a webhook or not, improving on this we can implement retry logic to make this more reliable

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
This PR adds support for automatically retrying outgoing webhook deliveries in case of failed deliveries, with the aid of the scheduler (process tracker).

### Behavior (Initial Delivery)

The behavior when an outgoing webhook is being sent is:

1. During the initial webhook delivery attempt, a process tracker task is added before sending the webhook, since we launch the webhook delivery task on a background thread that we have no control over. The scheduled time for the task is determined from configuration (persisted in database/Redis), which has been explained later.

2. Next, we trigger the webhook to the merchant and an analytics event is raised, as before.

   1. In case the initial delivery attempt succeeds, the process tracker task is marked completed. The end.
   2. In case the initial delivery attempt fails, no further action is taken, the webhook delivery will be attempted again by the process tracker at the scheduled time.

### Behavior (Automatic Retry)

1. During the automatic retry attempt(s) from process tracker, the current information about the resource (about which the webhook is being sent) is fetched from the database and will be included in the webhook payload.

2. If the status of the resource is different than when the event for the webhook was created (the resource has now transitioned to another status), then the process tracker task is aborted with business status indicating the status mismatch.

3. When triggering the webhook to the merchant:

   1. If the retry attempt succeeds, the process tracker task is marked completed.
   2. If the retry attempt fails, the process tracker task is scheduled again for retry at the scheduled time determined from configuration.
   3. If the number of retries exceeds the maximum number of retries, then the process tracker task is marked completed with business status indicating that retries exceeded.

### Configuring Retry Intervals

The retry intervals for webhook deliveries are determined by runtime configuration (persisted in the database/Redis), using the key `pt_mapping_outgoing_webhooks`. In case the configuration value is not available in storage, a default configuration value is used, which is harcoded in the application:

https://github.com/juspay/hyperswitch/blob/1d3bf5cf9ae800d4eba78c2ae03824c9a623ed87/crates/scheduler/src/consumer/types/process_data.rs#L69-L89

In case the default configuration needs to be overridden, it can be done so using the configuration create (or update) endpoint:

```shell
curl --location 'http://localhost:8080/configs/' \
  --header 'Content-Type: application/json' \
  --header 'Accept: application/json' \
  --header 'api-key: <ADMIN_API_KEY>' \
  --data '{
    "key": "pt_mapping_outgoing_webhooks",
    "value": "{\"default_mapping\":{\"start_after\":60,\"frequency\":[15,30],\"count\":[2,3]},\"custom_merchant_mapping\":{}}"
  }'
```

This would override the configuration for all merchants. Note that `frequency` and `start_after` are specified in seconds. If it needs to be overridden only for a specific merchant, then the `custom_merchant_mapping` field must have a similar object as `default_mapping`, keyed by the merchant ID:

```json
{
  "default_mapping": {
    "start_after": 60,
    "frequency": [15, 30],
    "count": [2, 3]
  },
  "custom_merchant_mapping": {
    "merchant_id1": {
      "start_after": 30,
      "frequency": [300],
      "count": [2]
    }
  }
}
```

## 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).
-->
Closes #217.

## 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)?
-->

As of now, outgoing webhooks are supported for payments, refunds, disputes and mandates. I've extensively tested payments outgoing webhooks for the different cases, and done a sanity testing on disputes and refunds webhooks to verify that they are retried in case the initial delivery attempt fails. Incoming mandates webhooks are integrated for the GoCardless connector, but the integration is broken and I couldn't test mandates webhook retries.

As for simulating failed webhook deliveries, the merchant webhook URL would have to be configured to a URL which does not accept `POST` requests. After a couple of failed retries, the URL can be updated to a valid URL to try out the case where the retried delivery attempt succeeds.

Since the hardcoded default retry configuration spans multiple hours, I configured the application to use shorter intervals:

```shell
curl --location 'http://localhost:8080/configs/' \
  --header 'Content-Type: application/json' \
  --header 'Accept: application/json' \
  --header 'api-key: <ADMIN_API_KEY>' \
  --data '{
    "key": "pt_mapping_outgoing_webhooks",
    "value": "{\"default_mapping\":{\"start_after\":60,\"frequency\":[15,30],\"count\":[2,3]},\"custom_merchant_mapping\":{}}"
  }'
```

The `process_tracker` table can be queried for relevant record using this query:

```sql
SELECT * FROM process_tracker WHERE runner = 'OUTGOING_WEBHOOK_RETRY_WORKFLOW' ORDER BY schedule_time DESC;
```

1. If the initial delivery attempt is successful, the business status of the process tracker entry is set to `INITIAL_DELIVERY_ATTEMPT_SUCCESSFUL`:

   ![Screenshot of process tracker entry where initial delivery attempt was successful](https://github.com/juspay/hyperswitch/assets/22217505/f9c77bc5-4834-43dc-aa15-892d9ef2db28)

2. If one of the retried delivery attempts are successful, the business status of the process tracker entry is set to `COMPLETED_BY_PT`:

   ![Screenshot of process tracker entry where retried delivery attempt was successful](https://github.com/juspay/hyperswitch/assets/22217505/7d3b6beb-3cbf-4aa6-8141-7481df8aea54)

4. If none of the delivery attempts are successful, the business status of the process tracker entry is set to `RETRIES_EXCEEDED`:

   ![Screenshot of process tracker entry where retries failed](https://github.com/juspay/hyperswitch/assets/22217505/aad8c1bb-837a-40f4-9bbf-db3169a4751e)

   In my case, the configured `count` was `[2,3]`, so the maximum number of retries turns out to be 2 + 3 = 5.

As for testing refunds and disputes webhooks, the screenshots are attached below:

1. Refunds (note the `event_type` and `event_class` fields):

   ![Screenshot of failed refund webhook being retried](https://github.com/juspay/hyperswitch/assets/22217505/20cc14df-fa1f-49aa-911a-844bf103149b)

2. Disputes (note the `event_type` and `event_class` fields), the business status is `RESOURCE_STATUS_MISMATCH` since the dispute status changed since the time it was created to the time the webhook was being retried:

   ![Screenshot of dispute webhook being aborted due to status mismatch](https://github.com/juspay/hyperswitch/assets/22217505/965934a1-f42d-4bb4-9a4d-815e838fe6dc)

In addition, successful and failed task additions should raise appropriate metrics (`TASKS_ADDED_COUNT` and `TASK_ADDITION_FAILURES_COUNT` respectively), and suitable logs are being thrown when delivering webhooks to the merchant fails, and when retry configs are being read from the database.

The PR also adds unit tests for a scheduler utility function I refactored, which can be run using the command:

```shell
cargo test --package scheduler --lib -- utils::tests
```

![Screenshot of unit tests run](https://github.com/juspay/hyperswitch/assets/22217505/9481f100-3532-4846-97da-a935b915e609)


## 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


Testing:


As of now, outgoing webhooks are supported for payments, refunds, disputes and mandates. I've extensively tested payments outgoing webhooks for the different cases, and done a sanity testing on disputes and refunds webhooks to verify that they are retried in case the initial delivery attempt fails. Incoming mandates webhooks are integrated for the GoCardless connector, but the integration is broken and I couldn't test mandates webhook retries.

As for simulating failed webhook deliveries, the merchant webhook URL would have to be configured to a URL which does not accept `POST` requests. After a couple of failed retries, the URL can be updated to a valid URL to try out the case where the retried delivery attempt succeeds.

Since the hardcoded default retry configuration spans multiple hours, I configured the application to use shorter intervals:

```shell
curl --location 'http://localhost:8080/configs/' \
  --header 'Content-Type: application/json' \
  --header 'Accept: application/json' \
  --header 'api-key: <ADMIN_API_KEY>' \
  --data '{
    "key": "pt_mapping_outgoing_webhooks",
    "value": "{\"default_mapping\":{\"start_after\":60,\"frequency\":[15,30],\"count\":[2,3]},\"custom_merchant_mapping\":{}}"
  }'
```

The `process_tracker` table can be queried for relevant record using this query:

```sql
SELECT * FROM process_tracker WHERE runner = 'OUTGOING_WEBHOOK_RETRY_WORKFLOW' ORDER BY schedule_time DESC;
```

1. If the initial delivery attempt is successful, the business status of the process tracker entry is set to `INITIAL_DELIVERY_ATTEMPT_SUCCESSFUL`:

   ![Screenshot of process tracker entry where initial delivery attempt was successful](https://github.com/juspay/hyperswitch/assets/22217505/f9c77bc5-4834-43dc-aa15-892d9ef2db28)

2. If one of the retried delivery attempts are successful, the business status of the process tracker entry is set to `COMPLETED_BY_PT`:

   ![Screenshot of process tracker entry where retried delivery attempt was successful](https://github.com/juspay/hyperswitch/assets/22217505/7d3b6beb-3cbf-4aa6-8141-7481df8aea54)

4. If none of the delivery attempts are successful, the business status of the process tracker entry is set to `RETRIES_EXCEEDED`:

   ![Screenshot of process tracker entry where retries failed](https://github.com/juspay/hyperswitch/assets/22217505/aad8c1bb-837a-40f4-9bbf-db3169a4751e)

   In my case, the configured `count` was `[2,3]`, so the maximum number of retries turns out to be 2 + 3 = 5.

As for testing refunds and disputes webhooks, the screenshots are attached below:

1. Refunds (note the `event_type` and `event_class` fields):

   ![Screenshot of failed refund webhook being retried](https://github.com/juspay/hyperswitch/assets/22217505/20cc14df-fa1f-49aa-911a-844bf103149b)

2. Disputes (note the `event_type` and `event_class` fields), the business status is `RESOURCE_STATUS_MISMATCH` since the dispute status changed since the time it was created to the time the webhook was being retried:

   ![Screenshot of dispute webhook being aborted due to status mismatch](https://github.com/juspay/hyperswitch/assets/22217505/965934a1-f42d-4bb4-9a4d-815e838fe6dc)

In addition, successful and failed task additions should raise appropriate metrics (`TASKS_ADDED_COUNT` and `TASK_ADDITION_FAILURES_COUNT` respectively), and suitable logs are being thrown when delivering webhooks to the merchant fails, and when retry configs are being read from the database.

The PR also adds unit tests for a scheduler utility function I refactored, which can be run using the command:

```shell
cargo test --package scheduler --lib -- utils::tests
```

![Screenshot of unit tests run](https://github.com/juspay/hyperswitch/assets/22217505/9481f100-3532-4846-97da-a935b915e609)

Patch:
diff --git a/.typos.toml b/.typos.toml
index 40acb130589..2d2e165544a 100644
--- a/.typos.toml
+++ b/.typos.toml
@@ -2,6 +2,7 @@
 check-filename = true
 
 [default.extend-identifiers]
+"ABD" = "ABD"  # Aberdeenshire, UK ISO 3166-2 code
 BA = "BA"  # Bosnia and Herzegovina country code
 CAF = "CAF"  # Central African Republic country code
 flate2 = "flate2"
diff --git a/crates/api_models/src/webhooks.rs b/crates/api_models/src/webhooks.rs
index 37aaac42e0d..25e1f5827ad 100644
--- a/crates/api_models/src/webhooks.rs
+++ b/crates/api_models/src/webhooks.rs
@@ -160,9 +160,9 @@ pub struct OutgoingWebhook {
 
     /// This is specific to the flow, for ex: it will be `PaymentsResponse` for payments flow
     pub content: OutgoingWebhookContent,
-    #[serde(default, with = "custom_serde::iso8601")]
 
     /// The time at which webhook was sent
+    #[serde(default, with = "custom_serde::iso8601")]
     pub timestamp: PrimitiveDateTime,
 }
 
diff --git a/crates/diesel_models/src/process_tracker.rs b/crates/diesel_models/src/process_tracker.rs
index 76e280d6bc0..74fa75a2751 100644
--- a/crates/diesel_models/src/process_tracker.rs
+++ b/crates/diesel_models/src/process_tracker.rs
@@ -229,6 +229,7 @@ pub enum ProcessTrackerRunner {
     RefundWorkflowRouter,
     DeleteTokenizeDataWorkflow,
     ApiKeyExpiryWorkflow,
+    OutgoingWebhookRetryWorkflow,
 }
 
 #[cfg(test)]
diff --git a/crates/diesel_models/src/query/events.rs b/crates/diesel_models/src/query/events.rs
index 735bd892508..9175823e5f9 100644
--- a/crates/diesel_models/src/query/events.rs
+++ b/crates/diesel_models/src/query/events.rs
@@ -14,6 +14,14 @@ impl EventNew {
 }
 
 impl Event {
+    pub async fn find_by_event_id(conn: &PgPooledConn, event_id: &str) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::event_id.eq(event_id.to_owned()),
+        )
+        .await
+    }
+
     pub async fn update(
         conn: &PgPooledConn,
         event_id: &str,
diff --git a/crates/router/src/bin/scheduler.rs b/crates/router/src/bin/scheduler.rs
index c2877535daa..c586bfecdb7 100644
--- a/crates/router/src/bin/scheduler.rs
+++ b/crates/router/src/bin/scheduler.rs
@@ -255,6 +255,9 @@ impl ProcessTrackerWorkflows<routes::AppState> for WorkflowRunner {
                             )
                     }
                 }
+                storage::ProcessTrackerRunner::OutgoingWebhookRetryWorkflow => Ok(Box::new(
+                    workflows::outgoing_webhook_retry::OutgoingWebhookRetryWorkflow,
+                )),
             }
         };
 
diff --git a/crates/router/src/compatibility/wrap.rs b/crates/router/src/compatibility/wrap.rs
index da14475f120..da9a7f3d163 100644
--- a/crates/router/src/compatibility/wrap.rs
+++ b/crates/router/src/compatibility/wrap.rs
@@ -135,7 +135,7 @@ where
             .map_into_boxed_body()
         }
 
-        Ok(api::ApplicationResponse::PaymenkLinkForm(boxed_payment_link_data)) => {
+        Ok(api::ApplicationResponse::PaymentLinkForm(boxed_payment_link_data)) => {
             match *boxed_payment_link_data {
                 api::PaymentLinkAction::PaymentLinkFormData(payment_link_data) => {
                     match api::build_payment_link_html(payment_link_data) {
diff --git a/crates/router/src/core/errors.rs b/crates/router/src/core/errors.rs
index cec563b9631..968181061a8 100644
--- a/crates/router/src/core/errors.rs
+++ b/crates/router/src/core/errors.rs
@@ -292,6 +292,10 @@ pub enum WebhooksFlowError {
     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,
 }
 
 #[derive(Debug, thiserror::Error)]
diff --git a/crates/router/src/core/payment_link.rs b/crates/router/src/core/payment_link.rs
index 5af948bcf3e..096d5c88743 100644
--- a/crates/router/src/core/payment_link.rs
+++ b/crates/router/src/core/payment_link.rs
@@ -195,7 +195,7 @@ pub async fn intiate_payment_link_flow(
             js_script,
             css_script,
         };
-        return Ok(services::ApplicationResponse::PaymenkLinkForm(Box::new(
+        return Ok(services::ApplicationResponse::PaymentLinkForm(Box::new(
             services::api::PaymentLinkAction::PaymentLinkStatus(payment_link_error_data),
         )));
     };
@@ -225,7 +225,7 @@ pub async fn intiate_payment_link_flow(
         sdk_url: state.conf.payment_link.sdk_url.clone(),
         css_script,
     };
-    Ok(services::ApplicationResponse::PaymenkLinkForm(Box::new(
+    Ok(services::ApplicationResponse::PaymentLinkForm(Box::new(
         services::api::PaymentLinkAction::PaymentLinkFormData(payment_link_data),
     )))
 }
@@ -574,7 +574,7 @@ pub async fn get_payment_link_status(
         js_script,
         css_script,
     };
-    Ok(services::ApplicationResponse::PaymenkLinkForm(Box::new(
+    Ok(services::ApplicationResponse::PaymentLinkForm(Box::new(
         services::api::PaymentLinkAction::PaymentLinkStatus(payment_link_status_data),
     )))
 }
diff --git a/crates/router/src/core/payment_methods/vault.rs b/crates/router/src/core/payment_methods/vault.rs
index 622fed0738a..af9f94a0bdf 100644
--- a/crates/router/src/core/payment_methods/vault.rs
+++ b/crates/router/src/core/payment_methods/vault.rs
@@ -1084,8 +1084,8 @@ pub async fn get_delete_tokenize_schedule_time(
     .await;
     let mapping = match redis_mapping {
         Ok(x) => x,
-        Err(err) => {
-            logger::info!("Redis Mapping Error: {}", err);
+        Err(error) => {
+            logger::info!(?error, "Redis Mapping Error");
             process_data::PaymentMethodsPTMapping::default()
         }
     };
diff --git a/crates/router/src/core/refunds.rs b/crates/router/src/core/refunds.rs
index 06030262fbc..80a0c285e41 100644
--- a/crates/router/src/core/refunds.rs
+++ b/crates/router/src/core/refunds.rs
@@ -1167,34 +1167,3 @@ pub async fn get_refund_sync_process_schedule_time(
 
     Ok(process_tracker_utils::get_time_from_delta(time_delta))
 }
-
-pub async fn retry_refund_sync_task(
-    db: &dyn db::StorageInterface,
-    connector: String,
-    merchant_id: String,
-    pt: storage::ProcessTracker,
-) -> Result<(), errors::ProcessTrackerError> {
-    let schedule_time =
-        get_refund_sync_process_schedule_time(db, &connector, &merchant_id, pt.retry_count).await?;
-
-    match schedule_time {
-        Some(s_time) => {
-            let retry_schedule = db
-                .as_scheduler()
-                .retry_process(pt, s_time)
-                .await
-                .map_err(Into::into);
-            metrics::TASKS_RESET_COUNT.add(
-                &metrics::CONTEXT,
-                1,
-                &[metrics::request::add_attributes("flow", "Refund")],
-            );
-            retry_schedule
-        }
-        None => db
-            .as_scheduler()
-            .finish_process_with_business_status(pt, "RETRIES_EXCEEDED".to_string())
-            .await
-            .map_err(Into::into),
-    }
-}
diff --git a/crates/router/src/core/webhooks.rs b/crates/router/src/core/webhooks.rs
index 9ec8de21f94..b3e04e72f33 100644
--- a/crates/router/src/core/webhooks.rs
+++ b/crates/router/src/core/webhooks.rs
@@ -32,6 +32,7 @@ use crate::{
     events::{
         api_logs::ApiEvent,
         outgoing_webhook_logs::{OutgoingWebhookEvent, OutgoingWebhookEventMetric},
+        RawEvent,
     },
     logger,
     routes::{app::AppStateInfo, lock_utils, metrics::request::add_attributes, AppState},
@@ -43,15 +44,13 @@ use crate::{
         transformers::{ForeignInto, ForeignTryInto},
     },
     utils::{self as helper_utils, generate_id, OptionExt, ValueExt},
+    workflows::outgoing_webhook_retry,
 };
 
 const OUTGOING_WEBHOOK_TIMEOUT_SECS: u64 = 5;
 const MERCHANT_ID: &str = "merchant_id";
 
-pub async fn payments_incoming_webhook_flow<
-    W: types::OutgoingWebhookType,
-    Ctx: PaymentMethodRetrieve,
->(
+pub async fn payments_incoming_webhook_flow<Ctx: PaymentMethodRetrieve>(
     state: AppState,
     merchant_account: domain::MerchantAccount,
     business_profile: diesel_models::business_profile::BusinessProfile,
@@ -169,7 +168,7 @@ pub async fn payments_incoming_webhook_flow<
 
             // If event is NOT an UnsupportedEvent, trigger Outgoing Webhook
             if let Some(outgoing_event_type) = event_type {
-                create_event_and_trigger_outgoing_webhook::<W>(
+                create_event_and_trigger_outgoing_webhook(
                     state,
                     merchant_account,
                     business_profile,
@@ -196,7 +195,7 @@ pub async fn payments_incoming_webhook_flow<
 
 #[instrument(skip_all)]
 #[allow(clippy::too_many_arguments)]
-pub async fn refunds_incoming_webhook_flow<W: types::OutgoingWebhookType>(
+pub async fn refunds_incoming_webhook_flow(
     state: AppState,
     merchant_account: domain::MerchantAccount,
     business_profile: diesel_models::business_profile::BusinessProfile,
@@ -275,7 +274,7 @@ pub async fn refunds_incoming_webhook_flow<W: types::OutgoingWebhookType>(
     if let Some(outgoing_event_type) = event_type {
         let refund_response: api_models::refunds::RefundResponse =
             updated_refund.clone().foreign_into();
-        create_event_and_trigger_outgoing_webhook::<W>(
+        create_event_and_trigger_outgoing_webhook(
             state,
             merchant_account,
             business_profile,
@@ -414,7 +413,7 @@ pub async fn get_or_update_dispute_object(
     }
 }
 
-pub async fn mandates_incoming_webhook_flow<W: types::OutgoingWebhookType>(
+pub async fn mandates_incoming_webhook_flow(
     state: AppState,
     merchant_account: domain::MerchantAccount,
     business_profile: diesel_models::business_profile::BusinessProfile,
@@ -471,7 +470,7 @@ pub async fn mandates_incoming_webhook_flow<W: types::OutgoingWebhookType>(
         );
         let event_type: Option<enums::EventType> = updated_mandate.mandate_status.foreign_into();
         if let Some(outgoing_event_type) = event_type {
-            create_event_and_trigger_outgoing_webhook::<W>(
+            create_event_and_trigger_outgoing_webhook(
                 state,
                 merchant_account,
                 business_profile,
@@ -496,7 +495,7 @@ pub async fn mandates_incoming_webhook_flow<W: types::OutgoingWebhookType>(
 
 #[allow(clippy::too_many_arguments)]
 #[instrument(skip_all)]
-pub async fn disputes_incoming_webhook_flow<W: types::OutgoingWebhookType>(
+pub async fn disputes_incoming_webhook_flow(
     state: AppState,
     merchant_account: domain::MerchantAccount,
     business_profile: diesel_models::business_profile::BusinessProfile,
@@ -538,7 +537,7 @@ pub async fn disputes_incoming_webhook_flow<W: types::OutgoingWebhookType>(
         let disputes_response = Box::new(dispute_object.clone().foreign_into());
         let event_type: enums::EventType = dispute_object.dispute_status.foreign_into();
 
-        create_event_and_trigger_outgoing_webhook::<W>(
+        create_event_and_trigger_outgoing_webhook(
             state,
             merchant_account,
             business_profile,
@@ -562,7 +561,7 @@ pub async fn disputes_incoming_webhook_flow<W: types::OutgoingWebhookType>(
     }
 }
 
-async fn bank_transfer_webhook_flow<W: types::OutgoingWebhookType, Ctx: PaymentMethodRetrieve>(
+async fn bank_transfer_webhook_flow<Ctx: PaymentMethodRetrieve>(
     state: AppState,
     merchant_account: domain::MerchantAccount,
     business_profile: diesel_models::business_profile::BusinessProfile,
@@ -624,7 +623,7 @@ async fn bank_transfer_webhook_flow<W: types::OutgoingWebhookType, Ctx: PaymentM
 
             // If event is NOT an UnsupportedEvent, trigger Outgoing Webhook
             if let Some(outgoing_event_type) = event_type {
-                create_event_and_trigger_outgoing_webhook::<W>(
+                create_event_and_trigger_outgoing_webhook(
                     state,
                     merchant_account,
                     business_profile,
@@ -649,53 +648,7 @@ async fn bank_transfer_webhook_flow<W: types::OutgoingWebhookType, Ctx: PaymentM
 
 #[allow(clippy::too_many_arguments)]
 #[instrument(skip_all)]
-pub async fn create_event_and_trigger_appropriate_outgoing_webhook(
-    state: AppState,
-    merchant_account: domain::MerchantAccount,
-    business_profile: diesel_models::business_profile::BusinessProfile,
-    event_type: enums::EventType,
-    event_class: enums::EventClass,
-    intent_reference_id: Option<String>,
-    primary_object_id: String,
-    primary_object_type: enums::EventObjectType,
-    content: api::OutgoingWebhookContent,
-) -> CustomResult<(), errors::ApiErrorResponse> {
-    match merchant_account.get_compatible_connector() {
-        #[cfg(feature = "stripe")]
-        Some(api_models::enums::Connector::Stripe) => {
-            create_event_and_trigger_outgoing_webhook::<stripe_webhooks::StripeOutgoingWebhook>(
-                state.clone(),
-                merchant_account,
-                business_profile,
-                event_type,
-                event_class,
-                intent_reference_id,
-                primary_object_id,
-                primary_object_type,
-                content,
-            )
-            .await
-        }
-        _ => {
-            create_event_and_trigger_outgoing_webhook::<api_models::webhooks::OutgoingWebhook>(
-                state.clone(),
-                merchant_account,
-                business_profile,
-                event_type,
-                event_class,
-                intent_reference_id,
-                primary_object_id,
-                primary_object_type,
-                content,
-            )
-            .await
-        }
-    }
-}
-
-#[allow(clippy::too_many_arguments)]
-#[instrument(skip_all)]
-pub async fn create_event_and_trigger_outgoing_webhook<W: types::OutgoingWebhookType>(
+pub(crate) async fn create_event_and_trigger_outgoing_webhook(
     state: AppState,
     merchant_account: domain::MerchantAccount,
     business_profile: diesel_models::business_profile::BusinessProfile,
@@ -706,6 +659,7 @@ pub async fn create_event_and_trigger_outgoing_webhook<W: types::OutgoingWebhook
     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}");
     let new_event = storage::EventNew {
         event_id: event_id.clone(),
@@ -723,7 +677,7 @@ pub async fn create_event_and_trigger_outgoing_webhook<W: types::OutgoingWebhook
         Ok(event) => Ok(event),
         Err(error) => {
             if error.current_context().is_db_unique_violation() {
-                logger::info!("Merchant already notified about the event {event_id}");
+                logger::debug!("Event `{event_id}` already exists in the database");
                 return Ok(());
             } else {
                 logger::error!(event_insertion_failure=?error);
@@ -736,57 +690,132 @@ pub async fn create_event_and_trigger_outgoing_webhook<W: types::OutgoingWebhook
 
     if state.conf.webhooks.outgoing_enabled {
         let outgoing_webhook = api::OutgoingWebhook {
-            merchant_id: merchant_account.merchant_id.clone(),
+            merchant_id: merchant_id.clone(),
             event_id: event.event_id.clone(),
             event_type: event.event_type,
             content: content.clone(),
             timestamp: event.created_at,
         };
-        let state_clone = state.clone();
+
+        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 {
-            let mut error = None;
-            let result =
-                trigger_webhook_to_merchant::<W>(business_profile, outgoing_webhook, state).await;
-
-            if let Err(e) = result {
-                error.replace(
-                    serde_json::to_value(e.current_context())
-                        .into_report()
-                        .attach_printable("Failed to serialize json error response")
-                        .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
-                        .ok()
-                        .into(),
-                );
-                logger::error!(?e);
-            }
-            let outgoing_webhook_event_type = content.get_outgoing_webhook_event_type();
-            let webhook_event = OutgoingWebhookEvent::new(
-                merchant_account.merchant_id.clone(),
-                event.event_id.clone(),
-                event_type,
-                outgoing_webhook_event_type,
-                error,
-            );
-            match webhook_event.clone().try_into() {
-                Ok(event) => {
-                    state_clone.event_handler().log_event(event);
-                }
-                Err(err) => {
-                    logger::error!(error=?err, event=?webhook_event, "Error Logging Outgoing Webhook Event");
-                }
+        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());
+            .in_current_span(),
+        );
     }
 
     Ok(())
 }
 
-pub async fn trigger_webhook_to_merchant<W: types::OutgoingWebhookType>(
+#[allow(clippy::too_many_arguments)]
+pub(crate) async fn trigger_appropriate_webhook_and_raise_event(
+    state: AppState,
+    merchant_account: domain::MerchantAccount,
     business_profile: diesel_models::business_profile::BusinessProfile,
-    webhook: api::OutgoingWebhook,
+    outgoing_webhook: api::OutgoingWebhook,
+    delivery_attempt: types::WebhookDeliveryAttempt,
+    content: api::OutgoingWebhookContent,
+    event_id: String,
+    event_type: enums::EventType,
+    process_tracker: Option<storage::ProcessTracker>,
+) {
+    match merchant_account.get_compatible_connector() {
+        #[cfg(feature = "stripe")]
+        Some(api_models::enums::Connector::Stripe) => {
+            trigger_webhook_and_raise_event::<stripe_webhooks::StripeOutgoingWebhook>(
+                state,
+                business_profile,
+                outgoing_webhook,
+                delivery_attempt,
+                content,
+                event_id,
+                event_type,
+                process_tracker,
+            )
+            .await
+        }
+        _ => {
+            trigger_webhook_and_raise_event::<api_models::webhooks::OutgoingWebhook>(
+                state,
+                business_profile,
+                outgoing_webhook,
+                delivery_attempt,
+                content,
+                event_id,
+                event_type,
+                process_tracker,
+            )
+            .await
+        }
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn trigger_webhook_and_raise_event<W: types::OutgoingWebhookType>(
+    state: AppState,
+    business_profile: diesel_models::business_profile::BusinessProfile,
+    outgoing_webhook: api::OutgoingWebhook,
+    delivery_attempt: types::WebhookDeliveryAttempt,
+    content: api::OutgoingWebhookContent,
+    event_id: String,
+    event_type: enums::EventType,
+    process_tracker: Option<storage::ProcessTracker>,
+) {
+    let merchant_id = business_profile.merchant_id.clone();
+    let trigger_webhook_result = trigger_webhook_to_merchant::<W>(
+        state.clone(),
+        business_profile,
+        outgoing_webhook,
+        delivery_attempt,
+        process_tracker,
+    )
+    .await;
+
+    raise_webhooks_analytics_event(
+        state,
+        trigger_webhook_result,
+        content,
+        &merchant_id,
+        &event_id,
+        event_type,
+    );
+}
+
+async fn trigger_webhook_to_merchant<W: types::OutgoingWebhookType>(
     state: AppState,
+    business_profile: diesel_models::business_profile::BusinessProfile,
+    webhook: api::OutgoingWebhook,
+    delivery_attempt: types::WebhookDeliveryAttempt,
+    process_tracker: Option<storage::ProcessTracker>,
 ) -> CustomResult<(), errors::WebhooksFlowError> {
     let webhook_details_json = business_profile
         .webhook_details
@@ -843,40 +872,135 @@ pub async fn trigger_webhook_to_merchant<W: types::OutgoingWebhookType>(
     );
     logger::debug!(outgoing_webhook_response=?response);
 
-    match response {
-        Err(e) => {
-            // [#217]: Schedule webhook for retry.
-            Err(e).change_context(errors::WebhooksFlowError::CallToMerchantFailed)?;
-        }
-        Ok(res) => {
-            if res.status().is_success() {
-                metrics::WEBHOOK_OUTGOING_RECEIVED_COUNT.add(
-                    &metrics::CONTEXT,
-                    1,
-                    &[metrics::KeyValue::new(
-                        MERCHANT_ID,
-                        business_profile.merchant_id.clone(),
-                    )],
-                );
-                let update_event = storage::EventUpdate::UpdateWebhookNotified {
-                    is_webhook_notified: Some(true),
-                };
-                state
+    let api_client_error_handler =
+        |client_error: error_stack::Report<errors::ApiClientError>,
+         delivery_attempt: types::WebhookDeliveryAttempt| {
+            let error =
+                client_error.change_context(errors::WebhooksFlowError::CallToMerchantFailed);
+            logger::error!(
+                ?error,
+                ?delivery_attempt,
+                "An error occurred when sending webhook to merchant"
+            );
+        };
+    let success_response_handler =
+        |state: AppState,
+         merchant_id: String,
+         outgoing_webhook_event_id: String,
+         process_tracker: Option<storage::ProcessTracker>,
+         business_status: &'static str| async move {
+            metrics::WEBHOOK_OUTGOING_RECEIVED_COUNT.add(
+                &metrics::CONTEXT,
+                1,
+                &[metrics::KeyValue::new(MERCHANT_ID, merchant_id)],
+            );
+
+            let update_event = storage::EventUpdate::UpdateWebhookNotified {
+                is_webhook_notified: Some(true),
+            };
+            state
+                .store
+                .update_event(outgoing_webhook_event_id, update_event)
+                .await
+                .change_context(errors::WebhooksFlowError::WebhookEventUpdationFailed)?;
+
+            match process_tracker {
+                Some(process_tracker) => state
                     .store
-                    .update_event(outgoing_webhook_event_id, update_event)
+                    .as_scheduler()
+                    .finish_process_with_business_status(process_tracker, business_status.into())
                     .await
-                    .change_context(errors::WebhooksFlowError::WebhookEventUpdationFailed)?;
-            } else {
-                metrics::WEBHOOK_OUTGOING_NOT_RECEIVED_COUNT.add(
-                    &metrics::CONTEXT,
-                    1,
-                    &[metrics::KeyValue::new(
-                        MERCHANT_ID,
-                        business_profile.merchant_id.clone(),
-                    )],
-                );
-                // [#217]: Schedule webhook for retry.
-                Err(errors::WebhooksFlowError::NotReceivedByMerchant).into_report()?;
+                    .change_context(
+                        errors::WebhooksFlowError::OutgoingWebhookProcessTrackerTaskUpdateFailed,
+                    ),
+                None => Ok(()),
+            }
+        };
+    let error_response_handler = |merchant_id: String,
+                                  delivery_attempt: types::WebhookDeliveryAttempt,
+                                  status_code: u16,
+                                  log_message: &'static str| {
+        metrics::WEBHOOK_OUTGOING_NOT_RECEIVED_COUNT.add(
+            &metrics::CONTEXT,
+            1,
+            &[metrics::KeyValue::new(MERCHANT_ID, merchant_id)],
+        );
+
+        let error = report!(errors::WebhooksFlowError::NotReceivedByMerchant);
+        logger::warn!(?error, ?delivery_attempt, ?status_code, %log_message);
+    };
+
+    match delivery_attempt {
+        types::WebhookDeliveryAttempt::InitialAttempt => match response {
+            Err(client_error) => api_client_error_handler(client_error, delivery_attempt),
+            Ok(response) => {
+                if response.status().is_success() {
+                    success_response_handler(
+                        state.clone(),
+                        business_profile.merchant_id,
+                        outgoing_webhook_event_id,
+                        process_tracker,
+                        "INITIAL_DELIVERY_ATTEMPT_SUCCESSFUL",
+                    )
+                    .await?;
+                } else {
+                    error_response_handler(
+                        business_profile.merchant_id,
+                        delivery_attempt,
+                        response.status().as_u16(),
+                        "Ignoring error when sending webhook to merchant",
+                    );
+                }
+            }
+        },
+        types::WebhookDeliveryAttempt::AutomaticRetry => {
+            let process_tracker = process_tracker
+                .get_required_value("process_tracker")
+                .change_context(errors::WebhooksFlowError::OutgoingWebhookRetrySchedulingFailed)
+                .attach_printable("`process_tracker` is unavailable in automatic retry flow")?;
+            match response {
+                Err(client_error) => {
+                    api_client_error_handler(client_error, delivery_attempt);
+                    // Schedule a retry attempt for webhook delivery
+                    outgoing_webhook_retry::retry_webhook_delivery_task(
+                        &*state.store,
+                        &business_profile.merchant_id,
+                        process_tracker,
+                    )
+                    .await
+                    .change_context(
+                        errors::WebhooksFlowError::OutgoingWebhookRetrySchedulingFailed,
+                    )?;
+                }
+                Ok(response) => {
+                    if response.status().is_success() {
+                        success_response_handler(
+                            state.clone(),
+                            business_profile.merchant_id,
+                            outgoing_webhook_event_id,
+                            Some(process_tracker),
+                            "COMPLETED_BY_PT",
+                        )
+                        .await?;
+                    } else {
+                        error_response_handler(
+                            business_profile.merchant_id.clone(),
+                            delivery_attempt,
+                            response.status().as_u16(),
+                            "An error occurred when sending webhook to merchant",
+                        );
+                        // Schedule a retry attempt for webhook delivery
+                        outgoing_webhook_retry::retry_webhook_delivery_task(
+                            &*state.store,
+                            &business_profile.merchant_id,
+                            process_tracker,
+                        )
+                        .await
+                        .change_context(
+                            errors::WebhooksFlowError::OutgoingWebhookRetrySchedulingFailed,
+                        )?;
+                    }
+                }
             }
         }
     }
@@ -884,6 +1008,48 @@ pub async fn trigger_webhook_to_merchant<W: types::OutgoingWebhookType>(
     Ok(())
 }
 
+fn raise_webhooks_analytics_event(
+    state: AppState,
+    trigger_webhook_result: CustomResult<(), errors::WebhooksFlowError>,
+    content: api::OutgoingWebhookContent,
+    merchant_id: &str,
+    event_id: &str,
+    event_type: enums::EventType,
+) {
+    let error = if let Err(error) = trigger_webhook_result {
+        logger::error!(?error, "Failed to send webhook to merchant");
+
+        serde_json::to_value(error.current_context())
+            .into_report()
+            .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
+            .map_err(|error| {
+                logger::error!(?error, "Failed to serialize outgoing webhook error as JSON");
+                error
+            })
+            .ok()
+    } else {
+        None
+    };
+
+    let outgoing_webhook_event_content = content.get_outgoing_webhook_event_content();
+    let webhook_event = OutgoingWebhookEvent::new(
+        merchant_id.to_owned(),
+        event_id.to_owned(),
+        event_type,
+        outgoing_webhook_event_content,
+        error,
+    );
+
+    match RawEvent::try_from(webhook_event.clone()) {
+        Ok(event) => {
+            state.event_handler().log_event(event);
+        }
+        Err(error) => {
+            logger::error!(?error, event=?webhook_event, "Error logging outgoing webhook event");
+        }
+    }
+}
+
 pub async fn webhooks_wrapper<W: types::OutgoingWebhookType, Ctx: PaymentMethodRetrieve>(
     flow: &impl router_env::types::FlowMetric,
     state: AppState,
@@ -1196,7 +1362,7 @@ pub async fn webhooks_core<W: types::OutgoingWebhookType, Ctx: PaymentMethodRetr
             })?;
 
         match flow_type {
-            api::WebhookFlow::Payment => Box::pin(payments_incoming_webhook_flow::<W, Ctx>(
+            api::WebhookFlow::Payment => Box::pin(payments_incoming_webhook_flow::<Ctx>(
                 state.clone(),
                 merchant_account,
                 business_profile,
@@ -1207,7 +1373,7 @@ pub async fn webhooks_core<W: types::OutgoingWebhookType, Ctx: PaymentMethodRetr
             .await
             .attach_printable("Incoming webhook flow for payments failed")?,
 
-            api::WebhookFlow::Refund => Box::pin(refunds_incoming_webhook_flow::<W>(
+            api::WebhookFlow::Refund => Box::pin(refunds_incoming_webhook_flow(
                 state.clone(),
                 merchant_account,
                 business_profile,
@@ -1220,7 +1386,7 @@ pub async fn webhooks_core<W: types::OutgoingWebhookType, Ctx: PaymentMethodRetr
             .await
             .attach_printable("Incoming webhook flow for refunds failed")?,
 
-            api::WebhookFlow::Dispute => disputes_incoming_webhook_flow::<W>(
+            api::WebhookFlow::Dispute => disputes_incoming_webhook_flow(
                 state.clone(),
                 merchant_account,
                 business_profile,
@@ -1233,7 +1399,7 @@ pub async fn webhooks_core<W: types::OutgoingWebhookType, Ctx: PaymentMethodRetr
             .await
             .attach_printable("Incoming webhook flow for disputes failed")?,
 
-            api::WebhookFlow::BankTransfer => Box::pin(bank_transfer_webhook_flow::<W, Ctx>(
+            api::WebhookFlow::BankTransfer => Box::pin(bank_transfer_webhook_flow::<Ctx>(
                 state.clone(),
                 merchant_account,
                 business_profile,
@@ -1246,7 +1412,7 @@ pub async fn webhooks_core<W: types::OutgoingWebhookType, Ctx: PaymentMethodRetr
 
             api::WebhookFlow::ReturnResponse => WebhookResponseTracker::NoEffect,
 
-            api::WebhookFlow::Mandate => mandates_incoming_webhook_flow::<W>(
+            api::WebhookFlow::Mandate => mandates_incoming_webhook_flow(
                 state.clone(),
                 merchant_account,
                 business_profile,
@@ -1380,3 +1546,68 @@ async fn fetch_optional_mca_and_connector(
         Ok((None, connector))
     }
 }
+
+pub async fn add_outgoing_webhook_retry_task_to_process_tracker(
+    db: &dyn StorageInterface,
+    business_profile: &diesel_models::business_profile::BusinessProfile,
+    event: &storage::Event,
+) -> CustomResult<storage::ProcessTracker, errors::StorageError> {
+    let schedule_time = outgoing_webhook_retry::get_webhook_delivery_retry_schedule_time(
+        db,
+        &business_profile.merchant_id,
+        0,
+    )
+    .await
+    .ok_or(errors::StorageError::ValueNotFound(
+        "Process tracker schedule time".into(), // Can raise a better error here
+    ))
+    .into_report()
+    .attach_printable("Failed to obtain initial process tracker schedule time")?;
+
+    let tracking_data = types::OutgoingWebhookTrackingData {
+        merchant_id: business_profile.merchant_id.clone(),
+        business_profile_id: business_profile.profile_id.clone(),
+        event_type: event.event_type,
+        event_class: event.event_class,
+        primary_object_id: event.primary_object_id.clone(),
+        primary_object_type: event.primary_object_type,
+    };
+
+    let runner = storage::ProcessTrackerRunner::OutgoingWebhookRetryWorkflow;
+    let task = "OUTGOING_WEBHOOK_RETRY";
+    let tag = ["OUTGOING_WEBHOOKS"];
+    let process_tracker_id = scheduler::utils::get_process_tracker_id(
+        runner,
+        task,
+        &event.primary_object_id,
+        &business_profile.merchant_id,
+    );
+    let process_tracker_entry = storage::ProcessTrackerNew::new(
+        process_tracker_id,
+        task,
+        runner,
+        tag,
+        tracking_data,
+        schedule_time,
+    )
+    .map_err(errors::StorageError::from)?;
+
+    match db.insert_process(process_tracker_entry).await {
+        Ok(process_tracker) => {
+            crate::routes::metrics::TASKS_ADDED_COUNT.add(
+                &metrics::CONTEXT,
+                1,
+                &[add_attributes("flow", "OutgoingWebhookRetry")],
+            );
+            Ok(process_tracker)
+        }
+        Err(error) => {
+            crate::routes::metrics::TASK_ADDITION_FAILURES_COUNT.add(
+                &metrics::CONTEXT,
+                1,
+                &[add_attributes("flow", "OutgoingWebhookRetry")],
+            );
+            Err(error)
+        }
+    }
+}
diff --git a/crates/router/src/core/webhooks/types.rs b/crates/router/src/core/webhooks/types.rs
index a0e999971c5..a6bc7fbc23e 100644
--- a/crates/router/src/core/webhooks/types.rs
+++ b/crates/router/src/core/webhooks/types.rs
@@ -3,7 +3,7 @@ use common_utils::{crypto::SignMessage, ext_traits::Encode};
 use error_stack::ResultExt;
 use serde::Serialize;
 
-use crate::{core::errors, headers, services::request::Maskable};
+use crate::{core::errors, headers, services::request::Maskable, types::storage::enums};
 
 pub trait OutgoingWebhookType:
     Serialize + From<webhooks::OutgoingWebhook> + Sync + Send + std::fmt::Debug + 'static
@@ -43,3 +43,19 @@ impl OutgoingWebhookType for webhooks::OutgoingWebhook {
         header.push((headers::X_WEBHOOK_SIGNATURE.to_string(), signature.into()))
     }
 }
+
+#[derive(Debug, Clone, Copy)]
+pub(crate) enum WebhookDeliveryAttempt {
+    InitialAttempt,
+    AutomaticRetry,
+}
+
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub(crate) struct OutgoingWebhookTrackingData {
+    pub(crate) merchant_id: String,
+    pub(crate) business_profile_id: String,
+    pub(crate) event_type: enums::EventType,
+    pub(crate) event_class: enums::EventClass,
+    pub(crate) primary_object_id: String,
+    pub(crate) primary_object_type: enums::EventObjectType,
+}
diff --git a/crates/router/src/db/events.rs b/crates/router/src/db/events.rs
index af99d6a8e38..28fe4ef32d6 100644
--- a/crates/router/src/db/events.rs
+++ b/crates/router/src/db/events.rs
@@ -14,6 +14,12 @@ pub trait EventInterface {
         &self,
         event: storage::EventNew,
     ) -> CustomResult<storage::Event, errors::StorageError>;
+
+    async fn find_event_by_event_id(
+        &self,
+        event_id: &str,
+    ) -> CustomResult<storage::Event, errors::StorageError>;
+
     async fn update_event(
         &self,
         event_id: String,
@@ -31,6 +37,19 @@ impl EventInterface for Store {
         let conn = connection::pg_connection_write(self).await?;
         event.insert(&conn).await.map_err(Into::into).into_report()
     }
+
+    #[instrument(skip_all)]
+    async fn find_event_by_event_id(
+        &self,
+        event_id: &str,
+    ) -> CustomResult<storage::Event, errors::StorageError> {
+        let conn = connection::pg_connection_read(self).await?;
+        storage::Event::find_by_event_id(&conn, event_id)
+            .await
+            .map_err(Into::into)
+            .into_report()
+    }
+
     #[instrument(skip_all)]
     async fn update_event(
         &self,
@@ -74,6 +93,24 @@ impl EventInterface for MockDb {
 
         Ok(stored_event)
     }
+
+    async fn find_event_by_event_id(
+        &self,
+        event_id: &str,
+    ) -> CustomResult<storage::Event, errors::StorageError> {
+        let locked_events = self.events.lock().await;
+        locked_events
+            .iter()
+            .find(|event| event.event_id == event_id)
+            .cloned()
+            .ok_or(
+                errors::StorageError::ValueNotFound(format!(
+                    "No event available with event_id  = {event_id}"
+                ))
+                .into(),
+            )
+    }
+
     async fn update_event(
         &self,
         event_id: String,
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 35c93f334ab..8ba977befcc 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -487,6 +487,13 @@ impl EventInterface for KafkaStore {
         self.diesel_store.insert_event(event).await
     }
 
+    async fn find_event_by_event_id(
+        &self,
+        event_id: &str,
+    ) -> CustomResult<storage::Event, errors::StorageError> {
+        self.diesel_store.find_event_by_event_id(event_id).await
+    }
+
     async fn update_event(
         &self,
         event_id: String,
diff --git a/crates/router/src/events/outgoing_webhook_logs.rs b/crates/router/src/events/outgoing_webhook_logs.rs
index c2b2366fac9..bc5907b0cc4 100644
--- a/crates/router/src/events/outgoing_webhook_logs.rs
+++ b/crates/router/src/events/outgoing_webhook_logs.rs
@@ -43,10 +43,10 @@ pub enum OutgoingWebhookEventContent {
     },
 }
 pub trait OutgoingWebhookEventMetric {
-    fn get_outgoing_webhook_event_type(&self) -> Option<OutgoingWebhookEventContent>;
+    fn get_outgoing_webhook_event_content(&self) -> Option<OutgoingWebhookEventContent>;
 }
 impl OutgoingWebhookEventMetric for OutgoingWebhookContent {
-    fn get_outgoing_webhook_event_type(&self) -> Option<OutgoingWebhookEventContent> {
+    fn get_outgoing_webhook_event_content(&self) -> Option<OutgoingWebhookEventContent> {
         match self {
             Self::PaymentDetails(payment_payload) => Some(OutgoingWebhookEventContent::Payment {
                 payment_id: payment_payload.payment_id.clone(),
diff --git a/crates/router/src/routes/disputes.rs b/crates/router/src/routes/disputes.rs
index 60f6dd74186..28b3e0db211 100644
--- a/crates/router/src/routes/disputes.rs
+++ b/crates/router/src/routes/disputes.rs
@@ -224,7 +224,8 @@ pub async fn attach_dispute_evidence(
     ))
     .await
 }
-/// Diputes - Retrieve Dispute
+
+/// Disputes - Retrieve Dispute
 #[utoipa::path(
     get,
     path = "/disputes/evidence/{dispute_id}",
diff --git a/crates/router/src/routes/metrics.rs b/crates/router/src/routes/metrics.rs
index 32f2a614d03..780dd51bbde 100644
--- a/crates/router/src/routes/metrics.rs
+++ b/crates/router/src/routes/metrics.rs
@@ -118,7 +118,9 @@ counter_metric!(AUTO_PAYOUT_RETRY_GSM_MATCH_COUNT, GLOBAL_METER);
 counter_metric!(AUTO_PAYOUT_RETRY_EXHAUSTED_COUNT, GLOBAL_METER);
 counter_metric!(AUTO_RETRY_PAYOUT_COUNT, GLOBAL_METER);
 
+// Scheduler / Process Tracker related metrics
 counter_metric!(TASKS_ADDED_COUNT, GLOBAL_METER); // Tasks added to process tracker
+counter_metric!(TASK_ADDITION_FAILURES_COUNT, GLOBAL_METER); // Failures in task addition to process tracker
 counter_metric!(TASKS_RESET_COUNT, GLOBAL_METER); // Tasks reset in process tracker for requeue flow
 
 pub mod request;
diff --git a/crates/router/src/routes/metrics/request.rs b/crates/router/src/routes/metrics/request.rs
index 30db586b7f9..029c1c61f24 100644
--- a/crates/router/src/routes/metrics/request.rs
+++ b/crates/router/src/routes/metrics/request.rs
@@ -60,7 +60,7 @@ pub fn track_response_status_code<Q>(response: &ApplicationResponse<Q>) -> i64 {
         | ApplicationResponse::StatusOk
         | ApplicationResponse::TextPlain(_)
         | ApplicationResponse::Form(_)
-        | ApplicationResponse::PaymenkLinkForm(_)
+        | ApplicationResponse::PaymentLinkForm(_)
         | ApplicationResponse::FileData(_)
         | ApplicationResponse::JsonWithHeaders(_) => 200,
         ApplicationResponse::JsonForRedirection(_) => 302,
diff --git a/crates/router/src/services/api.rs b/crates/router/src/services/api.rs
index 10239834669..f01b0ffc346 100644
--- a/crates/router/src/services/api.rs
+++ b/crates/router/src/services/api.rs
@@ -845,7 +845,7 @@ pub enum ApplicationResponse<R> {
     TextPlain(String),
     JsonForRedirection(api::RedirectionResponse),
     Form(Box<RedirectionFormData>),
-    PaymenkLinkForm(Box<PaymentLinkAction>),
+    PaymentLinkForm(Box<PaymentLinkAction>),
     FileData((Vec<u8>, mime::Mime)),
     JsonWithHeaders((R, Vec<(String, String)>)),
 }
@@ -1186,7 +1186,7 @@ where
             .map_into_boxed_body()
         }
 
-        Ok(ApplicationResponse::PaymenkLinkForm(boxed_payment_link_data)) => {
+        Ok(ApplicationResponse::PaymentLinkForm(boxed_payment_link_data)) => {
             match *boxed_payment_link_data {
                 PaymentLinkAction::PaymentLinkFormData(payment_link_data) => {
                     match build_payment_link_html(payment_link_data) {
diff --git a/crates/router/src/utils.rs b/crates/router/src/utils.rs
index f218751204b..473afd1bf65 100644
--- a/crates/router/src/utils.rs
+++ b/crates/router/src/utils.rs
@@ -802,21 +802,19 @@ where
             if let Some(event_type) = event_type {
                 tokio::spawn(
                     async move {
-                        Box::pin(
-                            webhooks_core::create_event_and_trigger_appropriate_outgoing_webhook(
-                                m_state,
-                                merchant_account,
-                                business_profile,
-                                event_type,
-                                diesel_models::enums::EventClass::Payments,
-                                None,
-                                payment_id,
-                                diesel_models::enums::EventObjectType::PaymentDetails,
-                                webhooks::OutgoingWebhookContent::PaymentDetails(
-                                    payments_response_json,
-                                ),
+                        Box::pin(webhooks_core::create_event_and_trigger_outgoing_webhook(
+                            m_state,
+                            merchant_account,
+                            business_profile,
+                            event_type,
+                            diesel_models::enums::EventClass::Payments,
+                            None,
+                            payment_id,
+                            diesel_models::enums::EventObjectType::PaymentDetails,
+                            webhooks::OutgoingWebhookContent::PaymentDetails(
+                                payments_response_json,
                             ),
-                        )
+                        ))
                         .await
                     }
                     .in_current_span(),
diff --git a/crates/router/src/workflows.rs b/crates/router/src/workflows.rs
index deb5bf785fa..6158031079c 100644
--- a/crates/router/src/workflows.rs
+++ b/crates/router/src/workflows.rs
@@ -1,5 +1,6 @@
 #[cfg(feature = "email")]
 pub mod api_key_expiry;
+pub mod outgoing_webhook_retry;
 pub mod payment_sync;
 pub mod refund_router;
 pub mod tokenized_data;
diff --git a/crates/router/src/workflows/outgoing_webhook_retry.rs b/crates/router/src/workflows/outgoing_webhook_retry.rs
new file mode 100644
index 00000000000..6ce7e2454f8
--- /dev/null
+++ b/crates/router/src/workflows/outgoing_webhook_retry.rs
@@ -0,0 +1,374 @@
+use api_models::{
+    enums::EventType,
+    webhooks::{OutgoingWebhook, OutgoingWebhookContent},
+};
+use common_utils::ext_traits::{StringExt, ValueExt};
+use error_stack::ResultExt;
+use router_env::tracing::{self, instrument};
+use scheduler::{
+    consumer::{self, workflows::ProcessTrackerWorkflow},
+    types::process_data,
+    utils as scheduler_utils,
+};
+
+use crate::{
+    core::webhooks::{self as webhooks_core, types::OutgoingWebhookTrackingData},
+    db::StorageInterface,
+    errors, logger,
+    routes::AppState,
+    types::{domain, storage},
+};
+
+pub struct OutgoingWebhookRetryWorkflow;
+
+#[async_trait::async_trait]
+impl ProcessTrackerWorkflow<AppState> for OutgoingWebhookRetryWorkflow {
+    #[instrument(skip_all)]
+    async fn execute_workflow<'a>(
+        &'a self,
+        state: &'a AppState,
+        process: storage::ProcessTracker,
+    ) -> Result<(), errors::ProcessTrackerError> {
+        let tracking_data: OutgoingWebhookTrackingData = process
+            .tracking_data
+            .clone()
+            .parse_value("OutgoingWebhookTrackingData")?;
+
+        let db = &*state.store;
+        let key_store = db
+            .get_merchant_key_store_by_merchant_id(
+                &tracking_data.merchant_id,
+                &db.get_master_key().to_vec().into(),
+            )
+            .await?;
+        let merchant_account = db
+            .find_merchant_account_by_merchant_id(&tracking_data.merchant_id, &key_store)
+            .await?;
+        let business_profile = db
+            .find_business_profile_by_profile_id(&tracking_data.business_profile_id)
+            .await?;
+
+        let event_id = format!(
+            "{}_{}",
+            tracking_data.primary_object_id, tracking_data.event_type
+        );
+        let event = db.find_event_by_event_id(&event_id).await?;
+
+        let (content, event_type) = get_outgoing_webhook_content_and_event_type(
+            state.clone(),
+            merchant_account.clone(),
+            key_store,
+            &tracking_data,
+        )
+        .await?;
+
+        match event_type {
+            // Resource status is same as the event type of the current event
+            Some(event_type) if event_type == tracking_data.event_type => {
+                let outgoing_webhook = OutgoingWebhook {
+                    merchant_id: tracking_data.merchant_id.clone(),
+                    event_id: event_id.clone(),
+                    event_type,
+                    content: content.clone(),
+                    timestamp: event.created_at,
+                };
+
+                webhooks_core::trigger_appropriate_webhook_and_raise_event(
+                    state.clone(),
+                    merchant_account,
+                    business_profile,
+                    outgoing_webhook,
+                    webhooks_core::types::WebhookDeliveryAttempt::AutomaticRetry,
+                    content,
+                    event_id,
+                    event_type,
+                    Some(process),
+                )
+                .await;
+            }
+            // Resource status has changed since the event was created, finish task
+            _ => {
+                logger::warn!(
+                    %event_id,
+                    "The current status of the resource `{}` (event type: {:?}) and the status of \
+                    the resource when the event was created (event type: {:?}) differ, finishing task",
+                    tracking_data.primary_object_id,
+                    event_type,
+                    tracking_data.event_type
+                );
+                db.as_scheduler()
+                    .finish_process_with_business_status(
+                        process.clone(),
+                        "RESOURCE_STATUS_MISMATCH".to_string(),
+                    )
+                    .await?;
+            }
+        }
+
+        Ok(())
+    }
+
+    #[instrument(skip_all)]
+    async fn error_handler<'a>(
+        &'a self,
+        state: &'a AppState,
+        process: storage::ProcessTracker,
+        error: errors::ProcessTrackerError,
+    ) -> errors::CustomResult<(), errors::ProcessTrackerError> {
+        consumer::consumer_error_handler(state.store.as_scheduler(), process, error).await
+    }
+}
+
+/// Get the schedule time for the specified retry count.
+///
+/// The schedule time can be configured in configs with this key: `pt_mapping_outgoing_webhooks`.
+///
+/// ```json
+/// {
+///   "default_mapping": {
+///     "start_after": 60,
+///     "frequency": [300],
+///     "count": [5]
+///   },
+///   "custom_merchant_mapping": {
+///     "merchant_id1": {
+///       "start_after": 30,
+///       "frequency": [300],
+///       "count": [2]
+///     }
+///   }
+/// }
+/// ```
+///
+/// This configuration value represents:
+/// - `default_mapping.start_after`: The first retry attempt should happen after 60 seconds by
+///   default.
+/// - `default_mapping.frequency` and `count`: The next 5 retries should have an interval of 300
+///   seconds between them by default.
+/// - `custom_merchant_mapping.merchant_id1`: Merchant-specific retry configuration for merchant
+///   with merchant ID `merchant_id1`.
+#[instrument(skip_all)]
+pub(crate) async fn get_webhook_delivery_retry_schedule_time(
+    db: &dyn StorageInterface,
+    merchant_id: &str,
+    retry_count: i32,
+) -> Option<time::PrimitiveDateTime> {
+    let key = "pt_mapping_outgoing_webhooks";
+
+    let result = db
+        .find_config_by_key(key)
+        .await
+        .map(|value| value.config)
+        .and_then(|config| {
+            config
+                .parse_struct("OutgoingWebhookRetryProcessTrackerMapping")
+                .change_context(errors::StorageError::DeserializationFailed)
+        });
+    let mapping = result.map_or_else(
+        |error| {
+            if error.current_context().is_db_not_found() {
+                logger::debug!("Outgoing webhooks retry config `{key}` not found, ignoring");
+            } else {
+                logger::error!(
+                    ?error,
+                    "Failed to read outgoing webhooks retry config `{key}`"
+                );
+            }
+            process_data::OutgoingWebhookRetryProcessTrackerMapping::default()
+        },
+        |mapping| {
+            logger::debug!(?mapping, "Using custom outgoing webhooks retry config");
+            mapping
+        },
+    );
+
+    let time_delta = scheduler_utils::get_outgoing_webhook_retry_schedule_time(
+        mapping,
+        merchant_id,
+        retry_count,
+    );
+
+    scheduler_utils::get_time_from_delta(time_delta)
+}
+
+/// Schedule the webhook delivery task for retry
+#[instrument(skip_all)]
+pub(crate) async fn retry_webhook_delivery_task(
+    db: &dyn StorageInterface,
+    merchant_id: &str,
+    process: storage::ProcessTracker,
+) -> errors::CustomResult<(), errors::StorageError> {
+    let schedule_time =
+        get_webhook_delivery_retry_schedule_time(db, merchant_id, process.retry_count + 1).await;
+
+    match schedule_time {
+        Some(schedule_time) => {
+            db.as_scheduler()
+                .retry_process(process, schedule_time)
+                .await
+        }
+        None => {
+            db.as_scheduler()
+                .finish_process_with_business_status(process, "RETRIES_EXCEEDED".to_string())
+                .await
+        }
+    }
+}
+
+#[instrument(skip_all)]
+async fn get_outgoing_webhook_content_and_event_type(
+    state: AppState,
+    merchant_account: domain::MerchantAccount,
+    key_store: domain::MerchantKeyStore,
+    tracking_data: &OutgoingWebhookTrackingData,
+) -> Result<(OutgoingWebhookContent, Option<EventType>), errors::ProcessTrackerError> {
+    use api_models::{
+        mandates::MandateId,
+        payments::{HeaderPayload, PaymentIdType, PaymentsResponse, PaymentsRetrieveRequest},
+        refunds::{RefundResponse, RefundsRetrieveRequest},
+    };
+
+    use crate::{
+        core::{
+            disputes::retrieve_dispute,
+            mandate::get_mandate,
+            payment_methods::Oss,
+            payments::{payments_core, CallConnectorAction, PaymentStatus},
+            refunds::refund_retrieve_core,
+        },
+        services::{ApplicationResponse, AuthFlow},
+        types::{
+            api::{DisputeId, PSync},
+            transformers::ForeignFrom,
+        },
+    };
+
+    match tracking_data.event_class {
+        diesel_models::enums::EventClass::Payments => {
+            let payment_id = tracking_data.primary_object_id.clone();
+            let request = PaymentsRetrieveRequest {
+                resource_id: PaymentIdType::PaymentIntentId(payment_id),
+                merchant_id: Some(tracking_data.merchant_id.clone()),
+                force_sync: false,
+                ..Default::default()
+            };
+
+            let payments_response = match payments_core::<PSync, PaymentsResponse, _, _, _, Oss>(
+                state,
+                merchant_account,
+                key_store,
+                PaymentStatus,
+                request,
+                AuthFlow::Client,
+                CallConnectorAction::Avoid,
+                None,
+                HeaderPayload::default(),
+            )
+            .await?
+            {
+                ApplicationResponse::Json(payments_response)
+                | ApplicationResponse::JsonWithHeaders((payments_response, _)) => {
+                    Ok(payments_response)
+                }
+                ApplicationResponse::StatusOk
+                | ApplicationResponse::TextPlain(_)
+                | ApplicationResponse::JsonForRedirection(_)
+                | ApplicationResponse::Form(_)
+                | ApplicationResponse::PaymentLinkForm(_)
+                | ApplicationResponse::FileData(_) => {
+                    Err(errors::ProcessTrackerError::ResourceFetchingFailed {
+                        resource_name: tracking_data.primary_object_id.clone(),
+                    })
+                }
+            }?;
+            let event_type = Option::<EventType>::foreign_from(payments_response.status);
+            logger::debug!(current_resource_status=%payments_response.status);
+
+            Ok((
+                OutgoingWebhookContent::PaymentDetails(payments_response),
+                event_type,
+            ))
+        }
+
+        diesel_models::enums::EventClass::Refunds => {
+            let refund_id = tracking_data.primary_object_id.clone();
+            let request = RefundsRetrieveRequest {
+                refund_id,
+                force_sync: Some(false),
+                merchant_connector_details: None,
+            };
+
+            let refund = refund_retrieve_core(state, merchant_account, key_store, request).await?;
+            let event_type = Option::<EventType>::foreign_from(refund.refund_status);
+            logger::debug!(current_resource_status=%refund.refund_status);
+            let refund_response = RefundResponse::foreign_from(refund);
+
+            Ok((
+                OutgoingWebhookContent::RefundDetails(refund_response),
+                event_type,
+            ))
+        }
+
+        diesel_models::enums::EventClass::Disputes => {
+            let dispute_id = tracking_data.primary_object_id.clone();
+            let request = DisputeId { dispute_id };
+
+            let dispute_response =
+                match retrieve_dispute(state, merchant_account, request).await? {
+                    ApplicationResponse::Json(dispute_response)
+                    | ApplicationResponse::JsonWithHeaders((dispute_response, _)) => {
+                        Ok(dispute_response)
+                    }
+                    ApplicationResponse::StatusOk
+                    | ApplicationResponse::TextPlain(_)
+                    | ApplicationResponse::JsonForRedirection(_)
+                    | ApplicationResponse::Form(_)
+                    | ApplicationResponse::PaymentLinkForm(_)
+                    | ApplicationResponse::FileData(_) => {
+                        Err(errors::ProcessTrackerError::ResourceFetchingFailed {
+                            resource_name: tracking_data.primary_object_id.clone(),
+                        })
+                    }
+                }
+                .map(Box::new)?;
+            let event_type = Some(EventType::foreign_from(dispute_response.dispute_status));
+            logger::debug!(current_resource_status=%dispute_response.dispute_status);
+
+            Ok((
+                OutgoingWebhookContent::DisputeDetails(dispute_response),
+                event_type,
+            ))
+        }
+
+        diesel_models::enums::EventClass::Mandates => {
+            let mandate_id = tracking_data.primary_object_id.clone();
+            let request = MandateId { mandate_id };
+
+            let mandate_response =
+                match get_mandate(state, merchant_account, key_store, request).await? {
+                    ApplicationResponse::Json(mandate_response)
+                    | ApplicationResponse::JsonWithHeaders((mandate_response, _)) => {
+                        Ok(mandate_response)
+                    }
+                    ApplicationResponse::StatusOk
+                    | ApplicationResponse::TextPlain(_)
+                    | ApplicationResponse::JsonForRedirection(_)
+                    | ApplicationResponse::Form(_)
+                    | ApplicationResponse::PaymentLinkForm(_)
+                    | ApplicationResponse::FileData(_) => {
+                        Err(errors::ProcessTrackerError::ResourceFetchingFailed {
+                            resource_name: tracking_data.primary_object_id.clone(),
+                        })
+                    }
+                }
+                .map(Box::new)?;
+            let event_type = Option::<EventType>::foreign_from(mandate_response.status);
+            logger::debug!(current_resource_status=%mandate_response.status);
+
+            Ok((
+                OutgoingWebhookContent::MandateDetails(mandate_response),
+                event_type,
+            ))
+        }
+    }
+}
diff --git a/crates/router/src/workflows/payment_sync.rs b/crates/router/src/workflows/payment_sync.rs
index 92057b08899..c328a67a0ff 100644
--- a/crates/router/src/workflows/payment_sync.rs
+++ b/crates/router/src/workflows/payment_sync.rs
@@ -247,12 +247,12 @@ pub async fn get_sync_process_schedule_time(
         });
     let mapping = match mapping {
         Ok(x) => x,
-        Err(err) => {
-            logger::info!("Redis Mapping Error: {}", err);
+        Err(error) => {
+            logger::info!(?error, "Redis Mapping Error");
             process_data::ConnectorPTMapping::default()
         }
     };
-    let time_delta = scheduler_utils::get_schedule_time(mapping, merchant_id, retry_count + 1);
+    let time_delta = scheduler_utils::get_schedule_time(mapping, merchant_id, retry_count);
 
     Ok(scheduler_utils::get_time_from_delta(time_delta))
 }
@@ -267,7 +267,7 @@ pub async fn retry_sync_task(
     pt: storage::ProcessTracker,
 ) -> Result<bool, sch_errors::ProcessTrackerError> {
     let schedule_time =
-        get_sync_process_schedule_time(db, &connector, &merchant_id, pt.retry_count).await?;
+        get_sync_process_schedule_time(db, &connector, &merchant_id, pt.retry_count + 1).await?;
 
     match schedule_time {
         Some(s_time) => {
diff --git a/crates/scheduler/src/consumer/types/process_data.rs b/crates/scheduler/src/consumer/types/process_data.rs
index d5299493b1f..267305c37f9 100644
--- a/crates/scheduler/src/consumer/types/process_data.rs
+++ b/crates/scheduler/src/consumer/types/process_data.rs
@@ -3,7 +3,7 @@ use std::collections::HashMap;
 use diesel_models::enums;
 use serde::{Deserialize, Serialize};
 
-#[derive(Serialize, Deserialize, Clone)]
+#[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct RetryMapping {
     pub start_after: i32,
     pub frequency: Vec<i32>,
@@ -11,7 +11,6 @@ pub struct RetryMapping {
 }
 
 #[derive(Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
 pub struct ConnectorPTMapping {
     pub default_mapping: RetryMapping,
     pub custom_merchant_mapping: HashMap<String, RetryMapping>,
@@ -33,7 +32,6 @@ impl Default for ConnectorPTMapping {
 }
 
 #[derive(Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
 pub struct PaymentMethodsPTMapping {
     pub default_mapping: RetryMapping,
     pub custom_pm_mapping: HashMap<enums::PaymentMethod, RetryMapping>,
@@ -53,3 +51,43 @@ impl Default for PaymentMethodsPTMapping {
         }
     }
 }
+
+/// Configuration for outgoing webhook retries.
+#[derive(Debug, Serialize, Deserialize)]
+pub struct OutgoingWebhookRetryProcessTrackerMapping {
+    /// Default (fallback) retry configuration used when no merchant-specific retry configuration
+    /// exists.
+    pub default_mapping: RetryMapping,
+
+    /// Merchant-specific retry configuration.
+    pub custom_merchant_mapping: HashMap<String, RetryMapping>,
+}
+
+impl Default for OutgoingWebhookRetryProcessTrackerMapping {
+    fn default() -> Self {
+        Self {
+            default_mapping: RetryMapping {
+                // 1st attempt happens after 1 minute
+                start_after: 60,
+
+                frequency: vec![
+                    // 2nd and 3rd attempts happen at intervals of 5 minutes each
+                    60 * 5,
+                    // 4th, 5th, 6th, 7th and 8th attempts happen at intervals of 10 minutes each
+                    60 * 10,
+                    // 9th, 10th, 11th, 12th and 13th attempts happen at intervals of 1 hour each
+                    60 * 60,
+                    // 14th, 15th and 16th attempts happen at intervals of 6 hours each
+                    60 * 60 * 6,
+                ],
+                count: vec![
+                    2, // 2nd and 3rd attempts
+                    5, // 4th, 5th, 6th, 7th and 8th attempts
+                    5, // 9th, 10th, 11th, 12th and 13th attempts
+                    3, // 14th, 15th and 16th attempts
+                ],
+            },
+            custom_merchant_mapping: HashMap::new(),
+        }
+    }
+}
diff --git a/crates/scheduler/src/db/queue.rs b/crates/scheduler/src/db/queue.rs
index 2c02b405d81..a463111ab3b 100644
--- a/crates/scheduler/src/db/queue.rs
+++ b/crates/scheduler/src/db/queue.rs
@@ -144,7 +144,7 @@ impl QueueInterface for MockDb {
     ) -> CustomResult<Vec<storage::ProcessTracker>, ProcessTrackerError> {
         // [#172]: Implement function for `MockDb`
         Err(ProcessTrackerError::ResourceFetchingFailed {
-            resource_name: "consumer_tasks",
+            resource_name: "consumer_tasks".to_string(),
         })?
     }
 
diff --git a/crates/scheduler/src/errors.rs b/crates/scheduler/src/errors.rs
index 78a0bdee624..6a14bd2611e 100644
--- a/crates/scheduler/src/errors.rs
+++ b/crates/scheduler/src/errors.rs
@@ -34,7 +34,7 @@ pub enum ProcessTrackerError {
     #[error("Failed to fetch processes from database")]
     ProcessFetchingFailed,
     #[error("Failed while fetching: {resource_name}")]
-    ResourceFetchingFailed { resource_name: &'static str },
+    ResourceFetchingFailed { resource_name: String },
     #[error("Failed while executing: {flow}")]
     FlowExecutionError { flow: &'static str },
     #[error("Not Implemented")]
diff --git a/crates/scheduler/src/utils.rs b/crates/scheduler/src/utils.rs
index 174efd637e9..48d02f7fced 100644
--- a/crates/scheduler/src/utils.rs
+++ b/crates/scheduler/src/utils.rs
@@ -342,22 +342,46 @@ pub fn get_pm_schedule_time(
     }
 }
 
-/// Get the delay based on the retry count
-fn get_delay<'a>(
+pub fn get_outgoing_webhook_retry_schedule_time(
+    mapping: process_data::OutgoingWebhookRetryProcessTrackerMapping,
+    merchant_name: &str,
     retry_count: i32,
-    mut array: impl Iterator<Item = (&'a i32, &'a i32)>,
 ) -> Option<i32> {
-    match array.next() {
-        Some(ele) => {
-            let v = retry_count - ele.0;
-            if v <= 0 {
-                Some(*ele.1)
-            } else {
-                get_delay(v, array)
-            }
+    let retry_mapping = match mapping.custom_merchant_mapping.get(merchant_name) {
+        Some(map) => map.clone(),
+        None => mapping.default_mapping,
+    };
+
+    // For first try, get the `start_after` time
+    if retry_count == 0 {
+        Some(retry_mapping.start_after)
+    } else {
+        get_delay(
+            retry_count,
+            retry_mapping
+                .count
+                .iter()
+                .zip(retry_mapping.frequency.iter()),
+        )
+    }
+}
+
+/// Get the delay based on the retry count
+fn get_delay<'a>(retry_count: i32, array: impl Iterator<Item = (&'a i32, &'a i32)>) -> Option<i32> {
+    // Preferably, fix this by using unsigned ints
+    if retry_count <= 0 {
+        return None;
+    }
+
+    let mut cumulative_count = 0;
+    for (&count, &frequency) in array {
+        cumulative_count += count;
+        if cumulative_count >= retry_count {
+            return Some(frequency);
         }
-        None => None,
     }
+
+    None
 }
 
 pub(crate) async fn lock_acquire_release<T, F, Fut>(
@@ -392,3 +416,38 @@ where
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_get_delay() {
+        let count = [10, 5, 3, 2];
+        let frequency = [300, 600, 1800, 3600];
+
+        let retry_counts_and_expected_delays = [
+            (-4, None),
+            (-2, None),
+            (0, None),
+            (4, Some(300)),
+            (7, Some(300)),
+            (10, Some(300)),
+            (12, Some(600)),
+            (16, Some(1800)),
+            (18, Some(1800)),
+            (20, Some(3600)),
+            (24, None),
+            (30, None),
+        ];
+
+        for (retry_count, expected_delay) in retry_counts_and_expected_delays {
+            let delay = get_delay(retry_count, count.iter().zip(frequency.iter()));
+
+            assert_eq!(
+                delay, expected_delay,
+                "Delay and expected delay differ for `retry_count` = {retry_count}"
+            );
+        }
+    }
+}