fix(process): add process tracker support for instant refunds (#5818)

Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>

crates/router/src/compatibility/stripe/refunds.rs

@@ -100,7 +100,7 @@ pub async fn refund_retrieve_with_gateway_creds(
                 None,
                 auth.key_store,
                 refund_request,
-                refunds::refund_retrieve_core,
+                refunds::refund_retrieve_core_with_refund_id,
             )
         },
         &auth::HeaderAuth(auth::ApiKeyAuth),
@@ -143,7 +143,7 @@ pub async fn refund_retrieve(
                 None,
                 auth.key_store,
                 refund_request,
-                refunds::refund_retrieve_core,
+                refunds::refund_retrieve_core_with_refund_id,
             )
         },
         &auth::HeaderAuth(auth::ApiKeyAuth),

crates/router/src/core/refunds.rs

@@ -404,24 +404,13 @@ pub async fn refund_retrieve_core(
     profile_id: Option<common_utils::id_type::ProfileId>,
     key_store: domain::MerchantKeyStore,
     request: refunds::RefundsRetrieveRequest,
+    refund: storage::Refund,
 ) -> RouterResult<storage::Refund> {
-    let refund_id = request.refund_id;
     let db = &*state.store;
-    let (merchant_id, payment_intent, payment_attempt, refund, response);
-
-    merchant_id = merchant_account.get_id();
-
-    refund = db
-        .find_refund_by_merchant_id_refund_id(
-            merchant_id,
-            refund_id.as_str(),
-            merchant_account.storage_scheme,
-        )
-        .await
-        .to_not_found_response(errors::ApiErrorResponse::RefundNotFound)?;
+    let merchant_id = merchant_account.get_id();
     core_utils::validate_profile_id_from_auth_layer(profile_id, &refund)?;
     let payment_id = &refund.payment_id;
-    payment_intent = db
+    let payment_intent = db
         .find_payment_intent_by_payment_id_merchant_id(
             &(&state).into(),
             payment_id,
@@ -432,7 +421,7 @@ pub async fn refund_retrieve_core(
         .await
         .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
 
-    payment_attempt = db
+    let payment_attempt = db
         .find_payment_attempt_by_connector_transaction_id_payment_id_merchant_id(
             &refund.connector_transaction_id,
             payment_id,
@@ -455,7 +444,7 @@ pub async fn refund_retrieve_core(
         .await
         .transpose()?;
 
-    response = if should_call_refund(&refund, request.force_sync.unwrap_or(false)) {
+    let response = if should_call_refund(&refund, request.force_sync.unwrap_or(false)) {
         sync_refund_with_gateway(
             &state,
             &merchant_account,
@@ -930,6 +919,76 @@ pub async fn refund_filter_list(
     Ok(services::ApplicationResponse::Json(filter_list))
 }
 
+#[instrument(skip_all)]
+pub async fn refund_retrieve_core_with_internal_reference_id(
+    state: SessionState,
+    merchant_account: domain::MerchantAccount,
+    profile_id: Option<common_utils::id_type::ProfileId>,
+    key_store: domain::MerchantKeyStore,
+    refund_internal_request_id: String,
+    force_sync: Option<bool>,
+) -> RouterResult<storage::Refund> {
+    let db = &*state.store;
+    let merchant_id = merchant_account.get_id();
+
+    let refund = db
+        .find_refund_by_internal_reference_id_merchant_id(
+            &refund_internal_request_id,
+            merchant_id,
+            merchant_account.storage_scheme,
+        )
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::RefundNotFound)?;
+
+    let request = refunds::RefundsRetrieveRequest {
+        refund_id: refund.refund_id.clone(),
+        force_sync,
+        merchant_connector_details: None,
+    };
+
+    Box::pin(refund_retrieve_core(
+        state.clone(),
+        merchant_account,
+        profile_id,
+        key_store,
+        request,
+        refund,
+    ))
+    .await
+}
+
+#[instrument(skip_all)]
+pub async fn refund_retrieve_core_with_refund_id(
+    state: SessionState,
+    merchant_account: domain::MerchantAccount,
+    profile_id: Option<common_utils::id_type::ProfileId>,
+    key_store: domain::MerchantKeyStore,
+    request: refunds::RefundsRetrieveRequest,
+) -> RouterResult<storage::Refund> {
+    let refund_id = request.refund_id.clone();
+    let db = &*state.store;
+    let merchant_id = merchant_account.get_id();
+
+    let refund = db
+        .find_refund_by_merchant_id_refund_id(
+            merchant_id,
+            refund_id.as_str(),
+            merchant_account.storage_scheme,
+        )
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::RefundNotFound)?;
+
+    Box::pin(refund_retrieve_core(
+        state.clone(),
+        merchant_account,
+        profile_id,
+        key_store,
+        request,
+        refund,
+    ))
+    .await
+}
+
 #[instrument(skip_all)]
 #[cfg(feature = "olap")]
 pub async fn refund_manual_update(
@@ -1126,7 +1185,7 @@ pub async fn schedule_refund_execution(
                             Ok(refund)
                         }
                         api_models::refunds::RefundType::Instant => {
-                            trigger_refund_to_gateway(
+                            let update_refund = trigger_refund_to_gateway(
                                 state,
                                 &refund,
                                 merchant_account,
@@ -1136,7 +1195,21 @@ pub async fn schedule_refund_execution(
                                 creds_identifier,
                                 charges,
                             )
+                            .await;
+
+                            match update_refund {
+                                Ok(updated_refund_data) => {
+                                    add_refund_sync_task(db, &updated_refund_data, runner)
                                         .await
+                                        .change_context(errors::ApiErrorResponse::InternalServerError)
+                                        .attach_printable_lazy(|| format!(
+                                            "Failed while pushing refund sync task in scheduler: refund_id: {}",
+                                            refund.refund_id
+                                        ))?;
+                                    Ok(updated_refund_data)
+                                }
+                                Err(err) => Err(err),
+                            }
                         }
                     }
                 }
@@ -1200,16 +1273,13 @@ pub async fn sync_refund_with_gateway_workflow(
         )
         .await?;
 
-    let response = Box::pin(refund_retrieve_core(
+    let response = Box::pin(refund_retrieve_core_with_internal_reference_id(
         state.clone(),
         merchant_account,
         None,
         key_store,
-        refunds::RefundsRetrieveRequest {
-            refund_id: refund_core.refund_internal_reference_id,
-            force_sync: Some(true),
-            merchant_connector_details: None,
-        },
+        refund_core.refund_internal_reference_id,
+        Some(true),
     ))
     .await?;
     let terminal_status = [

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

@@ -816,7 +816,7 @@ async fn refunds_incoming_webhook_flow(
         .to_not_found_response(errors::ApiErrorResponse::WebhookResourceNotFound)
         .attach_printable_lazy(|| format!("Failed while updating refund: refund_id: {refund_id}"))?
     } else {
-        Box::pin(refunds::refund_retrieve_core(
+        Box::pin(refunds::refund_retrieve_core_with_refund_id(
             state.clone(),
             merchant_account.clone(),
             None,

crates/router/src/routes/refunds.rs

@@ -107,7 +107,7 @@ pub async fn refunds_retrieve(
                 auth.profile_id,
                 auth.key_store,
                 refund_request,
-                refund_retrieve_core,
+                refund_retrieve_core_with_refund_id,
             )
         },
         auth::auth_type(
@@ -162,7 +162,7 @@ pub async fn refunds_retrieve_with_body(
                 auth.profile_id,
                 auth.key_store,
                 req,
-                refund_retrieve_core,
+                refund_retrieve_core_with_refund_id,
             )
         },
         &auth::HeaderAuth(auth::ApiKeyAuth),

crates/router/src/workflows/outgoing_webhook_retry.rs

@@ -350,7 +350,7 @@ async fn get_outgoing_webhook_content_and_event_type(
             disputes::retrieve_dispute,
             mandate::get_mandate,
             payments::{payments_core, CallConnectorAction, PaymentStatus},
-            refunds::refund_retrieve_core,
+            refunds::refund_retrieve_core_with_refund_id,
         },
         services::{ApplicationResponse, AuthFlow},
         types::{
@@ -433,7 +433,7 @@ async fn get_outgoing_webhook_content_and_event_type(
                 merchant_connector_details: None,
             };
 
-            let refund = Box::pin(refund_retrieve_core(
+            let refund = Box::pin(refund_retrieve_core_with_refund_id(
                 state,
                 merchant_account,
                 None,