feat(process_tracker): Task implementation for psync payments

crates/api_models/src/payments.rs

@@ -8677,6 +8677,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_connectors/src/connectors/chargebee/transformers.rs

@@ -488,13 +488,13 @@ impl TryFrom<ChargebeeWebhookBody> for revenue_recovery::RevenueRecoveryAttemptD
         Ok(Self {
             amount,
             currency,
-            merchant_reference_id,
+            merchant_reference_ids: Some(merchant_reference_id),
             connector_transaction_id,
             error_code,
             error_message,
             processor_payment_method_token: connector_mandate_details.mandate_id,
             connector_customer_id: connector_mandate_details.customer_id,
-            connector_account_reference_id,
+            connector_account_reference_ids: Some(connector_account_reference_id),
             transaction_created_at,
             status,
             payment_method_type,

crates/hyperswitch_domain_models/src/lib.rs

@@ -624,6 +624,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

@@ -169,16 +169,20 @@ impl MerchantConnectorAccount {
 
     pub fn get_payment_merchant_connector_account_id_using_account_reference_id(
         &self,
-        account_reference_id: String,
+        account_reference_id: Option<String>,
     ) -> Option<id_type::MerchantConnectorAccountId> {
         self.feature_metadata.as_ref().and_then(|metadata| {
-            metadata.revenue_recovery.as_ref().and_then(|recovery| {
-                recovery
-                    .mca_reference
-                    .billing_to_recovery
-                    .get(&account_reference_id)
-                    .cloned()
-            })
+            metadata
+                .revenue_recovery
+                .as_ref()
+                .zip(account_reference_id)
+                .and_then(|(recovery, account_reference_id)| {
+                    recovery
+                        .mca_reference
+                        .billing_to_recovery
+                        .get(&account_reference_id)
+                        .cloned()
+                })
         })
     }
 }

crates/hyperswitch_domain_models/src/payments.rs

@@ -42,7 +42,7 @@ use self::payment_attempt::PaymentAttempt;
 #[cfg(feature = "v2")]
 use crate::{
     address::Address, business_profile, customer, errors, merchant_account,
-    merchant_connector_account, payment_address, payment_method_data, routing,
+    merchant_connector_account, payment_address, payment_method_data, revenue_recovery, routing,
     ApiModelToDieselModelConvertor,
 };
 #[cfg(feature = "v1")]
@@ -675,6 +675,41 @@ 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,
+    ) -> revenue_recovery::RevenueRecoveryAttemptData {
+        revenue_recovery::RevenueRecoveryAttemptData {
+            amount: self.amount_details.order_amount,
+            currency: self.amount_details.currency,
+            merchant_reference_ids: self.merchant_reference_id.clone(), // is the intent 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_ids: None,
+            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> {
@@ -855,6 +890,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

@@ -705,7 +705,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/hyperswitch_domain_models/src/revenue_recovery.rs

@@ -13,7 +13,7 @@ pub struct RevenueRecoveryAttemptData {
     /// currency of the transaction
     pub currency: common_enums::Currency,
     /// merchant reference id at billing connector. ex: invoice_id
-    pub merchant_reference_id: id_type::PaymentReferenceId,
+    pub merchant_reference_ids: Option<id_type::PaymentReferenceId>,
     /// transaction id reference at payment connector
     pub connector_transaction_id: Option<util_types::ConnectorTransactionId>,
     /// error code sent by billing connector.
@@ -25,7 +25,7 @@ pub struct RevenueRecoveryAttemptData {
     /// customer id at payment connector for which mandate is attached.
     pub connector_customer_id: String,
     /// Payment gateway identifier id at billing processor.
-    pub connector_account_reference_id: String,
+    pub connector_account_reference_ids: Option<String>,
     /// timestamp at which transaction has been created at billing connector
     pub transaction_created_at: Option<PrimitiveDateTime>,
     /// transaction status at billing connector equivalent to payment attempt status.
@@ -225,13 +225,13 @@ impl From<&BillingConnectorPaymentsSyncResponse> for RevenueRecoveryAttemptData
         Self {
             amount: data.amount,
             currency: data.currency,
-            merchant_reference_id: data.merchant_reference_id.clone(),
+            merchant_reference_ids: Some(data.merchant_reference_id.clone()),
             connector_transaction_id: data.connector_transaction_id.clone(),
             error_code: data.error_code.clone(),
             error_message: data.error_message.clone(),
             processor_payment_method_token: data.processor_payment_method_token.clone(),
             connector_customer_id: data.connector_customer_id.clone(),
-            connector_account_reference_id: data.connector_account_reference_id.clone(),
+            connector_account_reference_ids: Some(data.connector_account_reference_id.clone()),
             transaction_created_at: data.transaction_created_at,
             status: data.status,
             payment_method_type: data.payment_method_type,

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

@@ -194,6 +194,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 {
@@ -233,12 +234,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

@@ -185,26 +185,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,
+                    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,
+                    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,
-                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(