Issue:
Bug: [REFACTOR] use `const` instead of `String` for `business_status`



As of now, we've around `9` distinct values for `process_tracker` and all these values are hard coded as Strings in various places.  

This makes it harder for us to keep track of the values that exist in the code base.  

As a measure to tackle this issue, we would have to create a set of `const`s in scheduler crate and use them in places where we've used hardcoded strings now. This will help us keep track of all the values that are used and helps increase readability.


Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [ ] New feature
- [ ] Enhancement
- [x] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->

When below command is executed in sandbox DB, we get a set of "distinct" `business_status`es that is specified for process tracker:

```sql
SELECT DISTINCT(business_status) FROM process_tracker;
```

For debugging and fun, execute to learn more about process tracker and its behavior:
```sql
SELECT DISTINCT(name, retry_count, business_status, status), created_at FROM process_tracker order by created_at desc LIMIT 10000;
```

The main goal for us here is to keep all the `business_status`es in a single place, confined for ease of access and increased readability. This also helps us keep track of the statuses we use.

### 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).
-->

We need to keep a track of the statuses that we use in `process_tracker`

## 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)?
-->

This should not require any testing as the change done is here just a mere replacement of hardcoded string with `const`s. If needed, the changes can be verified by following #3712's instructions. Reproduced below:

Perform PSync call from connector like `Checkout`. Payment Status should go to `processing` state. Wait for `process_tracker` to kick in and do a `force_sync` on its own. Wait for the status to get updated. I had webhooks set up (invalid configuration), even that is verified to be working.

Command used in DB: 

```sql
SELECT * FROM process_tracker ORDER BY created_at desc LIMIT 2;
```

![image](https://github.com/juspay/hyperswitch/assets/69745008/9d2ad8b1-6750-4d1a-ac5a-2493a0348e05)

![image](https://github.com/juspay/hyperswitch/assets/69745008/90d8221c-907a-4c1b-a99a-0936e0cc6225)

![image](https://github.com/juspay/hyperswitch/assets/69745008/75aaaca7-7b71-413a-9bb4-75c4652f5366)

![image](https://github.com/juspay/hyperswitch/assets/69745008/f04231d1-0703-45b4-9db1-ace2b21f8372)


## 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


Testing:


This should not require any testing as the change done is here just a mere replacement of hardcoded string with `const`s. If needed, the changes can be verified by following #3712's instructions. Reproduced below:

Perform PSync call from connector like `Checkout`. Payment Status should go to `processing` state. Wait for `process_tracker` to kick in and do a `force_sync` on its own. Wait for the status to get updated. I had webhooks set up (invalid configuration), even that is verified to be working.

Command used in DB: 

```sql
SELECT * FROM process_tracker ORDER BY created_at desc LIMIT 2;
```

![image](https://github.com/juspay/hyperswitch/assets/69745008/9d2ad8b1-6750-4d1a-ac5a-2493a0348e05)

![image](https://github.com/juspay/hyperswitch/assets/69745008/90d8221c-907a-4c1b-a99a-0936e0cc6225)

![image](https://github.com/juspay/hyperswitch/assets/69745008/75aaaca7-7b71-413a-9bb4-75c4652f5366)

![image](https://github.com/juspay/hyperswitch/assets/69745008/f04231d1-0703-45b4-9db1-ace2b21f8372)

Patch:
diff --git a/crates/diesel_models/src/process_tracker.rs b/crates/diesel_models/src/process_tracker.rs
index cd0dbed2c64..135c8e2b055 100644
--- a/crates/diesel_models/src/process_tracker.rs
+++ b/crates/diesel_models/src/process_tracker.rs
@@ -76,8 +76,6 @@ impl ProcessTrackerNew {
     where
         T: Serialize + std::fmt::Debug,
     {
-        const BUSINESS_STATUS_PENDING: &str = "Pending";
-
         let current_time = common_utils::date_time::now();
         Ok(Self {
             id: process_tracker_id.into(),
@@ -91,7 +89,7 @@ impl ProcessTrackerNew {
                 .encode_to_value()
                 .change_context(errors::DatabaseError::Others)
                 .attach_printable("Failed to serialize process tracker tracking data")?,
-            business_status: String::from(BUSINESS_STATUS_PENDING),
+            business_status: String::from(business_status::PENDING),
             status: storage_enums::ProcessTrackerStatus::New,
             event: vec![],
             created_at: current_time,
@@ -227,3 +225,42 @@ mod tests {
         assert_eq!(enum_format, ProcessTrackerRunner::PaymentsSyncWorkflow);
     }
 }
+
+pub mod business_status {
+    /// Indicates that an irrecoverable error occurred during the workflow execution.
+    pub const GLOBAL_FAILURE: &str = "GLOBAL_FAILURE";
+
+    /// Task successfully completed by consumer.
+    /// A task that reaches this status should not be retried (rescheduled for execution) later.
+    pub const COMPLETED_BY_PT: &str = "COMPLETED_BY_PT";
+
+    /// An error occurred during the workflow execution which prevents further execution and
+    /// retries.
+    /// A task that reaches this status should not be retried (rescheduled for execution) later.
+    pub const FAILURE: &str = "FAILURE";
+
+    /// The resource associated with the task was removed, due to which further retries can/should
+    /// not be done.
+    pub const REVOKED: &str = "Revoked";
+
+    /// The task was executed for the maximum possible number of times without a successful outcome.
+    /// A task that reaches this status should not be retried (rescheduled for execution) later.
+    pub const RETRIES_EXCEEDED: &str = "RETRIES_EXCEEDED";
+
+    /// The outgoing webhook was successfully delivered in the initial attempt.
+    /// Further retries of the task are not required.
+    pub const INITIAL_DELIVERY_ATTEMPT_SUCCESSFUL: &str = "INITIAL_DELIVERY_ATTEMPT_SUCCESSFUL";
+
+    /// Indicates that an error occurred during the workflow execution.
+    /// This status is typically set by the workflow error handler.
+    /// A task that reaches this status should not be retried (rescheduled for execution) later.
+    pub const GLOBAL_ERROR: &str = "GLOBAL_ERROR";
+
+    /// The resource associated with the task has been significantly modified since the task was
+    /// created, due to which further retries of the current task are not required.
+    /// A task that reaches this status should not be retried (rescheduled for execution) later.
+    pub const RESOURCE_STATUS_MISMATCH: &str = "RESOURCE_STATUS_MISMATCH";
+
+    /// Business status set for newly created tasks.
+    pub const PENDING: &str = "Pending";
+}
diff --git a/crates/router/src/bin/scheduler.rs b/crates/router/src/bin/scheduler.rs
index 7e78a355e5c..167e7a6b6ed 100644
--- a/crates/router/src/bin/scheduler.rs
+++ b/crates/router/src/bin/scheduler.rs
@@ -4,7 +4,7 @@ use std::{collections::HashMap, str::FromStr, sync::Arc};
 use actix_web::{dev::Server, web, Scope};
 use api_models::health_check::SchedulerHealthCheckResponse;
 use common_utils::ext_traits::{OptionExt, StringExt};
-use diesel_models::process_tracker as storage;
+use diesel_models::process_tracker::{self as storage, business_status};
 use error_stack::ResultExt;
 use router::{
     configs::settings::{CmdLineConf, Settings},
@@ -329,10 +329,13 @@ impl ProcessTrackerWorkflows<routes::SessionState> for WorkflowRunner {
                     let status = state
                         .get_db()
                         .as_scheduler()
-                        .finish_process_with_business_status(process, "GLOBAL_FAILURE".to_string())
+                        .finish_process_with_business_status(
+                            process,
+                            business_status::GLOBAL_FAILURE,
+                        )
                         .await;
                     if let Err(err) = status {
-                        logger::error!(%err, "Failed while performing database operation: GLOBAL_FAILURE");
+                        logger::error!(%err, "Failed while performing database operation: {}", business_status::GLOBAL_FAILURE);
                     }
                 }
             },
diff --git a/crates/router/src/core/api_keys.rs b/crates/router/src/core/api_keys.rs
index 8fe536731d3..ef81e3c190d 100644
--- a/crates/router/src/core/api_keys.rs
+++ b/crates/router/src/core/api_keys.rs
@@ -381,7 +381,9 @@ pub async fn update_api_key_expiry_task(
         retry_count: Some(0),
         schedule_time,
         tracking_data: Some(updated_api_key_expiry_workflow_model),
-        business_status: Some("Pending".to_string()),
+        business_status: Some(String::from(
+            diesel_models::process_tracker::business_status::PENDING,
+        )),
         status: Some(storage_enums::ProcessTrackerStatus::New),
         updated_at: Some(current_time),
     };
@@ -450,7 +452,7 @@ pub async fn revoke_api_key_expiry_task(
     let task_ids = vec![task_id];
     let updated_process_tracker_data = storage::ProcessTrackerUpdate::StatusUpdate {
         status: storage_enums::ProcessTrackerStatus::Finish,
-        business_status: Some("Revoked".to_string()),
+        business_status: Some(String::from(diesel_models::business_status::REVOKED)),
     };
 
     store
diff --git a/crates/router/src/core/payment_methods/vault.rs b/crates/router/src/core/payment_methods/vault.rs
index 9e20017d57f..f37385b80c6 100644
--- a/crates/router/src/core/payment_methods/vault.rs
+++ b/crates/router/src/core/payment_methods/vault.rs
@@ -1178,7 +1178,7 @@ pub async fn start_tokenize_data_workflow(
             db.as_scheduler()
                 .finish_process_with_business_status(
                     tokenize_tracker.clone(),
-                    "COMPLETED_BY_PT".to_string(),
+                    diesel_models::process_tracker::business_status::COMPLETED_BY_PT,
                 )
                 .await?;
         }
@@ -1241,7 +1241,10 @@ pub async fn retry_delete_tokenize(
         }
         None => db
             .as_scheduler()
-            .finish_process_with_business_status(pt, "RETRIES_EXCEEDED".to_string())
+            .finish_process_with_business_status(
+                pt,
+                diesel_models::process_tracker::business_status::RETRIES_EXCEEDED,
+            )
             .await
             .map_err(Into::into),
     }
diff --git a/crates/router/src/core/refunds.rs b/crates/router/src/core/refunds.rs
index 0d441e95382..164fd1ec784 100644
--- a/crates/router/src/core/refunds.rs
+++ b/crates/router/src/core/refunds.rs
@@ -9,6 +9,7 @@ use common_utils::{
     ext_traits::{AsyncExt, ValueExt},
     types::MinorUnit,
 };
+use diesel_models::process_tracker::business_status;
 use error_stack::{report, ResultExt};
 use masking::PeekInterface;
 use router_env::{instrument, tracing};
@@ -1028,7 +1029,7 @@ pub async fn sync_refund_with_gateway_workflow(
                 .as_scheduler()
                 .finish_process_with_business_status(
                     refund_tracker.clone(),
-                    "COMPLETED_BY_PT".to_string(),
+                    business_status::COMPLETED_BY_PT,
                 )
                 .await?
         }
@@ -1193,7 +1194,7 @@ pub async fn trigger_refund_execute_workflow(
             db.as_scheduler()
                 .finish_process_with_business_status(
                     refund_tracker.clone(),
-                    "COMPLETED_BY_PT".to_string(),
+                    business_status::COMPLETED_BY_PT,
                 )
                 .await?;
         }
diff --git a/crates/router/src/core/webhooks/outgoing.rs b/crates/router/src/core/webhooks/outgoing.rs
index 0be4e9272f0..07fe78f51ce 100644
--- a/crates/router/src/core/webhooks/outgoing.rs
+++ b/crates/router/src/core/webhooks/outgoing.rs
@@ -3,6 +3,7 @@ use api_models::{
     webhooks,
 };
 use common_utils::{ext_traits::Encode, request::RequestContent};
+use diesel_models::process_tracker::business_status;
 use error_stack::{report, ResultExt};
 use masking::{ExposeInterface, Mask, PeekInterface, Secret};
 use router_env::{
@@ -231,7 +232,7 @@ async fn trigger_webhook_to_merchant(
                 state
                     .store
                     .as_scheduler()
-                    .finish_process_with_business_status(process_tracker, "FAILURE".into())
+                    .finish_process_with_business_status(process_tracker, business_status::FAILURE)
                     .await
                     .change_context(
                         errors::WebhooksFlowError::OutgoingWebhookProcessTrackerTaskUpdateFailed,
@@ -304,7 +305,7 @@ async fn trigger_webhook_to_merchant(
                         state.clone(),
                         &business_profile.merchant_id,
                         process_tracker,
-                        "INITIAL_DELIVERY_ATTEMPT_SUCCESSFUL",
+                        business_status::INITIAL_DELIVERY_ATTEMPT_SUCCESSFUL,
                     )
                     .await?;
                 } else {
@@ -769,7 +770,7 @@ async fn success_response_handler(
         Some(process_tracker) => state
             .store
             .as_scheduler()
-            .finish_process_with_business_status(process_tracker, business_status.into())
+            .finish_process_with_business_status(process_tracker, business_status)
             .await
             .change_context(
                 errors::WebhooksFlowError::OutgoingWebhookProcessTrackerTaskUpdateFailed,
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 0545191318d..75b830cbb41 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -1821,7 +1821,7 @@ impl ProcessTrackerInterface for KafkaStore {
     async fn finish_process_with_business_status(
         &self,
         this: storage::ProcessTracker,
-        business_status: String,
+        business_status: &'static str,
     ) -> CustomResult<(), errors::StorageError> {
         self.diesel_store
             .finish_process_with_business_status(this, business_status)
diff --git a/crates/router/src/types/storage.rs b/crates/router/src/types/storage.rs
index 79c8b2ed2f0..f5626c267d2 100644
--- a/crates/router/src/types/storage.rs
+++ b/crates/router/src/types/storage.rs
@@ -40,7 +40,8 @@ pub mod user_role;
 use std::collections::HashMap;
 
 pub use diesel_models::{
-    ProcessTracker, ProcessTrackerNew, ProcessTrackerRunner, ProcessTrackerUpdate,
+    process_tracker::business_status, ProcessTracker, ProcessTrackerNew, ProcessTrackerRunner,
+    ProcessTrackerUpdate,
 };
 pub use hyperswitch_domain_models::payments::{
     payment_attempt::{PaymentAttempt, PaymentAttemptNew, PaymentAttemptUpdate},
diff --git a/crates/router/src/workflows/api_key_expiry.rs b/crates/router/src/workflows/api_key_expiry.rs
index dff7905a262..c24671be34d 100644
--- a/crates/router/src/workflows/api_key_expiry.rs
+++ b/crates/router/src/workflows/api_key_expiry.rs
@@ -1,5 +1,7 @@
 use common_utils::{errors::ValidationError, ext_traits::ValueExt};
-use diesel_models::{enums as storage_enums, ApiKeyExpiryTrackingData};
+use diesel_models::{
+    enums as storage_enums, process_tracker::business_status, ApiKeyExpiryTrackingData,
+};
 use router_env::logger;
 use scheduler::{workflows::ProcessTrackerWorkflow, SchedulerSessionState};
 
@@ -96,7 +98,7 @@ impl ProcessTrackerWorkflow<SessionState> for ApiKeyExpiryWorkflow {
             state
                 .get_db()
                 .as_scheduler()
-                .finish_process_with_business_status(process, "COMPLETED_BY_PT".to_string())
+                .finish_process_with_business_status(process, business_status::COMPLETED_BY_PT)
                 .await?
         }
         // If tasks are remaining that has to be scheduled
diff --git a/crates/router/src/workflows/outgoing_webhook_retry.rs b/crates/router/src/workflows/outgoing_webhook_retry.rs
index 9184a78b6b2..7560da210d1 100644
--- a/crates/router/src/workflows/outgoing_webhook_retry.rs
+++ b/crates/router/src/workflows/outgoing_webhook_retry.rs
@@ -6,6 +6,7 @@ use api_models::{
     webhooks::{OutgoingWebhook, OutgoingWebhookContent},
 };
 use common_utils::ext_traits::{StringExt, ValueExt};
+use diesel_models::process_tracker::business_status;
 use error_stack::ResultExt;
 use masking::PeekInterface;
 use router_env::tracing::{self, instrument};
@@ -197,7 +198,7 @@ impl ProcessTrackerWorkflow<SessionState> for OutgoingWebhookRetryWorkflow {
                         db.as_scheduler()
                             .finish_process_with_business_status(
                                 process.clone(),
-                                "RESOURCE_STATUS_MISMATCH".to_string(),
+                                business_status::RESOURCE_STATUS_MISMATCH,
                             )
                             .await?;
                     }
@@ -309,7 +310,7 @@ pub(crate) async fn retry_webhook_delivery_task(
         }
         None => {
             db.as_scheduler()
-                .finish_process_with_business_status(process, "RETRIES_EXCEEDED".to_string())
+                .finish_process_with_business_status(process, business_status::RETRIES_EXCEEDED)
                 .await
         }
     }
diff --git a/crates/router/src/workflows/payment_sync.rs b/crates/router/src/workflows/payment_sync.rs
index c34b8ebe73b..666de8c3376 100644
--- a/crates/router/src/workflows/payment_sync.rs
+++ b/crates/router/src/workflows/payment_sync.rs
@@ -1,4 +1,5 @@
 use common_utils::ext_traits::{OptionExt, StringExt, ValueExt};
+use diesel_models::process_tracker::business_status;
 use error_stack::ResultExt;
 use router_env::logger;
 use scheduler::{
@@ -89,7 +90,7 @@ impl ProcessTrackerWorkflow<SessionState> for PaymentsSyncWorkflow {
                 state
                     .store
                     .as_scheduler()
-                    .finish_process_with_business_status(process, "COMPLETED_BY_PT".to_string())
+                    .finish_process_with_business_status(process, business_status::COMPLETED_BY_PT)
                     .await?
             }
             _ => {
@@ -269,7 +270,7 @@ pub async fn retry_sync_task(
         }
         None => {
             db.as_scheduler()
-                .finish_process_with_business_status(pt, "RETRIES_EXCEEDED".to_string())
+                .finish_process_with_business_status(pt, business_status::RETRIES_EXCEEDED)
                 .await?;
             Ok(true)
         }
diff --git a/crates/scheduler/src/consumer.rs b/crates/scheduler/src/consumer.rs
index 3e15a7cde0f..b58e8ce60d7 100644
--- a/crates/scheduler/src/consumer.rs
+++ b/crates/scheduler/src/consumer.rs
@@ -30,7 +30,7 @@ use crate::{
 
 // Valid consumer business statuses
 pub fn valid_business_statuses() -> Vec<&'static str> {
-    vec!["Pending"]
+    vec![storage::business_status::PENDING]
 }
 
 #[instrument(skip_all)]
@@ -262,7 +262,7 @@ pub async fn consumer_error_handler(
             vec![process.id],
             storage::ProcessTrackerUpdate::StatusUpdate {
                 status: enums::ProcessTrackerStatus::Finish,
-                business_status: Some("GLOBAL_ERROR".to_string()),
+                business_status: Some(String::from(storage::business_status::GLOBAL_ERROR)),
             },
         )
         .await
diff --git a/crates/scheduler/src/consumer/workflows.rs b/crates/scheduler/src/consumer/workflows.rs
index bbece87f309..974f1ec4153 100644
--- a/crates/scheduler/src/consumer/workflows.rs
+++ b/crates/scheduler/src/consumer/workflows.rs
@@ -1,6 +1,7 @@
 use async_trait::async_trait;
 use common_utils::errors::CustomResult;
 pub use diesel_models::process_tracker as storage;
+use diesel_models::process_tracker::business_status;
 use router_env::logger;
 
 use crate::{errors, SchedulerSessionState};
@@ -45,7 +46,10 @@ pub trait ProcessTrackerWorkflows<T>: Send + Sync {
                     let status = app_state
                         .get_db()
                         .as_scheduler()
-                        .finish_process_with_business_status(process, "GLOBAL_FAILURE".to_string())
+                        .finish_process_with_business_status(
+                            process,
+                            business_status::GLOBAL_FAILURE,
+                        )
                         .await;
                     if let Err(error) = status {
                         logger::error!(?error, "Failed to update process business status");
diff --git a/crates/scheduler/src/db/process_tracker.rs b/crates/scheduler/src/db/process_tracker.rs
index feb201f8b75..c73b53b608c 100644
--- a/crates/scheduler/src/db/process_tracker.rs
+++ b/crates/scheduler/src/db/process_tracker.rs
@@ -52,7 +52,7 @@ pub trait ProcessTrackerInterface: Send + Sync + 'static {
     async fn finish_process_with_business_status(
         &self,
         this: storage::ProcessTracker,
-        business_status: String,
+        business_status: &'static str,
     ) -> CustomResult<(), errors::StorageError>;
 
     async fn find_processes_by_time_status(
@@ -166,13 +166,13 @@ impl ProcessTrackerInterface for Store {
     async fn finish_process_with_business_status(
         &self,
         this: storage::ProcessTracker,
-        business_status: String,
+        business_status: &'static str,
     ) -> CustomResult<(), errors::StorageError> {
         self.update_process(
             this,
             storage::ProcessTrackerUpdate::StatusUpdate {
                 status: storage_enums::ProcessTrackerStatus::Finish,
-                business_status: Some(business_status),
+                business_status: Some(String::from(business_status)),
             },
         )
         .await
@@ -284,7 +284,7 @@ impl ProcessTrackerInterface for MockDb {
     async fn finish_process_with_business_status(
         &self,
         _this: storage::ProcessTracker,
-        _business_status: String,
+        _business_status: &'static str,
     ) -> CustomResult<(), errors::StorageError> {
         // [#172]: Implement function for `MockDb`
         Err(errors::StorageError::MockDbError)?