feat(process_tracker): Task implementation for psync payments (#7178)

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

Author: Aprabhat19

crates/api_models/src/payments.rs

@@ -8705,6 +8705,10 @@ pub struct PaymentsAttemptRecordRequest {
     #[schema(example = "2022-09-10T10:11:12Z")]
     #[serde(default, with = "common_utils::custom_serde::iso8601::option")]
     pub invoice_next_billing_time: Option<PrimitiveDateTime>,
+
+    /// source where the payment was triggered by
+    #[schema(value_type = TriggeredBy, example = "internal" )]
+    pub triggered_by: common_enums::TriggeredBy,
 }
 
 /// Error details for the payment

crates/diesel_models/src/process_tracker.rs

@@ -262,6 +262,18 @@ pub mod business_status {
     pub const EXECUTE_WORKFLOW_COMPLETE_FOR_REVIEW: &str =
         "COMPLETED_EXECUTE_TASK_TO_TRIGGER_REVIEW";
 
+    /// This status indicates that the requeue was triggered for execute task
+    pub const EXECUTE_WORKFLOW_REQUEUE: &str = "TRIGGER_REQUEUE_FOR_EXECUTE_WORKFLOW";
+
     /// This status indicates the completion of a psync task
     pub const PSYNC_WORKFLOW_COMPLETE: &str = "COMPLETED_PSYNC_TASK";
+
+    /// This status indicates that the psync task was completed to trigger the review task
+    pub const PSYNC_WORKFLOW_COMPLETE_FOR_REVIEW: &str = "COMPLETED_PSYNC_TASK_TO_TRIGGER_REVIEW";
+
+    /// This status indicates that the requeue was triggered for psync task
+    pub const PSYNC_WORKFLOW_REQUEUE: &str = "TRIGGER_REQUEUE_FOR_PSYNC_WORKFLOW";
+
+    /// This status indicates the completion of a review task
+    pub const REVIEW_WORKFLOW_COMPLETE: &str = "COMPLETED_REVIEW_TASK";
 }

crates/hyperswitch_domain_models/src/lib.rs

@@ -628,6 +628,22 @@ impl From<&api_models::payments::PaymentAttemptAmountDetails>
     }
 }
 #[cfg(feature = "v2")]
+impl From<&payments::payment_attempt::AttemptAmountDetailsSetter>
+    for api_models::payments::PaymentAttemptAmountDetails
+{
+    fn from(amount: &payments::payment_attempt::AttemptAmountDetailsSetter) -> Self {
+        Self {
+            net_amount: amount.net_amount,
+            amount_to_capture: amount.amount_to_capture,
+            surcharge_amount: amount.surcharge_amount,
+            tax_on_surcharge: amount.tax_on_surcharge,
+            amount_capturable: amount.amount_capturable,
+            shipping_cost: amount.shipping_cost,
+            order_tax_amount: amount.order_tax_amount,
+        }
+    }
+}
+#[cfg(feature = "v2")]
 impl From<&api_models::payments::RecordAttemptErrorDetails>
     for payments::payment_attempt::ErrorDetails
 {

crates/hyperswitch_domain_models/src/merchant_connector_account.rs

@@ -181,6 +181,20 @@ impl MerchantConnectorAccount {
             })
         })
     }
+    pub fn get_account_reference_id_using_payment_merchant_connector_account_id(
+        &self,
+        payment_merchant_connector_account_id: id_type::MerchantConnectorAccountId,
+    ) -> Option<String> {
+        self.feature_metadata.as_ref().and_then(|metadata| {
+            metadata.revenue_recovery.as_ref().and_then(|recovery| {
+                recovery
+                    .mca_reference
+                    .recovery_to_billing
+                    .get(&payment_merchant_connector_account_id)
+                    .cloned()
+            })
+        })
+    }
 }
 
 #[cfg(feature = "v2")]

crates/hyperswitch_domain_models/src/payments.rs

@@ -42,8 +42,8 @@ use self::payment_attempt::PaymentAttempt;
 #[cfg(feature = "v2")]
 use crate::{
     address::Address, business_profile, customer, errors, merchant_account,
-    merchant_connector_account, merchant_context, payment_address, payment_method_data, routing,
-    ApiModelToDieselModelConvertor,
+    merchant_connector_account, merchant_context, payment_address, payment_method_data,
+    revenue_recovery, routing, ApiModelToDieselModelConvertor,
 };
 #[cfg(feature = "v1")]
 use crate::{payment_method_data, RemoteStorageObject};
@@ -683,6 +683,64 @@ impl PaymentIntent {
         self.feature_metadata.clone()
     }
 
+    pub fn create_revenue_recovery_attempt_data(
+        &self,
+        revenue_recovery_metadata: api_models::payments::PaymentRevenueRecoveryMetadata,
+        billing_connector_account: &merchant_connector_account::MerchantConnectorAccount,
+    ) -> CustomResult<
+        revenue_recovery::RevenueRecoveryAttemptData,
+        errors::api_error_response::ApiErrorResponse,
+    > {
+        let merchant_reference_id = self.merchant_reference_id.clone().ok_or_else(|| {
+            error_stack::report!(
+                errors::api_error_response::ApiErrorResponse::GenericNotFoundError {
+                    message: "mandate reference id not found".to_string()
+                }
+            )
+        })?;
+
+        let connector_account_reference_id = billing_connector_account
+            .get_account_reference_id_using_payment_merchant_connector_account_id(
+                revenue_recovery_metadata.active_attempt_payment_connector_id,
+            )
+            .ok_or_else(|| {
+                error_stack::report!(
+                    errors::api_error_response::ApiErrorResponse::GenericNotFoundError {
+                        message: "connector account reference id not found".to_string()
+                    }
+                )
+            })?;
+
+        Ok(revenue_recovery::RevenueRecoveryAttemptData {
+            amount: self.amount_details.order_amount,
+            currency: self.amount_details.currency,
+            merchant_reference_id,
+            connector_transaction_id: None, // No connector id
+            error_code: None,
+            error_message: None,
+            processor_payment_method_token: revenue_recovery_metadata
+                .billing_connector_payment_details
+                .payment_processor_token,
+            connector_customer_id: revenue_recovery_metadata
+                .billing_connector_payment_details
+                .connector_customer_id,
+            connector_account_reference_id,
+            transaction_created_at: None, // would unwrap_or as now
+            status: common_enums::AttemptStatus::Started,
+            payment_method_type: self
+                .get_payment_method_type()
+                .unwrap_or(revenue_recovery_metadata.payment_method_type),
+            payment_method_sub_type: self
+                .get_payment_method_sub_type()
+                .unwrap_or(revenue_recovery_metadata.payment_method_subtype),
+            network_advice_code: None,
+            network_decline_code: None,
+            network_error_message: None,
+            retry_count: None,
+            invoice_next_billing_time: None,
+        })
+    }
+
     pub fn get_optional_customer_id(
         &self,
     ) -> CustomResult<Option<id_type::CustomerId>, common_utils::errors::ValidationError> {
@@ -863,6 +921,7 @@ pub struct RevenueRecoveryData {
     pub connector_customer_id: String,
     pub retry_count: Option<u16>,
     pub invoice_next_billing_time: Option<PrimitiveDateTime>,
+    pub triggered_by: storage_enums::enums::TriggeredBy,
 }
 
 #[cfg(feature = "v2")]

crates/hyperswitch_domain_models/src/payments/payment_attempt.rs

@@ -713,7 +713,7 @@ impl PaymentAttempt {
         let feature_metadata = PaymentAttemptFeatureMetadata {
             revenue_recovery: Some({
                 PaymentAttemptRevenueRecoveryData {
-                    attempt_triggered_by: common_enums::TriggeredBy::External,
+                    attempt_triggered_by: request.triggered_by,
                 }
             }),
         };

crates/hyperswitch_domain_models/src/payments/payment_intent.rs

@@ -361,6 +361,7 @@ pub enum PaymentIntentUpdate {
         status: common_enums::IntentStatus,
         feature_metadata: Box<Option<diesel_models::types::FeatureMetadata>>,
         updated_by: String,
+        active_attempt_id: Option<id_type::GlobalAttemptId>,
     },
     /// UpdateIntent
     UpdateIntent(Box<PaymentIntentUpdateFields>),
@@ -717,10 +718,11 @@ impl TryFrom<PaymentIntentUpdate> for diesel_models::PaymentIntentUpdateInternal
                 status,
                 feature_metadata,
                 updated_by,
+                active_attempt_id,
             } => Ok(Self {
                 status: Some(status),
                 amount_captured: None,
-                active_attempt_id: None,
+                active_attempt_id: Some(active_attempt_id),
                 modified_at: common_utils::date_time::now(),
                 amount: None,
                 currency: None,

crates/router/src/core/errors.rs

@@ -520,4 +520,6 @@ pub enum RevenueRecoveryError {
     RetryCountFetchFailed,
     #[error("Failed to get the billing threshold retry count")]
     BillingThresholdRetryCountFetchFailed,
+    #[error("Failed to create the revenue recovery attempt data")]
+    RevenueRecoveryAttemptDataCreateFailed,
 }

crates/router/src/core/payments/operations/payment_attempt_record.rs

@@ -197,6 +197,7 @@ impl<F: Send + Clone + Sync>
             connector_customer_id: request.connector_customer_id.clone(),
             retry_count: request.retry_count,
             invoice_next_billing_time: request.invoice_next_billing_time,
+            triggered_by: request.triggered_by,
         };
 
         let payment_data = PaymentAttemptRecordData {
@@ -236,12 +237,20 @@ impl<F: Clone + Sync> UpdateTracker<F, PaymentAttemptRecordData<F>, PaymentsAtte
         F: 'b + Send,
     {
         let feature_metadata = payment_data.get_updated_feature_metadata()?;
-        let payment_intent_update = hyperswitch_domain_models::payments::payment_intent::PaymentIntentUpdate::RecordUpdate
+        let active_attempt_id = match payment_data.revenue_recovery_data.triggered_by {
+            common_enums::TriggeredBy::Internal => Some(payment_data.payment_attempt.id.clone()),
+            common_enums::TriggeredBy::External => None,
+        };
+        let payment_intent_update =
+
+    hyperswitch_domain_models::payments::payment_intent::PaymentIntentUpdate::RecordUpdate
         {
             status: common_enums::IntentStatus::from(payment_data.payment_attempt.status),
             feature_metadata: Box::new(feature_metadata),
             updated_by: storage_scheme.to_string(),
-        };
+            active_attempt_id
+        }
+    ;
         payment_data.payment_intent = state
             .store
             .update_payment_intent(

crates/router/src/core/payments/operations/proxy_payments_intent.rs

@@ -40,13 +40,13 @@ impl ValidateStatusForOperation for PaymentProxyIntent {
         match intent_status {
             //Failed state is included here so that in PCR, retries can be done for failed payments, otherwise for a failed attempt it was asking for new payment_intent
             common_enums::IntentStatus::RequiresPaymentMethod
-            | common_enums::IntentStatus::Failed => Ok(()),
+            | common_enums::IntentStatus::Failed
+            | common_enums::IntentStatus::Processing => Ok(()),
             //Failed state is included here so that in PCR, retries can be done for failed payments, otherwise for a failed attempt it was asking for new payment_intent
             common_enums::IntentStatus::RequiresPaymentMethod
             | common_enums::IntentStatus::Failed => Ok(()),
             common_enums::IntentStatus::Succeeded
             | common_enums::IntentStatus::Cancelled
-            | common_enums::IntentStatus::Processing
             | common_enums::IntentStatus::RequiresCustomerAction
             | common_enums::IntentStatus::RequiresMerchantAction
             | common_enums::IntentStatus::RequiresCapture
@@ -188,26 +188,40 @@ impl<F: Send + Clone + Sync> GetTracker<F, PaymentConfirmData<F>, ProxyPaymentsR
                 .change_context(errors::ApiErrorResponse::InternalServerError)
                 .attach_printable("Failed while encrypting payment intent details")?;
 
-        let payment_attempt_domain_model: hyperswitch_domain_models::payments::payment_attempt::PaymentAttempt =
-            hyperswitch_domain_models::payments::payment_attempt::PaymentAttempt::proxy_create_domain_model(
-                &payment_intent,
-                cell_id,
-                storage_scheme,
-                request,
-                encrypted_data
-            )
-            .await?;
+        let payment_attempt = match payment_intent.active_attempt_id.clone() {
+            Some(ref active_attempt_id) => db
+                .find_payment_attempt_by_id(
+                    key_manager_state,
+                    merchant_context.get_merchant_key_store(),
+                    active_attempt_id,
+                    storage_scheme,
+                )
+                .await
+                .change_context(errors::ApiErrorResponse::PaymentNotFound)
+                .attach_printable("Could not find payment attempt")?,
+
+            None => {
+                let payment_attempt_domain_model: hyperswitch_domain_models::payments::payment_attempt::PaymentAttempt =
+                hyperswitch_domain_models::payments::payment_attempt::PaymentAttempt::proxy_create_domain_model(
+                    &payment_intent,
+                    cell_id,
+                    storage_scheme,
+                    request,
+                    encrypted_data
+                )
+                .await?;
+                db.insert_payment_attempt(
+                    key_manager_state,
+                    merchant_context.get_merchant_key_store(),
+                    payment_attempt_domain_model,
+                    storage_scheme,
+                )
+                .await
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Could not insert payment attempt")?
+            }
+        };
 
-        let payment_attempt = db
-            .insert_payment_attempt(
-                key_manager_state,
-                merchant_context.get_merchant_key_store(),
-                payment_attempt_domain_model,
-                storage_scheme,
-            )
-            .await
-            .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("Could not insert payment attempt")?;
         let processor_payment_token = request.recurring_details.processor_payment_token.clone();
 
         let payment_address = hyperswitch_domain_models::payment_address::PaymentAddress::new(