feat(webhooks): implement automatic retries for failed webhook deliveries using scheduler

.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"

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,
 }
 

crates/diesel_models/src/process_tracker.rs

@@ -229,6 +229,7 @@ pub enum ProcessTrackerRunner {
     RefundWorkflowRouter,
     DeleteTokenizeDataWorkflow,
     ApiKeyExpiryWorkflow,
+    OutgoingWebhookRetryWorkflow,
 }
 
 #[cfg(test)]

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,

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,
+                )),
             }
         };
 

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) {

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)]

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),
     )))
 }

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()
         }
     };

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),
-    }
-}