Bug: [FEATURE] Process tracker flow for PaymentMethod Status update



### Feature Description

For vaulting flows, currently, the `PaymentMethodStatus` is kept as `awaiting_data` during the generation of client_secret, we want to implement a process tracker flow to update the status to `inactive` upon session expiry.

### Possible Implementation

Implementing a new flow for `PaymentMethodStatus` update.

### Have you spent some time checking if this feature request has been raised before?

- [X] I checked and didn't find a similar issue

### Have you read the Contributing Guidelines?

- [X] I have read the [Contributing Guidelines](https://github.com/juspay/hyperswitch/blob/main/docs/CONTRIBUTING.md)

### Are you willing to submit a PR?

Yes, I am willing to submit a PR!
diff --git a/crates/diesel_models/src/process_tracker.rs b/crates/diesel_models/src/process_tracker.rs
index 135c8e2b055..342ad8453af 100644
--- a/crates/diesel_models/src/process_tracker.rs
+++ b/crates/diesel_models/src/process_tracker.rs
@@ -208,6 +208,7 @@ pub enum ProcessTrackerRunner {
     ApiKeyExpiryWorkflow,
     OutgoingWebhookRetryWorkflow,
     AttachPayoutAccountWorkflow,
+    PaymentMethodStatusUpdateWorkflow,
 }
 
 #[cfg(test)]
diff --git a/crates/router/src/bin/scheduler.rs b/crates/router/src/bin/scheduler.rs
index 167e7a6b6ed..1afc09e5a6a 100644
--- a/crates/router/src/bin/scheduler.rs
+++ b/crates/router/src/bin/scheduler.rs
@@ -310,6 +310,9 @@ impl ProcessTrackerWorkflows<routes::SessionState> for WorkflowRunner {
                         )
                     }
                 }
+                storage::ProcessTrackerRunner::PaymentMethodStatusUpdateWorkflow => Ok(Box::new(
+                    workflows::payment_method_status_update::PaymentMethodStatusUpdateWorkflow,
+                )),
             }
         };
 
diff --git a/crates/router/src/core/payment_methods.rs b/crates/router/src/core/payment_methods.rs
index f0aa0480d09..a4d4f38bfaf 100644
--- a/crates/router/src/core/payment_methods.rs
+++ b/crates/router/src/core/payment_methods.rs
@@ -8,11 +8,18 @@ use api_models::payments::CardToken;
 #[cfg(feature = "payouts")]
 pub use api_models::{enums::PayoutConnectors, payouts as payout_types};
 use diesel_models::enums;
+use error_stack::ResultExt;
 use hyperswitch_domain_models::payments::{payment_attempt::PaymentAttempt, PaymentIntent};
 use router_env::{instrument, tracing};
 
 use crate::{
-    core::{errors::RouterResult, payments::helpers, pm_auth as core_pm_auth},
+    consts,
+    core::{
+        errors::{self, RouterResult},
+        payments::helpers,
+        pm_auth as core_pm_auth,
+    },
+    db,
     routes::SessionState,
     types::{
         api::{self, payments},
@@ -20,6 +27,9 @@ use crate::{
     },
 };
 
+const PAYMENT_METHOD_STATUS_UPDATE_TASK: &str = "PAYMENT_METHOD_STATUS_UPDATE";
+const PAYMENT_METHOD_STATUS_TAG: &str = "PAYMENT_METHOD_STATUS";
+
 #[instrument(skip_all)]
 pub async fn retrieve_payment_method(
     pm_data: &Option<payments::PaymentMethodData>,
@@ -94,6 +104,66 @@ pub async fn retrieve_payment_method(
     }
 }
 
+fn generate_task_id_for_payment_method_status_update_workflow(
+    key_id: &str,
+    runner: &storage::ProcessTrackerRunner,
+    task: &str,
+) -> String {
+    format!("{runner}_{task}_{key_id}")
+}
+
+pub async fn add_payment_method_status_update_task(
+    db: &dyn db::StorageInterface,
+    payment_method: &diesel_models::PaymentMethod,
+    prev_status: enums::PaymentMethodStatus,
+    curr_status: enums::PaymentMethodStatus,
+    merchant_id: &str,
+) -> Result<(), errors::ProcessTrackerError> {
+    let created_at = payment_method.created_at;
+    let schedule_time =
+        created_at.saturating_add(time::Duration::seconds(consts::DEFAULT_SESSION_EXPIRY));
+
+    let tracking_data = storage::PaymentMethodStatusTrackingData {
+        payment_method_id: payment_method.payment_method_id.clone(),
+        prev_status,
+        curr_status,
+        merchant_id: merchant_id.to_string(),
+    };
+
+    let runner = storage::ProcessTrackerRunner::PaymentMethodStatusUpdateWorkflow;
+    let task = PAYMENT_METHOD_STATUS_UPDATE_TASK;
+    let tag = [PAYMENT_METHOD_STATUS_TAG];
+
+    let process_tracker_id = generate_task_id_for_payment_method_status_update_workflow(
+        payment_method.payment_method_id.as_str(),
+        &runner,
+        task,
+    );
+    let process_tracker_entry = storage::ProcessTrackerNew::new(
+        process_tracker_id,
+        task,
+        runner,
+        tag,
+        tracking_data,
+        schedule_time,
+    )
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .attach_printable("Failed to construct PAYMENT_METHOD_STATUS_UPDATE process tracker task")?;
+
+    db
+        .insert_process(process_tracker_entry)
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable_lazy(|| {
+            format!(
+                "Failed while inserting PAYMENT_METHOD_STATUS_UPDATE reminder to process_tracker for payment_method_id: {}",
+                payment_method.payment_method_id.clone()
+            )
+        })?;
+
+    Ok(())
+}
+
 #[instrument(skip_all)]
 pub async fn retrieve_payment_method_with_token(
     state: &SessionState,
diff --git a/crates/router/src/core/payment_methods/cards.rs b/crates/router/src/core/payment_methods/cards.rs
index fad3e6d918d..ff4517a17d9 100644
--- a/crates/router/src/core/payment_methods/cards.rs
+++ b/crates/router/src/core/payment_methods/cards.rs
@@ -45,7 +45,9 @@ use crate::{
     configs::settings,
     core::{
         errors::{self, StorageErrorExt},
-        payment_methods::{transformers as payment_methods, vault},
+        payment_methods::{
+            add_payment_method_status_update_task, transformers as payment_methods, vault,
+        },
         payments::{
             helpers,
             routing::{self, SessionFlowRoutingInput},
@@ -297,6 +299,21 @@ pub async fn get_client_secret_or_add_payment_method(
         )
         .await?;
 
+        if res.status == enums::PaymentMethodStatus::AwaitingData {
+            add_payment_method_status_update_task(
+                db,
+                &res,
+                enums::PaymentMethodStatus::AwaitingData,
+                enums::PaymentMethodStatus::Inactive,
+                merchant_id,
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable(
+                "Failed to add payment method status update task in process tracker",
+            )?;
+        }
+
         Ok(services::api::ApplicationResponse::Json(
             api::PaymentMethodResponse::foreign_from(res),
         ))
@@ -357,7 +374,7 @@ pub async fn add_payment_method_data(
         .attach_printable("Unable to find payment method")?;
 
     if payment_method.status != enums::PaymentMethodStatus::AwaitingData {
-        return Err((errors::ApiErrorResponse::DuplicatePaymentMethod).into());
+        return Err((errors::ApiErrorResponse::ClientSecretExpired).into());
     }
 
     let customer_id = payment_method.customer_id.clone();
diff --git a/crates/router/src/types/storage/payment_method.rs b/crates/router/src/types/storage/payment_method.rs
index d9f96ef482c..308aa362562 100644
--- a/crates/router/src/types/storage/payment_method.rs
+++ b/crates/router/src/types/storage/payment_method.rs
@@ -110,3 +110,11 @@ impl DerefMut for PaymentsMandateReference {
         &mut self.0
     }
 }
+
+#[derive(Debug, serde::Deserialize, serde::Serialize, Clone)]
+pub struct PaymentMethodStatusTrackingData {
+    pub payment_method_id: String,
+    pub prev_status: enums::PaymentMethodStatus,
+    pub curr_status: enums::PaymentMethodStatus,
+    pub merchant_id: String,
+}
diff --git a/crates/router/src/workflows.rs b/crates/router/src/workflows.rs
index 7b29ded5185..2f858c59809 100644
--- a/crates/router/src/workflows.rs
+++ b/crates/router/src/workflows.rs
@@ -3,6 +3,7 @@ pub mod api_key_expiry;
 #[cfg(feature = "payouts")]
 pub mod attach_payout_account_workflow;
 pub mod outgoing_webhook_retry;
+pub mod payment_method_status_update;
 pub mod payment_sync;
 pub mod refund_router;
 pub mod tokenized_data;
diff --git a/crates/router/src/workflows/payment_method_status_update.rs b/crates/router/src/workflows/payment_method_status_update.rs
new file mode 100644
index 00000000000..b8e57360d90
--- /dev/null
+++ b/crates/router/src/workflows/payment_method_status_update.rs
@@ -0,0 +1,107 @@
+use common_utils::ext_traits::ValueExt;
+use scheduler::{
+    consumer::types::process_data, utils as pt_utils, workflows::ProcessTrackerWorkflow,
+};
+
+use crate::{
+    errors,
+    logger::error,
+    routes::SessionState,
+    types::storage::{self, PaymentMethodStatusTrackingData},
+};
+
+pub struct PaymentMethodStatusUpdateWorkflow;
+
+#[async_trait::async_trait]
+impl ProcessTrackerWorkflow<SessionState> for PaymentMethodStatusUpdateWorkflow {
+    async fn execute_workflow<'a>(
+        &'a self,
+        state: &'a SessionState,
+        process: storage::ProcessTracker,
+    ) -> Result<(), errors::ProcessTrackerError> {
+        let db = &*state.store;
+        let tracking_data: PaymentMethodStatusTrackingData = process
+            .tracking_data
+            .clone()
+            .parse_value("PaymentMethodStatusTrackingData")?;
+
+        let retry_count = process.retry_count;
+        let pm_id = tracking_data.payment_method_id;
+        let prev_pm_status = tracking_data.prev_status;
+        let curr_pm_status = tracking_data.curr_status;
+        let merchant_id = tracking_data.merchant_id;
+
+        let key_store = state
+            .store
+            .get_merchant_key_store_by_merchant_id(
+                merchant_id.as_str(),
+                &state.store.get_master_key().to_vec().into(),
+            )
+            .await?;
+
+        let merchant_account = db
+            .find_merchant_account_by_merchant_id(&merchant_id, &key_store)
+            .await?;
+
+        let payment_method = db
+            .find_payment_method(&pm_id, merchant_account.storage_scheme)
+            .await?;
+
+        if payment_method.status != prev_pm_status {
+            return db
+                .as_scheduler()
+                .finish_process_with_business_status(process, "PROCESS_ALREADY_COMPLETED")
+                .await
+                .map_err(Into::<errors::ProcessTrackerError>::into);
+        }
+
+        let pm_update = storage::PaymentMethodUpdate::StatusUpdate {
+            status: Some(curr_pm_status),
+        };
+
+        let res = db
+            .update_payment_method(payment_method, pm_update, merchant_account.storage_scheme)
+            .await
+            .map_err(errors::ProcessTrackerError::EStorageError);
+
+        if let Ok(_pm) = res {
+            db.as_scheduler()
+                .finish_process_with_business_status(process, "COMPLETED_BY_PT")
+                .await?;
+        } else {
+            let mapping = process_data::PaymentMethodsPTMapping::default();
+            let time_delta = if retry_count == 0 {
+                Some(mapping.default_mapping.start_after)
+            } else {
+                pt_utils::get_delay(retry_count + 1, &mapping.default_mapping.frequencies)
+            };
+
+            let schedule_time = pt_utils::get_time_from_delta(time_delta);
+
+            match schedule_time {
+                Some(s_time) => db
+                    .as_scheduler()
+                    .retry_process(process, s_time)
+                    .await
+                    .map_err(Into::<errors::ProcessTrackerError>::into)?,
+                None => db
+                    .as_scheduler()
+                    .finish_process_with_business_status(process, "RETRIES_EXCEEDED")
+                    .await
+                    .map_err(Into::<errors::ProcessTrackerError>::into)?,
+            };
+        };
+
+        Ok(())
+    }
+
+    async fn error_handler<'a>(
+        &'a self,
+        _state: &'a SessionState,
+        process: storage::ProcessTracker,
+        _error: errors::ProcessTrackerError,
+    ) -> errors::CustomResult<(), errors::ProcessTrackerError> {
+        error!(%process.id, "Failed while executing workflow");
+        Ok(())
+    }
+}
diff --git a/crates/scheduler/src/utils.rs b/crates/scheduler/src/utils.rs
index 7f073a8e10b..f3253e0ad34 100644
--- a/crates/scheduler/src/utils.rs
+++ b/crates/scheduler/src/utils.rs
@@ -342,7 +342,7 @@ pub fn get_outgoing_webhook_retry_schedule_time(
 }
 
 /// Get the delay based on the retry count
-fn get_delay<'a>(
+pub fn get_delay<'a>(
     retry_count: i32,
     frequencies: impl IntoIterator<Item = &'a (i32, i32)>,
 ) -> Option<i32> {

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

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

## Description
<!-- Describe your changes in detail -->
This PR - 
- Implemented a process tracker workflow for status update after pm client_secret expiry


### 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)?
-->
When vaulting is done within session - 
1. Create a Payment_method
```
curl --location --request POST 'http://localhost:8080/payment_methods' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: dev_cHN6hMdCFUUrNWjYSxnwhiuMsg83rXTX9jBmtHmCL6sq9a43PswY304MSt3a9blR' \
--data-raw '{
    "customer_id": "cus_LALaPcTuxxmCBORKevPC"
}'
```
Response - 
```
{
    "merchant_id": "sarthak1",
    "customer_id": "cus_LALaPcTuxxmCBORKevPC",
    "payment_method_id": "pm_gnnmqGeng12XWnsr3AWy",
    "payment_method": null,
    "payment_method_type": null,
    "card": null,
    "recurring_enabled": false,
    "installment_payment_enabled": false,
    "payment_experience": null,
    "metadata": null,
    "created": "2024-06-19T08:58:05.050Z",
    "last_used_at": null,
    "client_secret": "pm_gnnmqGeng12XWnsr3AWy_secret_ikc5JKsFFREjtR9hk5Co"
}
```
2. Vault using `/save`
```
curl --location --request POST 'http://localhost:8080/payment_methods/pm_gnnmqGeng12XWnsr3AWy/save' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: pk_dev_16e38a16176e45298a70da6c56c5028d' \
--data-raw '{
    "payment_method": "card",
    "payment_method_type": "credit",
    "payment_method_issuer": "visa",
    "client_secret": "pm_gnnmqGeng12XWnsr3AWy_secret_ikc5JKsFFREjtR9hk5Co",
    "payment_method_data": {
        "card": {
            "card_number": "4242424242424242",
            "card_exp_month": "08",
            "card_exp_year": "2028",
            "card_holder_name": "joseph"
        }
    }
}'
```
Response - 
```
{
    "merchant_id": "sarthak1",
    "customer_id": null,
    "payment_method_id": "pm_gnnmqGeng12XWnsr3AWy",
    "payment_method": "card",
    "payment_method_type": "credit",
    "card": {
        "scheme": null,
        "issuer_country": null,
        "last4_digits": "4242",
        "expiry_month": "08",
        "expiry_year": "2028",
        "card_token": null,
        "card_holder_name": "joseph",
        "card_fingerprint": null,
        "nick_name": null,
        "card_network": null,
        "card_isin": "424242",
        "card_issuer": null,
        "card_type": null,
        "saved_to_locker": true
    },
    "recurring_enabled": false,
    "installment_payment_enabled": false,
    "payment_experience": [
        "redirect_to_url"
    ],
    "metadata": null,
    "created": "2024-06-19T08:58:19.752Z",
    "last_used_at": "2024-06-19T08:58:19.752Z",
    "client_secret": "pm_gnnmqGeng12XWnsr3AWy_secret_ikc5JKsFFREjtR9hk5Co"
}
```
DB entry for PT - 
<img width="1509" alt="image" src="https://github.com/juspay/hyperswitch/assets/76486416/427e090a-79c0-4efe-ad6a-52968f71bdfb">

When vaulting is not done within session - 
1. Create Payment method - 
```
curl --location --request POST 'http://localhost:8080/payment_methods' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: dev_cHN6hMdCFUUrNWjYSxnwhiuMsg83rXTX9jBmtHmCL6sq9a43PswY304MSt3a9blR' \
--data-raw '{
    "customer_id": "cus_LALaPcTuxxmCBORKevPC"
}'
```
Response - 
```
{
    "merchant_id": "sarthak1",
    "customer_id": "cus_LALaPcTuxxmCBORKevPC",
    "payment_method_id": "pm_LlMdO3mS6HQ9EJLfDpL3",
    "payment_method": null,
    "payment_method_type": null,
    "card": null,
    "recurring_enabled": false,
    "installment_payment_enabled": false,
    "payment_experience": null,
    "metadata": null,
    "created": "2024-06-19T09:04:11.634Z",
    "last_used_at": null,
    "client_secret": "pm_LlMdO3mS6HQ9EJLfDpL3_secret_4565heI7J2bLsBgTLbJ5"
}
```
2. Vault using `/save` after session expiry - 
```
curl --location --request POST 'http://localhost:8080/payment_methods/pm_LlMdO3mS6HQ9EJLfDpL3/save' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: pk_dev_16e38a16176e45298a70da6c56c5028d' \
--data-raw '{
    "payment_method": "card",
    "payment_method_type": "credit",
    "payment_method_issuer": "visa",
    "client_secret": "pm_LlMdO3mS6HQ9EJLfDpL3_secret_4565heI7J2bLsBgTLbJ5",
    "payment_method_data": {
        "card": {
            "card_number": "4242424242424242",
            "card_exp_month": "08",
            "card_exp_year": "2028",
            "card_holder_name": "joseph"
        }
    }
}'
```
Response - 
```
{
    "error": {
        "type": "invalid_request",
        "message": "The payment method with the specified details already exists in our records",
        "code": "HE_01"
    }
}
```
DB entry after session expiry - 
<img width="1512" alt="image" src="https://github.com/juspay/hyperswitch/assets/76486416/4c574e3d-48f7-4844-b493-a306c1ac3c5a">
<img width="1512" alt="image" src="https://github.com/juspay/hyperswitch/assets/76486416/60daa679-d78a-438d-bc3c-7ad769363014">





## 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`
- [ ] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
