Issue:
Bug: [REFACTOR]: include api key expiry workflow into process tracker



Currently, `ApiKeyExpiryWorkflow` is not part of the workflows being run in scheduler. Refactor to include it, as well as update the email sending section to the latest email refactor logic

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 -->
Currently, `ApiKeyExpiryWorkflow` is not part of the workflows being run in scheduler. This PR refactors that workflow to be included in the existing workflow list, as well as update the email sending section to the latest email refactor logic. 
Includes changes for sending an html page in email as opposed to plain text

### 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).
-->
Re-enables api key expiry reminder workflow

## 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)?
-->
1. Locally set up email config
2. Created a merchant 
3. Created an api key with some expiry (Locally changed the schedule time to be 3 minutes before the expiry)

![image](https://github.com/juspay/hyperswitch/assets/70657455/f93cb042-383d-45e9-b111-19c4381f7fe6)

Since the email is triggered 7 days before expiry, complete flow cannot be tested in sandbox. However, api_key can be created with expiry set, then check the process tracker table whether an entry has been created in the table for this api_key with schedule time being 7 days prior to expiry.

Query - 

```
select * from process_tracker where runner = 'API_KEY_EXPIRY_WORKFLOW' order by created_at desc;
```

Check the `id` field of the latest entry in the table which has to be something like  `API_KEY_EXPIRY_WORKFLOW_API_KEY_EXPIRY_{YOUR_KEY_ID}` and `schedule_time` being 7 days prior to expiry

## 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
- [ ] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable


Testing:

1. Locally set up email config
2. Created a merchant 
3. Created an api key with some expiry (Locally changed the schedule time to be 3 minutes before the expiry)

![image](https://github.com/juspay/hyperswitch/assets/70657455/f93cb042-383d-45e9-b111-19c4381f7fe6)

Since the email is triggered 7 days before expiry, complete flow cannot be tested in sandbox. However, api_key can be created with expiry set, then check the process tracker table whether an entry has been created in the table for this api_key with schedule time being 7 days prior to expiry.

Query - 

```
select * from process_tracker where runner = 'API_KEY_EXPIRY_WORKFLOW' order by created_at desc;
```

Check the `id` field of the latest entry in the table which has to be something like  `API_KEY_EXPIRY_WORKFLOW_API_KEY_EXPIRY_{YOUR_KEY_ID}` and `schedule_time` being 7 days prior to expiry

Patch:
diff --git a/crates/diesel_models/src/api_keys.rs b/crates/diesel_models/src/api_keys.rs
index 6608d4ed2fd..1875fc4dc51 100644
--- a/crates/diesel_models/src/api_keys.rs
+++ b/crates/diesel_models/src/api_keys.rs
@@ -138,7 +138,7 @@ mod diesel_impl {
 
 // Tracking data by process_tracker
 #[derive(Default, Debug, Deserialize, Serialize, Clone)]
-pub struct ApiKeyExpiryWorkflow {
+pub struct ApiKeyExpiryTrackingData {
     pub key_id: String,
     pub merchant_id: String,
     pub api_key_expiry: Option<PrimitiveDateTime>,
diff --git a/crates/router/Cargo.toml b/crates/router/Cargo.toml
index 2f3dfa82831..88e9c88b4fc 100644
--- a/crates/router/Cargo.toml
+++ b/crates/router/Cargo.toml
@@ -13,7 +13,7 @@ default = ["kv_store", "stripe", "oltp", "olap", "backwards_compatibility", "acc
 aws_s3 = ["external_services/aws_s3"]
 aws_kms = ["external_services/aws_kms"]
 hashicorp-vault = ["external_services/hashicorp-vault"]
-email = ["external_services/email", "olap"]
+email = ["external_services/email", "scheduler/email", "olap"]
 frm = []
 stripe = ["dep:serde_qs"]
 release = ["aws_kms", "stripe", "aws_s3", "email", "backwards_compatibility", "business_profile_routing", "accounts_cache", "kv_store", "connector_choice_mca_id", "profile_specific_fallback_routing", "vergen", "recon"]
diff --git a/crates/router/src/bin/scheduler.rs b/crates/router/src/bin/scheduler.rs
index 59a108ef5e6..caa69ea1394 100644
--- a/crates/router/src/bin/scheduler.rs
+++ b/crates/router/src/bin/scheduler.rs
@@ -216,6 +216,8 @@ pub enum PTRunner {
     PaymentsSyncWorkflow,
     RefundWorkflowRouter,
     DeleteTokenizeDataWorkflow,
+    #[cfg(feature = "email")]
+    ApiKeyExpiryWorkflow,
 }
 
 #[derive(Debug, Copy, Clone)]
@@ -240,6 +242,10 @@ impl ProcessTrackerWorkflows<routes::AppState> for WorkflowRunner {
             Some(PTRunner::DeleteTokenizeDataWorkflow) => {
                 Box::new(workflows::tokenized_data::DeleteTokenizeDataWorkflow)
             }
+            #[cfg(feature = "email")]
+            Some(PTRunner::ApiKeyExpiryWorkflow) => {
+                Box::new(workflows::api_key_expiry::ApiKeyExpiryWorkflow)
+            }
             _ => Err(ProcessTrackerError::UnexpectedFlow)?,
         };
         let app_state = &state.clone();
diff --git a/crates/router/src/core/api_keys.rs b/crates/router/src/core/api_keys.rs
index b694d1291a8..39212ab3814 100644
--- a/crates/router/src/core/api_keys.rs
+++ b/crates/router/src/core/api_keys.rs
@@ -253,7 +253,7 @@ pub async fn add_api_key_expiry_task(
         }
     }
 
-    let api_key_expiry_tracker = &storage::ApiKeyExpiryWorkflow {
+    let api_key_expiry_tracker = &storage::ApiKeyExpiryTrackingData {
         key_id: api_key.key_id.clone(),
         merchant_id: api_key.merchant_id.clone(),
         // We need API key expiry too, because we need to decide on the schedule_time in
@@ -427,7 +427,7 @@ pub async fn update_api_key_expiry_task(
 
     let task_ids = vec![task_id.clone()];
 
-    let updated_tracking_data = &storage::ApiKeyExpiryWorkflow {
+    let updated_tracking_data = &storage::ApiKeyExpiryTrackingData {
         key_id: api_key.key_id.clone(),
         merchant_id: api_key.merchant_id.clone(),
         api_key_expiry: api_key.expires_at,
diff --git a/crates/router/src/services/email/assets/api_key_expiry_reminder.html b/crates/router/src/services/email/assets/api_key_expiry_reminder.html
new file mode 100644
index 00000000000..1865ae38141
--- /dev/null
+++ b/crates/router/src/services/email/assets/api_key_expiry_reminder.html
@@ -0,0 +1,203 @@
+<meta content="text/html; charset=UTF-8" http-equiv="Content-Type" />
+<title>API Key Expiry Notice</title>
+<body style="background-color: #ececec">
+  <style>
+    .apple-footer a {{
+      text-decoration: none !important;
+      color: #999 !important;
+      border: none !important;
+    }}
+    .apple-email a {{
+      text-decoration: none !important;
+      color: #448bff !important;
+      border: none !important;
+    }}
+  </style>
+  <div
+    id="wrapper"
+    style="
+      background-color: none;
+      margin: 0 auto;
+      text-align: center;
+      width: 60%;
+      -premailer-height: 200;
+    "
+  >
+    <table
+      align="center"
+      class="main-table"
+      style="
+        -premailer-cellpadding: 0;
+        -premailer-cellspacing: 0;
+        background-color: #fff;
+        border: 0;
+        border-top: 5px solid #0165ef;
+        margin: 0 auto;
+        mso-table-lspace: 0;
+        mso-table-rspace: 0;
+        padding: 0 40;
+        text-align: center;
+        width: 100%;
+      "
+      bgcolor="#ffffff"
+      cellpadding="0"
+      cellspacing="0"
+    >
+      <tr>
+        <td
+          class="spacer-lg"
+          style="
+            -premailer-height: 75;
+            -premailer-width: 100%;
+            line-height: 30px;
+            margin: 0 auto;
+            padding: 0;
+          "
+          height="75"
+          width="100%"
+        ></td>
+      </tr>
+      <tr>
+        <td
+          class="spacer-lg"
+          style="
+            -premailer-height: 75;
+            -premailer-width: 100%;
+            line-height: 30px;
+            margin: 0 auto;
+            padding: 0;
+          "
+          height="25"
+          width="100%"
+        ></td>
+      </tr>
+      <tr>
+        <td
+          class="spacer-lg"
+          style="
+            -premailer-height: 75;
+            -premailer-width: 100%;
+            line-height: 30px;
+            margin: 0 auto;
+            padding: 0;
+          "
+          height="50"
+          width="100%"
+        ></td>
+      </tr>
+      <tr>
+        <td
+          class="headline"
+          style="
+            color: #444;
+            font-family: Roboto, Helvetica, Arial, san-serif;
+            font-size: 30px;
+            font-weight: 100;
+            line-height: 36px;
+            margin: 0 auto;
+            padding: 0;
+            text-align: center;
+          "
+          align="center"
+        >
+        <p style="font-size: 18px">Dear Merchant,</p>
+        <span style="font-size: 18px">
+          It has come to our attention that your API key will expire in {expires_in} days. To ensure uninterrupted
+          access to our platform and continued smooth operation of your services, we kindly request that you take the
+          necessary actions as soon as possible.
+        </span>
+        </td>
+      </tr>
+      <tr>
+        <td
+          class="spacer-sm"
+          style="
+            -premailer-height: 20;
+            -premailer-width: 80%;
+            line-height: 10px;
+            margin: 0 auto;
+            padding: 0;
+          "
+          width="100%"
+        ></td>
+      </tr>
+      
+      <tr>
+        <td
+          class="spacer-sm"
+          style="
+            -premailer-height: 20;
+            -premailer-width: 100%;
+            line-height: 10px;
+            margin: 0 auto;
+            padding: 0;
+          "
+          height="20"
+          width="100%"
+        ></td>
+      </tr>
+
+      <tr>
+        <td
+          class="spacer-lg"
+          style="
+            -premailer-height: 75;
+            -premailer-width: 100%;
+            line-height: 30px;
+            margin: 0 auto;
+            padding: 0;
+          "
+          height="75"
+          width="100%"
+        ></td>
+      </tr>
+      <tr>
+        <td
+          class="headline"
+          style="
+            color: #444;
+            font-family: Roboto, Helvetica, Arial, san-serif;
+            font-size: 18px;
+            font-weight: 100;
+            line-height: 36px;
+            margin: 0 auto;
+            padding: 0;
+            text-align: center;
+          "
+          align="center"
+        >
+          Thanks,<br />
+          Team Hyperswitch
+        </td>
+      </tr>
+      <tr>
+        <td
+          class="spacer-lg"
+          style="
+            -premailer-height: 75;
+            -premailer-width: 100%;
+            line-height: 30px;
+            margin: 0 auto;
+            padding: 0;
+          "
+          height="75"
+          width="100%"
+        ></td>
+      </tr>
+      <tr>
+        <td
+          class="spacer-lg"
+          style="
+            -premailer-height: 75;
+            -premailer-width: 100%;
+            line-height: 30px;
+            margin: 0 auto;
+            padding: 0;
+          "
+          height="75"
+          width="100%"
+        ></td>
+      </tr>
+    </table>
+  </div>
+</body>
diff --git a/crates/router/src/services/email/types.rs b/crates/router/src/services/email/types.rs
index 6ad1a0eb99a..389979d5723 100644
--- a/crates/router/src/services/email/types.rs
+++ b/crates/router/src/services/email/types.rs
@@ -44,6 +44,9 @@ pub enum EmailBody {
         user_name: String,
         user_email: String,
     },
+    ApiKeyExpiryReminder {
+        expires_in: u8,
+    },
 }
 
 pub mod html {
@@ -113,6 +116,10 @@ Email         : {user_email}
 
 (note: This is an auto generated email. Use merchant email for any further communications)",
             ),
+            EmailBody::ApiKeyExpiryReminder { expires_in } => format!(
+                include_str!("assets/api_key_expiry_reminder.html"),
+                expires_in = expires_in,
+            ),
         }
     }
 }
@@ -381,3 +388,26 @@ impl EmailData for ProFeatureRequest {
         })
     }
 }
+
+pub struct ApiKeyExpiryReminder {
+    pub recipient_email: domain::UserEmail,
+    pub subject: &'static str,
+    pub expires_in: u8,
+}
+
+#[async_trait::async_trait]
+impl EmailData for ApiKeyExpiryReminder {
+    async fn get_email_data(&self) -> CustomResult<EmailContents, EmailError> {
+        let recipient = self.recipient_email.clone().into_inner();
+
+        let body = html::get_html_body(EmailBody::ApiKeyExpiryReminder {
+            expires_in: self.expires_in,
+        });
+
+        Ok(EmailContents {
+            subject: self.subject.to_string(),
+            body: external_services::email::IntermediateString::new(body),
+            recipient,
+        })
+    }
+}
diff --git a/crates/router/src/types/storage/api_keys.rs b/crates/router/src/types/storage/api_keys.rs
index 74c503d3c76..b1051c3d19c 100644
--- a/crates/router/src/types/storage/api_keys.rs
+++ b/crates/router/src/types/storage/api_keys.rs
@@ -1,3 +1,3 @@
 #[cfg(feature = "email")]
-pub use diesel_models::api_keys::ApiKeyExpiryWorkflow;
+pub use diesel_models::api_keys::ApiKeyExpiryTrackingData;
 pub use diesel_models::api_keys::{ApiKey, ApiKeyNew, ApiKeyUpdate, HashedApiKey};
diff --git a/crates/router/src/workflows.rs b/crates/router/src/workflows.rs
index b036193bb27..deb5bf785fa 100644
--- a/crates/router/src/workflows.rs
+++ b/crates/router/src/workflows.rs
@@ -1,3 +1,5 @@
+#[cfg(feature = "email")]
+pub mod api_key_expiry;
 pub mod payment_sync;
 pub mod refund_router;
 pub mod tokenized_data;
diff --git a/crates/router/src/workflows/api_key_expiry.rs b/crates/router/src/workflows/api_key_expiry.rs
index eb3c1d9c1ce..b9830c4ebc5 100644
--- a/crates/router/src/workflows/api_key_expiry.rs
+++ b/crates/router/src/workflows/api_key_expiry.rs
@@ -1,30 +1,35 @@
-use common_utils::ext_traits::ValueExt;
-use diesel_models::enums::{self as storage_enums};
+use common_utils::{errors::ValidationError, ext_traits::ValueExt};
+use diesel_models::{enums as storage_enums, ApiKeyExpiryTrackingData};
+use router_env::logger;
+use scheduler::{workflows::ProcessTrackerWorkflow, SchedulerAppState};
 
-use super::{ApiKeyExpiryWorkflow, ProcessTrackerWorkflow};
 use crate::{
     errors,
     logger::error,
-    routes::AppState,
+    routes::{metrics, AppState},
+    services::email::types::ApiKeyExpiryReminder,
     types::{
         api,
+        domain::UserEmail,
         storage::{self, ProcessTrackerExt},
     },
     utils::OptionExt,
 };
 
+pub struct ApiKeyExpiryWorkflow;
+
 #[async_trait::async_trait]
-impl ProcessTrackerWorkflow for ApiKeyExpiryWorkflow {
+impl ProcessTrackerWorkflow<AppState> for ApiKeyExpiryWorkflow {
     async fn execute_workflow<'a>(
         &'a self,
         state: &'a AppState,
         process: storage::ProcessTracker,
     ) -> Result<(), errors::ProcessTrackerError> {
         let db = &*state.store;
-        let tracking_data: storage::ApiKeyExpiryWorkflow = process
+        let tracking_data: ApiKeyExpiryTrackingData = process
             .tracking_data
             .clone()
-            .parse_value("ApiKeyExpiryWorkflow")?;
+            .parse_value("ApiKeyExpiryTrackingData")?;
 
         let key_store = state
             .store
@@ -41,7 +46,13 @@ impl ProcessTrackerWorkflow for ApiKeyExpiryWorkflow {
         let email_id = merchant_account
             .merchant_details
             .parse_value::<api::MerchantDetails>("MerchantDetails")?
-            .primary_email;
+            .primary_email
+            .ok_or(errors::ProcessTrackerError::EValidationError(
+                ValidationError::MissingRequiredField {
+                    field_name: "email".to_string(),
+                }
+                .into(),
+            ))?;
 
         let task_id = process.id.clone();
 
@@ -53,28 +64,26 @@ impl ProcessTrackerWorkflow for ApiKeyExpiryWorkflow {
                 usize::try_from(retry_count)
                     .map_err(|_| errors::ProcessTrackerError::TypeConversionError)?,
             )
-            .ok_or(errors::ProcessTrackerError::EApiErrorResponse(
-                errors::ApiErrorResponse::InvalidDataValue {
-                    field_name: "index",
-                }
-                .into(),
-            ))?;
+            .ok_or(errors::ProcessTrackerError::EApiErrorResponse)?;
+
+        let email_contents = ApiKeyExpiryReminder {
+            recipient_email: UserEmail::from_pii_email(email_id).map_err(|err| {
+                logger::error!(%err,"Failed to convert recipient's email to UserEmail from pii::Email");
+                errors::ProcessTrackerError::EApiErrorResponse
+            })?,
+            subject: "API Key Expiry Notice",
+            expires_in: *expires_in,
+        };
 
         state
             .email_client
             .clone()
-            .send_email(
-                email_id.ok_or_else(|| errors::ProcessTrackerError::MissingRequiredField)?,
-                "API Key Expiry Notice".to_string(),
-                format!("Dear Merchant,\n
-It has come to our attention that your API key will expire in {expires_in} days. To ensure uninterrupted access to our platform and continued smooth operation of your services, we kindly request that you take the necessary actions as soon as possible.\n\n
-Thanks,\n
-Team Hyperswitch"),
+            .compose_and_send_email(
+                Box::new(email_contents),
+                state.conf.proxy.https_url.as_ref(),
             )
             .await
-            .map_err(|_| errors::ProcessTrackerError::FlowExecutionError {
-                flow: "ApiKeyExpiryWorkflow",
-            })?;
+            .map_err(errors::ProcessTrackerError::EEmailError)?;
 
         // If all the mails have been sent, then retry_count would be equal to length of the expiry_reminder_days vector
         if retry_count
@@ -82,7 +91,7 @@ Team Hyperswitch"),
                 .map_err(|_| errors::ProcessTrackerError::TypeConversionError)?
         {
             process
-                .finish_with_status(db, format!("COMPLETED_BY_PT_{task_id}"))
+                .finish_with_status(state.get_db().as_scheduler(), "COMPLETED_BY_PT".to_string())
                 .await?
         }
         // If tasks are remaining that has to be scheduled
@@ -93,12 +102,7 @@ Team Hyperswitch"),
                     usize::try_from(retry_count + 1)
                         .map_err(|_| errors::ProcessTrackerError::TypeConversionError)?,
                 )
-                .ok_or(errors::ProcessTrackerError::EApiErrorResponse(
-                    errors::ApiErrorResponse::InvalidDataValue {
-                        field_name: "index",
-                    }
-                    .into(),
-                ))?;
+                .ok_or(errors::ProcessTrackerError::EApiErrorResponse)?;
 
             let updated_schedule_time = tracking_data.api_key_expiry.map(|api_key_expiry| {
                 api_key_expiry.saturating_sub(time::Duration::days(i64::from(*expiry_reminder_day)))
diff --git a/crates/scheduler/Cargo.toml b/crates/scheduler/Cargo.toml
index b281bc862a2..72eaea3be06 100644
--- a/crates/scheduler/Cargo.toml
+++ b/crates/scheduler/Cargo.toml
@@ -7,6 +7,7 @@ edition = "2021"
 default = ["kv_store", "olap"]
 olap = ["storage_impl/olap"]
 kv_store = []
+email = ["external_services/email"]
 
 [dependencies]
 # Third party crates
diff --git a/crates/scheduler/src/errors.rs b/crates/scheduler/src/errors.rs
index 481fae07937..78a0bdee624 100644
--- a/crates/scheduler/src/errors.rs
+++ b/crates/scheduler/src/errors.rs
@@ -1,4 +1,6 @@
 pub use common_utils::errors::{ParsingError, ValidationError};
+#[cfg(feature = "email")]
+use external_services::email::EmailError;
 pub use redis_interface::errors::RedisError;
 pub use storage_impl::errors::ApplicationError;
 use storage_impl::errors::StorageError;
@@ -51,6 +53,9 @@ pub enum ProcessTrackerError {
     EParsingError(error_stack::Report<ParsingError>),
     #[error("Validation Error Received: {0}")]
     EValidationError(error_stack::Report<ValidationError>),
+    #[cfg(feature = "email")]
+    #[error("Received Error EmailError: {0}")]
+    EEmailError(error_stack::Report<EmailError>),
     #[error("Type Conversion error")]
     TypeConversionError,
 }
@@ -111,3 +116,9 @@ error_to_process_tracker_error!(
     error_stack::Report<ValidationError>,
     ProcessTrackerError::EValidationError(error_stack::Report<ValidationError>)
 );
+
+#[cfg(feature = "email")]
+error_to_process_tracker_error!(
+    error_stack::Report<EmailError>,
+    ProcessTrackerError::EEmailError(error_stack::Report<EmailError>)
+);