Bug: feat(router):add connector field to PaymentRevenueRecoveryMetadata and update related logic



Add connector field in PaymentRevenueRecoveryMetadata and make insert_execute_pcr_task function
which creates and inserts a process tracker entry for executing a PCR (Payment Control & Reconciliation) workflow. It generates a unique process_tracker_id, fetches the schedule time for retrying MIT payments, and validates the presence of a payment_attempt_id. If any step fails, it returns an Internal Server Error. After constructing a PcrWorkflowTrackingData object with payment details, it inserts the process tracker entry into the database. If successful, it increments a task metrics counter and returns the inserted entry; otherwise, it returns an error.
diff --git a/crates/api_models/src/payments.rs b/crates/api_models/src/payments.rs
index f9540fd08fc..c55ef483287 100644
--- a/crates/api_models/src/payments.rs
+++ b/crates/api_models/src/payments.rs
@@ -7473,6 +7473,15 @@ pub struct FeatureMetadata {
     pub payment_revenue_recovery_metadata: Option<PaymentRevenueRecoveryMetadata>,
 }
 
+#[cfg(feature = "v2")]
+impl FeatureMetadata {
+    pub fn get_retry_count(&self) -> Option<u16> {
+        self.payment_revenue_recovery_metadata
+            .as_ref()
+            .map(|metadata| metadata.total_retry_count)
+    }
+}
+
 /// additional data that might be required by hyperswitch
 #[cfg(feature = "v1")]
 #[derive(Debug, Clone, serde::Deserialize, serde::Serialize, ToSchema)]
@@ -8378,6 +8387,9 @@ pub struct PaymentRevenueRecoveryMetadata {
     /// PaymentMethod Subtype
     #[schema(example = "klarna", value_type = PaymentMethodType)]
     pub payment_method_subtype: common_enums::PaymentMethodType,
+    /// The name of the payment connector through which the payment attempt was made.
+    #[schema(value_type = Connector, example = "stripe")]
+    pub connector: common_enums::connector_enums::Connector,
 }
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 #[cfg(feature = "v2")]
diff --git a/crates/diesel_models/src/process_tracker.rs b/crates/diesel_models/src/process_tracker.rs
index 64b9a66afba..294fb33c662 100644
--- a/crates/diesel_models/src/process_tracker.rs
+++ b/crates/diesel_models/src/process_tracker.rs
@@ -69,12 +69,14 @@ pub struct ProcessTrackerNew {
 }
 
 impl ProcessTrackerNew {
+    #[allow(clippy::too_many_arguments)]
     pub fn new<T>(
         process_tracker_id: impl Into<String>,
         task: impl Into<String>,
         runner: ProcessTrackerRunner,
         tag: impl IntoIterator<Item = impl Into<String>>,
         tracking_data: T,
+        retry_count: Option<i32>,
         schedule_time: PrimitiveDateTime,
         api_version: ApiVersion,
     ) -> StorageResult<Self>
@@ -87,7 +89,7 @@ impl ProcessTrackerNew {
             name: Some(task.into()),
             tag: tag.into_iter().map(Into::into).collect(),
             runner: Some(runner.to_string()),
-            retry_count: 0,
+            retry_count: retry_count.unwrap_or(0),
             schedule_time: Some(schedule_time),
             rule: String::new(),
             tracking_data: tracking_data
diff --git a/crates/diesel_models/src/types.rs b/crates/diesel_models/src/types.rs
index b79974c33fd..b1fa2197034 100644
--- a/crates/diesel_models/src/types.rs
+++ b/crates/diesel_models/src/types.rs
@@ -159,6 +159,8 @@ pub struct PaymentRevenueRecoveryMetadata {
     pub payment_method_type: common_enums::enums::PaymentMethod,
     /// PaymentMethod Subtype
     pub payment_method_subtype: common_enums::enums::PaymentMethodType,
+    /// The name of the payment connector through which the payment attempt was made.
+    pub connector: common_enums::connector_enums::Connector,
 }
 
 #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
diff --git a/crates/hyperswitch_domain_models/src/lib.rs b/crates/hyperswitch_domain_models/src/lib.rs
index f2dddf39b03..bb8cc6b50ca 100644
--- a/crates/hyperswitch_domain_models/src/lib.rs
+++ b/crates/hyperswitch_domain_models/src/lib.rs
@@ -272,6 +272,7 @@ impl ApiModelToDieselModelConvertor<ApiRevenueRecoveryMetadata> for PaymentReven
             ),
             payment_method_type: from.payment_method_type,
             payment_method_subtype: from.payment_method_subtype,
+            connector: from.connector,
         }
     }
 
@@ -286,6 +287,7 @@ impl ApiModelToDieselModelConvertor<ApiRevenueRecoveryMetadata> for PaymentReven
                 .convert_back(),
             payment_method_type: self.payment_method_type,
             payment_method_subtype: self.payment_method_subtype,
+            connector: self.connector,
         }
     }
 }
diff --git a/crates/hyperswitch_domain_models/src/payments.rs b/crates/hyperswitch_domain_models/src/payments.rs
index 36cc1c5f808..9921cee8a63 100644
--- a/crates/hyperswitch_domain_models/src/payments.rs
+++ b/crates/hyperswitch_domain_models/src/payments.rs
@@ -424,24 +424,6 @@ impl PaymentIntent {
             .and_then(|metadata| metadata.get_payment_method_type())
     }
 
-    pub fn set_payment_connector_transmission(
-        &self,
-        feature_metadata: Option<FeatureMetadata>,
-        status: bool,
-    ) -> Option<Box<FeatureMetadata>> {
-        feature_metadata.map(|fm| {
-            let mut updated_metadata = fm;
-            if let Some(ref mut rrm) = updated_metadata.payment_revenue_recovery_metadata {
-                rrm.payment_connector_transmission = if status {
-                    common_enums::PaymentConnectorTransmission::ConnectorCallFailed
-                } else {
-                    common_enums::PaymentConnectorTransmission::ConnectorCallSucceeded
-                };
-            }
-            Box::new(updated_metadata)
-        })
-    }
-
     pub fn get_connector_customer_id_from_feature_metadata(&self) -> Option<String> {
         self.feature_metadata
             .as_ref()
@@ -774,11 +756,13 @@ where
         let payment_intent_feature_metadata = self.payment_intent.get_feature_metadata();
 
         let revenue_recovery = self.payment_intent.get_revenue_recovery_metadata();
-
-        let payment_revenue_recovery_metadata =
-            Some(diesel_models::types::PaymentRevenueRecoveryMetadata {
+        let payment_attempt_connector = self.payment_attempt.connector.clone();
+        let payment_revenue_recovery_metadata = match payment_attempt_connector {
+            Some(connector) => Some(diesel_models::types::PaymentRevenueRecoveryMetadata {
                 // Update retry count by one.
-                total_retry_count: revenue_recovery.map_or(1, |data| (data.total_retry_count + 1)),
+                total_retry_count: revenue_recovery
+                    .as_ref()
+                    .map_or(1, |data| (data.total_retry_count + 1)),
                 // Since this is an external system call, marking this payment_connector_transmission to ConnectorCallSucceeded.
                 payment_connector_transmission:
                     common_enums::PaymentConnectorTransmission::ConnectorCallSucceeded,
@@ -799,7 +783,14 @@ where
                     },
                 payment_method_type: self.payment_attempt.payment_method_type,
                 payment_method_subtype: self.payment_attempt.payment_method_subtype,
-            });
+                connector: connector.parse().map_err(|err| {
+                    router_env::logger::error!(?err, "Failed to parse connector string to enum");
+                    errors::api_error_response::ApiErrorResponse::InternalServerError
+                })?,
+            }),
+            None => Err(errors::api_error_response::ApiErrorResponse::InternalServerError)
+                .attach_printable("Connector not found in payment attempt")?,
+        };
         Ok(Some(FeatureMetadata {
             redirect_response: payment_intent_feature_metadata
                 .as_ref()
diff --git a/crates/hyperswitch_domain_models/src/revenue_recovery.rs b/crates/hyperswitch_domain_models/src/revenue_recovery.rs
index 8adf5676f97..3e1bc93d91c 100644
--- a/crates/hyperswitch_domain_models/src/revenue_recovery.rs
+++ b/crates/hyperswitch_domain_models/src/revenue_recovery.rs
@@ -61,7 +61,6 @@ pub enum RecoveryAction {
     /// Invalid event has been received.
     InvalidAction,
 }
-
 pub struct RecoveryPaymentIntent {
     pub payment_id: id_type::GlobalPaymentId,
     pub status: common_enums::IntentStatus,
@@ -75,10 +74,11 @@ pub struct RecoveryPaymentAttempt {
 }
 
 impl RecoveryPaymentAttempt {
-    pub fn get_attempt_triggered_by(self) -> Option<common_enums::TriggeredBy> {
-        self.feature_metadata.and_then(|metadata| {
+    pub fn get_attempt_triggered_by(&self) -> Option<common_enums::TriggeredBy> {
+        self.feature_metadata.as_ref().and_then(|metadata| {
             metadata
                 .revenue_recovery
+                .as_ref()
                 .map(|recovery| recovery.attempt_triggered_by)
         })
     }
diff --git a/crates/hyperswitch_domain_models/src/router_data.rs b/crates/hyperswitch_domain_models/src/router_data.rs
index ae17e69710a..7eb8a8091aa 100644
--- a/crates/hyperswitch_domain_models/src/router_data.rs
+++ b/crates/hyperswitch_domain_models/src/router_data.rs
@@ -475,12 +475,24 @@ impl
     ) -> PaymentIntentUpdate {
         let amount_captured = self.get_captured_amount(payment_data);
         let status = payment_data.payment_attempt.status.is_terminal_status();
-        let updated_feature_metadata = payment_data
-            .payment_intent
-            .set_payment_connector_transmission(
-                payment_data.payment_intent.feature_metadata.clone(),
-                status,
-            );
+        let updated_feature_metadata =
+            payment_data
+                .payment_intent
+                .feature_metadata
+                .clone()
+                .map(|mut feature_metadata| {
+                    if let Some(ref mut payment_revenue_recovery_metadata) =
+                        feature_metadata.payment_revenue_recovery_metadata
+                    {
+                        payment_revenue_recovery_metadata.payment_connector_transmission =
+                            if self.response.is_ok() {
+                                common_enums::PaymentConnectorTransmission::ConnectorCallSucceeded
+                            } else {
+                                common_enums::PaymentConnectorTransmission::ConnectorCallFailed
+                            };
+                    }
+                    Box::new(feature_metadata)
+                });
 
         match self.response {
             Ok(ref _response) => PaymentIntentUpdate::ConfirmIntentPostUpdate {
diff --git a/crates/router/src/core/api_keys.rs b/crates/router/src/core/api_keys.rs
index fc87a490b65..bb970a4956e 100644
--- a/crates/router/src/core/api_keys.rs
+++ b/crates/router/src/core/api_keys.rs
@@ -228,6 +228,7 @@ pub async fn add_api_key_expiry_task(
         API_KEY_EXPIRY_RUNNER,
         [API_KEY_EXPIRY_TAG],
         api_key_expiry_tracker,
+        None,
         schedule_time,
         hyperswitch_domain_models::consts::API_VERSION,
     )
diff --git a/crates/router/src/core/errors.rs b/crates/router/src/core/errors.rs
index b32fe76f5ec..d922a9bc526 100644
--- a/crates/router/src/core/errors.rs
+++ b/crates/router/src/core/errors.rs
@@ -475,6 +475,8 @@ pub enum RevenueRecoveryError {
     PaymentIntentFetchFailed,
     #[error("Failed to fetch payment attempt")]
     PaymentAttemptFetchFailed,
+    #[error("Failed to fetch payment attempt")]
+    PaymentAttemptIdNotFound,
     #[error("Failed to get revenue recovery invoice webhook")]
     InvoiceWebhookProcessingFailed,
     #[error("Failed to get revenue recovery invoice transaction")]
@@ -485,4 +487,10 @@ pub enum RevenueRecoveryError {
     WebhookAuthenticationFailed,
     #[error("Payment merchant connector account not found using acccount reference id")]
     PaymentMerchantConnectorAccountNotFound,
+    #[error("Failed to fetch primitive date_time")]
+    ScheduleTimeFetchFailed,
+    #[error("Failed to create process tracker")]
+    ProcessTrackerCreationError,
+    #[error("Failed to get the response from process tracker")]
+    ProcessTrackerResponseError,
 }
diff --git a/crates/router/src/core/passive_churn_recovery.rs b/crates/router/src/core/passive_churn_recovery.rs
index 958baeee3f9..3b4ca0d7af6 100644
--- a/crates/router/src/core/passive_churn_recovery.rs
+++ b/crates/router/src/core/passive_churn_recovery.rs
@@ -142,6 +142,7 @@ async fn insert_psync_pcr_task(
         runner,
         tag,
         psync_workflow_tracking_data,
+        None,
         schedule_time,
         hyperswitch_domain_models::consts::API_VERSION,
     )
diff --git a/crates/router/src/core/payment_methods.rs b/crates/router/src/core/payment_methods.rs
index f9a6f9a9eb9..85b66f291d8 100644
--- a/crates/router/src/core/payment_methods.rs
+++ b/crates/router/src/core/payment_methods.rs
@@ -493,6 +493,7 @@ pub async fn add_payment_method_status_update_task(
         runner,
         tag,
         tracking_data,
+        None,
         schedule_time,
         hyperswitch_domain_models::consts::API_VERSION,
     )
diff --git a/crates/router/src/core/payment_methods/vault.rs b/crates/router/src/core/payment_methods/vault.rs
index f65f048885c..93c61793c16 100644
--- a/crates/router/src/core/payment_methods/vault.rs
+++ b/crates/router/src/core/payment_methods/vault.rs
@@ -1401,6 +1401,7 @@ pub async fn add_delete_tokenized_data_task(
         runner,
         tag,
         tracking_data,
+        None,
         schedule_time,
         hyperswitch_domain_models::consts::API_VERSION,
     )
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index d072463020a..0ab33dedf68 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -5941,6 +5941,7 @@ pub async fn add_process_sync_task(
         runner,
         tag,
         tracking_data,
+        None,
         schedule_time,
         hyperswitch_domain_models::consts::API_VERSION,
     )
diff --git a/crates/router/src/core/payments/operations/proxy_payments_intent.rs b/crates/router/src/core/payments/operations/proxy_payments_intent.rs
index c029f1439e1..bdf20635dfa 100644
--- a/crates/router/src/core/payments/operations/proxy_payments_intent.rs
+++ b/crates/router/src/core/payments/operations/proxy_payments_intent.rs
@@ -160,12 +160,6 @@ impl<F: Send + Clone + Sync> GetTracker<F, PaymentConfirmData<F>, ProxyPaymentsR
 
         self.validate_status_for_operation(payment_intent.status)?;
 
-        let client_secret = header_payload
-            .client_secret
-            .as_ref()
-            .get_required_value("client_secret header")?;
-        payment_intent.validate_client_secret(client_secret)?;
-
         let cell_id = state.conf.cell_information.id.clone();
 
         let batch_encrypted_data = domain_types::crypto_operation(
diff --git a/crates/router/src/core/payouts.rs b/crates/router/src/core/payouts.rs
index f1c8d29ca61..21a560e781b 100644
--- a/crates/router/src/core/payouts.rs
+++ b/crates/router/src/core/payouts.rs
@@ -3092,6 +3092,7 @@ pub async fn add_external_account_addition_task(
         runner,
         tag,
         tracking_data,
+        None,
         schedule_time,
         hyperswitch_domain_models::consts::API_VERSION,
     )
diff --git a/crates/router/src/core/refunds.rs b/crates/router/src/core/refunds.rs
index 1057f47b4c5..b850420dfd7 100644
--- a/crates/router/src/core/refunds.rs
+++ b/crates/router/src/core/refunds.rs
@@ -1605,6 +1605,7 @@ pub async fn add_refund_sync_task(
         runner,
         tag,
         refund_workflow_tracking_data,
+        None,
         schedule_time,
         hyperswitch_domain_models::consts::API_VERSION,
     )
@@ -1643,6 +1644,7 @@ pub async fn add_refund_execute_task(
         runner,
         tag,
         refund_workflow_tracking_data,
+        None,
         schedule_time,
         hyperswitch_domain_models::consts::API_VERSION,
     )
diff --git a/crates/router/src/core/webhooks/outgoing.rs b/crates/router/src/core/webhooks/outgoing.rs
index 595171114f7..92c29282b21 100644
--- a/crates/router/src/core/webhooks/outgoing.rs
+++ b/crates/router/src/core/webhooks/outgoing.rs
@@ -553,6 +553,7 @@ pub(crate) async fn add_outgoing_webhook_retry_task_to_process_tracker(
         runner,
         tag,
         tracking_data,
+        None,
         schedule_time,
         hyperswitch_domain_models::consts::API_VERSION,
     )
diff --git a/crates/router/src/core/webhooks/recovery_incoming.rs b/crates/router/src/core/webhooks/recovery_incoming.rs
index f27ee47d6aa..60e06ab0782 100644
--- a/crates/router/src/core/webhooks/recovery_incoming.rs
+++ b/crates/router/src/core/webhooks/recovery_incoming.rs
@@ -1,18 +1,22 @@
 use api_models::{payments as api_payments, webhooks};
-use common_utils::ext_traits::AsyncExt;
+use common_utils::{ext_traits::AsyncExt, id_type};
+use diesel_models::{process_tracker as storage, schema::process_tracker::retry_count};
 use error_stack::{report, ResultExt};
-use hyperswitch_domain_models::revenue_recovery;
+use hyperswitch_domain_models::{errors::api_error_response, revenue_recovery};
 use hyperswitch_interfaces::webhooks as interface_webhooks;
 use router_env::{instrument, tracing};
+use serde_with::rust::unwrap_or_skip;
 
 use crate::{
     core::{
         errors::{self, CustomResult},
         payments,
     },
-    routes::{app::ReqState, SessionState},
+    db::StorageInterface,
+    routes::{app::ReqState, metrics, SessionState},
     services::{self, connector_integration_interface},
-    types::{api, domain},
+    types::{api, domain, storage::passive_churn_recovery as storage_churn_recovery},
+    workflows::passive_churn_recovery_workflow,
 };
 
 #[allow(clippy::too_many_arguments)]
@@ -122,6 +126,7 @@ pub async fn recovery_incoming_webhook_flow(
     };
 
     let attempt_triggered_by = payment_attempt
+        .as_ref()
         .and_then(revenue_recovery::RecoveryPaymentAttempt::get_attempt_triggered_by);
 
     let action = revenue_recovery::RecoveryAction::get_action(event_type, attempt_triggered_by);
@@ -129,10 +134,21 @@ pub async fn recovery_incoming_webhook_flow(
     match action {
         revenue_recovery::RecoveryAction::CancelInvoice => todo!(),
         revenue_recovery::RecoveryAction::ScheduleFailedPayment => {
-            todo!()
+            Ok(RevenueRecoveryAttempt::insert_execute_pcr_task(
+                &*state.store,
+                merchant_account.get_id().to_owned(),
+                payment_intent,
+                business_profile.get_id().to_owned(),
+                payment_attempt.map(|attempt| attempt.attempt_id.clone()),
+                storage::ProcessTrackerRunner::PassiveRecoveryWorkflow,
+            )
+            .await
+            .change_context(errors::RevenueRecoveryError::InvoiceWebhookProcessingFailed)?)
         }
         revenue_recovery::RecoveryAction::SuccessPaymentExternal => {
-            todo!()
+            // Need to add recovery stop flow for this scenario
+            router_env::logger::info!("Payment has been succeeded via external system");
+            Ok(webhooks::WebhookResponseTracker::NoEffect)
         }
         revenue_recovery::RecoveryAction::PendingPayment => {
             router_env::logger::info!(
@@ -217,8 +233,7 @@ impl RevenueRecoveryInvoice {
         key_store: &domain::MerchantKeyStore,
     ) -> CustomResult<revenue_recovery::RecoveryPaymentIntent, errors::RevenueRecoveryError> {
         let payload = api_payments::PaymentsCreateIntentRequest::from(&self.0);
-        let global_payment_id =
-            common_utils::id_type::GlobalPaymentId::generate(&state.conf.cell_information.id);
+        let global_payment_id = id_type::GlobalPaymentId::generate(&state.conf.cell_information.id);
 
         let create_intent_response = Box::pin(payments::payments_intent_core::<
             hyperswitch_domain_models::router_flow_types::payments::PaymentCreateIntent,
@@ -264,7 +279,7 @@ impl RevenueRecoveryAttempt {
         merchant_account: &domain::MerchantAccount,
         profile: &domain::Profile,
         key_store: &domain::MerchantKeyStore,
-        payment_id: common_utils::id_type::GlobalPaymentId,
+        payment_id: id_type::GlobalPaymentId,
     ) -> CustomResult<Option<revenue_recovery::RecoveryPaymentAttempt>, errors::RevenueRecoveryError>
     {
         let attempt_response = Box::pin(payments::payments_core::<
@@ -332,8 +347,8 @@ impl RevenueRecoveryAttempt {
         merchant_account: &domain::MerchantAccount,
         profile: &domain::Profile,
         key_store: &domain::MerchantKeyStore,
-        payment_id: common_utils::id_type::GlobalPaymentId,
-        billing_connector_account_id: &common_utils::id_type::MerchantConnectorAccountId,
+        payment_id: id_type::GlobalPaymentId,
+        billing_connector_account_id: &id_type::MerchantConnectorAccountId,
         payment_connector_account: Option<domain::MerchantConnectorAccount>,
     ) -> CustomResult<revenue_recovery::RecoveryPaymentAttempt, errors::RevenueRecoveryError> {
         let request_payload = self
@@ -372,7 +387,7 @@ impl RevenueRecoveryAttempt {
 
     pub fn create_payment_record_request(
         &self,
-        billing_merchant_connector_account_id: &common_utils::id_type::MerchantConnectorAccountId,
+        billing_merchant_connector_account_id: &id_type::MerchantConnectorAccountId,
         payment_merchant_connector_account: Option<domain::MerchantConnectorAccount>,
     ) -> api_payments::PaymentsAttemptRecordRequest {
         let amount_details = api_payments::PaymentAttemptAmountDetails::from(&self.0);
@@ -431,4 +446,80 @@ impl RevenueRecoveryAttempt {
             )?;
         Ok(payment_merchant_connector_account)
     }
+
+    async fn insert_execute_pcr_task(
+        db: &dyn StorageInterface,
+        merchant_id: id_type::MerchantId,
+        payment_intent: revenue_recovery::RecoveryPaymentIntent,
+        profile_id: id_type::ProfileId,
+        payment_attempt_id: Option<id_type::GlobalAttemptId>,
+        runner: storage::ProcessTrackerRunner,
+    ) -> CustomResult<webhooks::WebhookResponseTracker, errors::RevenueRecoveryError> {
+        let task = "EXECUTE_WORKFLOW";
+
+        let payment_id = payment_intent.payment_id.clone();
+
+        let process_tracker_id = format!("{runner}_{task}_{}", payment_id.get_string_repr());
+
+        let total_retry_count = payment_intent
+            .feature_metadata
+            .and_then(|feature_metadata| feature_metadata.get_retry_count())
+            .unwrap_or(0);
+
+        let schedule_time =
+            passive_churn_recovery_workflow::get_schedule_time_to_retry_mit_payments(
+                db,
+                &merchant_id,
+                (total_retry_count + 1).into(),
+            )
+            .await
+            .map_or_else(
+                || {
+                    Err(
+                        report!(errors::RevenueRecoveryError::ScheduleTimeFetchFailed)
+                            .attach_printable("Failed to get schedule time for pcr workflow"),
+                    )
+                },
+                Ok, // Simply returns `time` wrapped in `Ok`
+            )?;
+
+        let payment_attempt_id = payment_attempt_id
+            .ok_or(report!(
+                errors::RevenueRecoveryError::PaymentAttemptIdNotFound
+            ))
+            .attach_printable("payment attempt id is required for pcr workflow tracking")?;
+
+        let execute_workflow_tracking_data = storage_churn_recovery::PcrWorkflowTrackingData {
+            global_payment_id: payment_id.clone(),
+            merchant_id,
+            profile_id,
+            payment_attempt_id,
+        };
+
+        let tag = ["PCR"];
+
+        let process_tracker_entry = storage::ProcessTrackerNew::new(
+            process_tracker_id,
+            task,
+            runner,
+            tag,
+            execute_workflow_tracking_data,
+            Some(total_retry_count.into()),
+            schedule_time,
+            common_enums::ApiVersion::V2,
+        )
+        .change_context(errors::RevenueRecoveryError::ProcessTrackerCreationError)
+        .attach_printable("Failed to construct process tracker entry")?;
+
+        db.insert_process(process_tracker_entry)
+            .await
+            .change_context(errors::RevenueRecoveryError::ProcessTrackerResponseError)
+            .attach_printable("Failed to enter process_tracker_entry in DB")?;
+        metrics::TASKS_ADDED_COUNT.add(1, router_env::metric_attributes!(("flow", "ExecutePCR")));
+
+        Ok(webhooks::WebhookResponseTracker::Payment {
+            payment_id,
+            status: payment_intent.status,
+        })
+    }
 }

## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [ ] New feature
- [x] Enhancement
- [ ] Dependency updates
- [ ] Refactoring
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
Added connector field in `PaymentRevenueRecoveryMetadata` and made `insert_execute_pcr_task` function 
 which creates and inserts a process tracker entry for executing a PCR (Payment Control & Reconciliation) workflow. It generates a unique `process_tracker_id`, fetches the schedule time for retrying MIT payments, and validates the presence of a payment_attempt_id. If any step fails, it returns an Internal Server Error. After constructing a `PcrWorkflowTrackingData` object with payment details, it inserts the process tracker entry into the database. If successful, it increments a task metrics counter and returns the inserted entry; otherwise, it returns an error.

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## 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).
-->


## 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)?
-->
Testcase 1:
create payment connector mca 
```
curl --location 'http://localhost:8080/v2/connector-accounts' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'x-merchant-id: cloth_seller_FFzEuYGiJhp3SfL7AXoZ' \
--header 'x-profile-id: pro_upiJkbJOOOTQny6DRwAF' \
--header 'api-key: test_admin' \
--data '{
  "connector_type": "payment_processor",
  "connector_name": "stripe",
  "connector_account_details": {
        "auth_type": "HeaderKey",
        "api_key": "stripe apikey"
    },
  "connector_webhook_details": {
    "merchant_secret": ""
  },
  "profile_id": "pro_upiJkbJOOOTQny6DRwAF"
}'
```
response :
```
{
    "connector_type": "payment_processor",
    "connector_name": "stripe",
    "connector_label": "stripe_business",
    "id": "mca_LGU3THy8GsYCwNmIY226",
    "profile_id": "pro_upiJkbJOOOTQny6DRwAF",
    "connector_account_details": {
        "auth_type": "HeaderKey",
        "api_key": "{{*****************}}"
    },
    "payment_methods_enabled": null,
    "connector_webhook_details": {
        "merchant_secret": "",
        "additional_secret": null
    },
    "metadata": null,
    "disabled": false,
    "frm_configs": null,
    "applepay_verified_domains": null,
    "pm_auth_config": null,
    "status": "active",
    "additional_merchant_data": null,
    "connector_wallets_details": null,
    "feature_metadata": null
}
```
get the payment connector mca and create billing connector mca

Create billing connector mca:-
```
curl --location 'http://localhost:8080/v2/connector-accounts' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'x-merchant-id: cloth_seller_43a1PFUTB0Ga8Fk4pMfy' \
--header 'x-profile-id: pro_upiJkbJOOOTQny6DRwAF' \
--header 'api-key: test_admin' \
--data '{
    "connector_type": "billing_processor",
    "connector_name": "chargebee",
    "connector_account_details": {
        "auth_type": "HeaderKey",
        "api_key": "{{connector_api_key}}"
    },
    "connector_webhook_details": {
        "merchant_secret": "chargebee",
        "additional_secret" : "password"
    },
    "profile_id": "pro_upiJkbJOOOTQny6DRwAF",
    "feature_metadata" : {
        "revenue_recovery" : {
            "max_retry_count" : 27,
            "billing_connector_retry_threshold" : 16,
            "billing_account_reference" :{
                "mca_LGU3THy8GsYCwNmIY226" : "gw_17BlyOUaPA0eq12MZ"
            }
        }
    }
}'
```
response:-
```
{
    "connector_type": "billing_processor",
    "connector_name": "chargebee",
    "connector_label": "chargebee_business",
    "id": "mca_JbxRy7MozRPhluRWOcWs",
    "profile_id": "pro_U7N6Id4cHXirpSZfTG08",
    "connector_account_details": {
        "auth_type": "HeaderKey",
        "api_key": "{{*****************}}"
    },
    "payment_methods_enabled": null,
    "connector_webhook_details": {
        "merchant_secret": "chargebee",
        "additional_secret": "password"
    },
    "metadata": null,
    "disabled": false,
    "frm_configs": null,
    "applepay_verified_domains": null,
    "pm_auth_config": null,
    "status": "active",
    "additional_merchant_data": null,
    "connector_wallets_details": null,
    "feature_metadata": {
        "revenue_recovery": {
            "max_retry_count": 27,
            "billing_connector_retry_threshold": 16,
            "billing_account_reference": {
                "mca_B16pBbm16yn5Ud2pnOQx": "gw_16BlyOUaPA0eq12MZ"
            }
        }
    }
}
```
Send chargebee simulated webhook either from dashboard or postman

request :
```
curl --location 'http://localhost:8080/v2/webhooks/cloth_seller_nIw80EwT6E01JF4OC7SQ/pro_U7N6Id4cHXirpSZfTG08/mca_JbxRy7MozRPhluRWOcWs' \
--header 'Content-Type: application/json' \
--header 'Authorization: ••••••' \
--data-raw '{
  "id": "ev_169vy3UaPFn4L4mf",
  "occurred_at": 1737361021,
  "source": "admin_console",
  "user": "ns22eem1r18@student.nitw.ac.in",
  "object": "event",
  "api_version": "v2",
  "content": {
    "transaction": {
      "id": "txn_169vy3UaPFmC44me",
      "customer_id": "Azq8o5UaLqWWvyGd",
      "subscription_id": "Azq8o5UaLr2WnyHG",
      "gateway_account_id": "gw_16BlyOUaPA0eq12MZ",
      "payment_source_id": "pm_169vy3UaPDo0t4hL",
      "payment_method": "card",
      "gateway": "stripe",
      "type": "payment",
      "date": 1737361019,
      "exchange_rate": 1,
      "amount": 5,
      "id_at_gateway": "ch_3QjGAJSHworDX2hs0B120C0A",
      "status": "failure",
      "updated_at": 1737361021,
      "fraud_reason": "Payment complete.",
      "resource_version": 1737361021397,
      "deleted": false,
      "object": "transaction",
      "masked_card_number": "************4242",
      "currency_code": "INR",
      "base_currency_code": "INR",
      "amount_unused": 0,
      "linked_invoices": [
        {
          "invoice_id": "9",
          "applied_amount": 500000,
          "applied_at": 1737361021,
          "invoice_date": 1737361018,
          "invoice_total": 500000,
          "invoice_status": "paid"
        }
      ],
      "linked_refunds": [],
      "initiator_type": "merchant",
      "three_d_secure": false,
      "payment_method_details": "{\"card\":{\"first_name\":\"test2\",\"last_name\":\"name2\",\"iin\":\"424242\",\"last4\":\"4242\",\"funding_type\":\"credit\",\"expiry_month\":12,\"expiry_year\":2026,\"billing_addr1\":\"asdf\",\"billing_addr2\":\"asd\",\"billing_city\":\"asdf\",\"billing_state\":\"asdfaf\",\"billing_country\":\"AF\",\"billing_zip\":\"12345\",\"masked_number\":\"************4242\",\"object\":\"card\",\"brand\":\"visa\"}}"
    },
    "invoice": {
      "id": "invoice_1234",
      "customer_id": "Azq8o5UaLqWWvyGd",
      "subscription_id": "Azq8o5UaLr2WnyHG",
      "recurring": false,
      "status": "paid",
      "price_type": "tax_exclusive",
      "date": 1737361018,
      "due_date": 1737361018,
      "net_term_days": 0,
      "exchange_rate": 1,
      "total": 5,
      "amount_paid": 0,
      "amount_adjusted": 0,
      "write_off_amount": 0,
      "credits_applied": 0,
      "amount_due": 0,
      "paid_at": 1737361019,
      "updated_at": 1737361021,
      "resource_version": 1737361021401,
      "deleted": false,
      "object": "invoice",
      "first_invoice": false,
      "amount_to_collect": 0,
      "round_off_amount": 0,
      "has_advance_charges": false,
      "currency_code": "INR",
      "base_currency_code": "INR",
      "generated_at": 1737361018,
      "is_gifted": false,
      "term_finalized": true,
      "channel": "web",
      "tax": 0,
      "line_items": [
        {
          "id": "li_169vy3UaPFmBR4md",
          "date_from": 1737361004,
          "date_to": 1737361004,
          "unit_amount": 500000,
          "quantity": 1,
          "amount": 500000,
          "pricing_model": "flat_fee",
          "is_taxed": false,
          "tax_amount": 0,
          "object": "line_item",
          "subscription_id": "Azq8o5UaLr2WnyHG",
          "customer_id": "Azq8o5UaLqWWvyGd",
          "description": "Implementation Charge",
          "entity_type": "charge_item_price",
          "entity_id": "cbdemo_implementation-charge-INR",
          "tax_exempt_reason": "tax_not_configured",
          "discount_amount": 0,
          "item_level_discount_amount": 0
        }
      ],
      "sub_total": 500000,
      "linked_payments": [
        {
          "txn_id": "txn_169vy3UaPFmC44me",
          "applied_amount": 500000,
          "applied_at": 1737361021,
          "txn_status": "success",
          "txn_date": 1737361019,
          "txn_amount": 500000
        }
      ],
      "applied_credits": [],
      "adjustment_credit_notes": [],
      "issued_credit_notes": [],
      "linked_orders": [],
      "dunning_attempts": [],
      "billing_address": {
        "first_name": "test1",
        "last_name": "name",
        "email": "johndoe@gmail.com",
        "company": "johndoe",
        "phone": "+91 83 17 575848",
        "line1": "asdf",
        "line2": "asd",
        "line3": "ahjkd",
        "city": "asdf",
        "state_code": "TG",
        "state": "Telangana",
        "country": "IN",
        "zip": "561432",
        "validation_status": "not_validated",
        "object": "billing_address"
      },
      "site_details_at_creation": {
        "timezone": "Asia/Calcutta"
      }
    },
    "customer": {
      "id": "Azq8o5UaLqWWvyGd",
      "first_name": "john",
      "last_name": "doe",
      "email": "john@gmail.com",
      "phone": "831 757 5848",
      "company": "johndoe",
      "auto_collection": "on",
      "net_term_days": 0,
      "allow_direct_debit": false,
      "created_at": 1737310670,
      "created_from_ip": "205.254.163.189",
      "taxability": "taxable",
      "updated_at": 1737360899,
      "pii_cleared": "active",
      "channel": "web",
      "resource_version": 1737360899990,
      "deleted": false,
      "object": "customer",
      "billing_address": {
        "first_name": "test1",
        "last_name": "name",
        "email": "johndoe@gmail.com",
        "company": "johndoe",
        "phone": "+91 83 17 575848",
        "line1": "asdf",
        "line2": "asd",
        "line3": "ahjkd",
        "city": "asdf",
        "state_code": "TG",
        "state": "Telangana",
        "country": "IN",
        "zip": "561432",
        "validation_status": "not_validated",
        "object": "billing_address"
      },
      "card_status": "valid",
      "promotional_credits": 0,
      "refundable_credits": 0,
      "excess_payments": 0,
      "unbilled_charges": 0,
      "preferred_currency_code": "INR",
      "mrr": 0,
      "primary_payment_source_id": "pm_169vy3UaPDo0t4hL",
      "payment_method": {
        "object": "payment_method",
        "type": "card",
        "reference_id": "cus_RcUo8xTwe0sHP7/card_1QjG2dSHworDX2hs6YIjKdML",
        "gateway": "stripe",
        "gateway_account_id": "gw_16BlyOUaPA0eq12MZ",
        "status": "valid"
      }
    },
    "subscription": {
      "id": "Azq8o5UaLr2WnyHG",
      "billing_period": 1,
      "billing_period_unit": "month",
      "customer_id": "Azq8o5UaLqWWvyGd",
      "status": "active",
      "current_term_start": 1737310793,
      "current_term_end": 1739989193,
      "next_billing_at": 1739989193,
      "created_at": 1737310793,
      "started_at": 1737310793,
      "activated_at": 1737310793,
      "created_from_ip": "205.254.163.189",
      "updated_at": 1737310799,
      "has_scheduled_changes": false,
      "channel": "web",
      "resource_version": 1737310799688,
      "deleted": false,
      "object": "subscription",
      "currency_code": "INR",
      "subscription_items": [
        {
          "item_price_id": "cbdemo_premium-INR-monthly",
          "item_type": "plan",
          "quantity": 1,
          "unit_price": 100000,
          "amount": 0,
          "free_quantity": 3,
          "object": "subscription_item"
        }
      ],
      "shipping_address": {
        "first_name": "test1",
        "last_name": "name",
        "email": "johndoe@gmail.com",
        "company": "johndoe",
        "phone": "+91 83 17 575848",
        "line1": "asdf",
        "line2": "asd",
        "line3": "ahjkd",
        "city": "asdf",
        "state_code": "TG",
        "state": "Telangana",
        "country": "IN",
        "zip": "561432",
        "validation_status": "not_validated",
        "object": "shipping_address"
      },
      "due_invoices_count": 0,
      "mrr": 0,
      "exchange_rate": 1,
      "base_currency_code": "INR",
      "has_scheduled_advance_invoices": false
    }
  },
  "event_type": "payment_failed",
  "webhook_status": "scheduled",
  "webhooks": [
    {
      "id": "whv2_6oZfZUaLmtchA5Bl",
      "webhook_status": "scheduled",
      "object": "webhook"
    }
  ]
}'
```
Response:-
200 OK (Postman)
This should create Payment intent, payment attempt in our system with right values in DB

Payment intent details in DB
<img width="1199" alt="Screenshot 2025-03-10 at 14 08 56" src="https://github.com/user-attachments/assets/baa26e0f-dd6b-42bb-90d8-2f0ce6a1aab7" />
Payment attempt details in DB
<img width="1138" alt="Screenshot 2025-03-10 at 14 09 53" src="https://github.com/user-attachments/assets/101f79e1-67a8-4aad-b8f7-dd20c565bfcc" />
Test case 2 :
change the transaction id in above simulated webhooks and hit the request again, it should create new payment attempt but update the payment intent feature metadata.
<img width="1177" alt="Screenshot 2025-03-10 at 14 26 53" src="https://github.com/user-attachments/assets/29873c6d-79b8-43b4-8ec8-fec1bb755d8f" />

Scheduling to failed payments in log(schedule_time: Some(2025-03-10 8:46:43.815835)):-
<img width="858" alt="Screenshot 2025-03-10 at 14 52 45" src="https://github.com/user-attachments/assets/c9b7f9b2-41c3-44f7-9249-a34747fe0b28" />



## 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
