Issue:
Bug: feat: Email blacklist





Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [X] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
- Adding blacklisting for email tokens, this will prevent use of same email token to be used twice.
- Change Password & Reset Password logs out user from all open dashboard sessions.
<!-- Describe your changes in detail -->


### 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
Prevention of use of same email link to be used twice.
Forcing Users to re-enter password if password is changed.
<!--
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?
With Dashboard Frontend.
Following links should not work twice:
- Magic Link
To generate magic link email use dashboard frontend or below curl.
```sh
curl --location --request POST '<URL>/user/connect_account' \
--header 'Content-Type: application/json' \
--data-raw '{
    "email": ""
}'
```
- Reset Password
To generate reset password email use dashboard frontend or below curl.
```sh
curl --location --request POST '<URL>/user/forgot_password' \
--header 'Content-Type: application/json' \
--data-raw '{
    "email": ""
}'
```

All sessions should be logged out after change password or reset password.

Expected error response for the use of email token twice
```json
{
    "error": {
        "type": "invalid_request",
        "message": "Invalid or expired link",
        "code": "UR_04"
    }
}
```
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->


## 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:
With Dashboard Frontend.
Following links should not work twice:
- Magic Link
To generate magic link email use dashboard frontend or below curl.
```sh
curl --location --request POST '<URL>/user/connect_account' \
--header 'Content-Type: application/json' \
--data-raw '{
    "email": ""
}'
```
- Reset Password
To generate reset password email use dashboard frontend or below curl.
```sh
curl --location --request POST '<URL>/user/forgot_password' \
--header 'Content-Type: application/json' \
--data-raw '{
    "email": ""
}'
```

All sessions should be logged out after change password or reset password.

Expected error response for the use of email token twice
```json
{
    "error": {
        "type": "invalid_request",
        "message": "Invalid or expired link",
        "code": "UR_04"
    }
}
```


Patch:
diff --git a/crates/router/src/core/user.rs b/crates/router/src/core/user.rs
index 002452a8a35..54fbecc64f3 100644
--- a/crates/router/src/core/user.rs
+++ b/crates/router/src/core/user.rs
@@ -311,6 +311,10 @@ pub async fn change_password(
         .await
         .change_context(UserErrors::InternalServerError)?;
 
+    let _ = auth::blacklist::insert_user_in_blacklist(&state, user.get_user_id())
+        .await
+        .map_err(|e| logger::error!(?e));
+
     #[cfg(not(feature = "email"))]
     {
         state
@@ -372,10 +376,12 @@ pub async fn reset_password(
     state: AppState,
     request: user_api::ResetPasswordRequest,
 ) -> UserResponse<()> {
-    let token =
-        auth::decode_jwt::<email_types::EmailToken>(request.token.expose().as_str(), &state)
-            .await
-            .change_context(UserErrors::LinkInvalid)?;
+    let token = request.token.expose();
+    let email_token = auth::decode_jwt::<email_types::EmailToken>(&token, &state)
+        .await
+        .change_context(UserErrors::LinkInvalid)?;
+
+    auth::blacklist::check_email_token_in_blacklist(&state, &token).await?;
 
     let password = domain::UserPassword::new(request.password)?;
 
@@ -384,7 +390,7 @@ pub async fn reset_password(
     let user = state
         .store
         .update_user_by_email(
-            token.get_email(),
+            email_token.get_email(),
             storage_user::UserUpdate::AccountUpdate {
                 name: None,
                 password: Some(hash_password),
@@ -395,7 +401,7 @@ pub async fn reset_password(
         .await
         .change_context(UserErrors::InternalServerError)?;
 
-    if let Some(inviter_merchant_id) = token.get_merchant_id() {
+    if let Some(inviter_merchant_id) = email_token.get_merchant_id() {
         let update_status_result = state
             .store
             .update_user_role_by_user_id_merchant_id(
@@ -403,13 +409,20 @@ pub async fn reset_password(
                 inviter_merchant_id,
                 UserRoleUpdate::UpdateStatus {
                     status: UserStatus::Active,
-                    modified_by: user.user_id,
+                    modified_by: user.user_id.clone(),
                 },
             )
             .await;
         logger::info!(?update_status_result);
     }
 
+    let _ = auth::blacklist::insert_email_token_in_blacklist(&state, &token)
+        .await
+        .map_err(|e| logger::error!(?e));
+    let _ = auth::blacklist::insert_user_in_blacklist(&state, &user.user_id)
+        .await
+        .map_err(|e| logger::error!(?e));
+
     Ok(ApplicationResponse::StatusOk)
 }
 
@@ -454,7 +467,13 @@ pub async fn invite_user(
                 merchant_id: user_from_token.merchant_id,
                 role_id: request.role_id,
                 org_id: user_from_token.org_id,
-                status: UserStatus::Active,
+                status: {
+                    if cfg!(feature = "email") {
+                        UserStatus::InvitationSent
+                    } else {
+                        UserStatus::Active
+                    }
+                },
                 created_by: user_from_token.user_id.clone(),
                 last_modified_by: user_from_token.user_id,
                 created_at: now,
@@ -1050,12 +1069,14 @@ pub async fn verify_email_without_invite_checks(
     state: AppState,
     req: user_api::VerifyEmailRequest,
 ) -> UserResponse<user_api::DashboardEntryResponse> {
-    let token = auth::decode_jwt::<email_types::EmailToken>(&req.token.clone().expose(), &state)
+    let token = req.token.clone().expose();
+    let email_token = auth::decode_jwt::<email_types::EmailToken>(&token, &state)
         .await
         .change_context(UserErrors::LinkInvalid)?;
+    auth::blacklist::check_email_token_in_blacklist(&state, &token).await?;
     let user = state
         .store
-        .find_user_by_email(token.get_email())
+        .find_user_by_email(email_token.get_email())
         .await
         .change_context(UserErrors::InternalServerError)?;
     let user = state
@@ -1065,6 +1086,9 @@ pub async fn verify_email_without_invite_checks(
         .change_context(UserErrors::InternalServerError)?;
     let user_from_db: domain::UserFromStorage = user.into();
     let user_role = user_from_db.get_role_from_db(state.clone()).await?;
+    let _ = auth::blacklist::insert_email_token_in_blacklist(&state, &token)
+        .await
+        .map_err(|e| logger::error!(?e));
     let token = utils::user::generate_jwt_auth_token(&state, &user_from_db, &user_role).await?;
 
     Ok(ApplicationResponse::Json(
@@ -1077,13 +1101,16 @@ pub async fn verify_email(
     state: AppState,
     req: user_api::VerifyEmailRequest,
 ) -> UserResponse<user_api::SignInResponse> {
-    let token = auth::decode_jwt::<email_types::EmailToken>(&req.token.clone().expose(), &state)
+    let token = req.token.clone().expose();
+    let email_token = auth::decode_jwt::<email_types::EmailToken>(&token, &state)
         .await
         .change_context(UserErrors::LinkInvalid)?;
 
+    auth::blacklist::check_email_token_in_blacklist(&state, &token).await?;
+
     let user = state
         .store
-        .find_user_by_email(token.get_email())
+        .find_user_by_email(email_token.get_email())
         .await
         .change_context(UserErrors::InternalServerError)?;
 
@@ -1115,6 +1142,10 @@ pub async fn verify_email(
             .await?
         };
 
+    let _ = auth::blacklist::insert_email_token_in_blacklist(&state, &token)
+        .await
+        .map_err(|e| logger::error!(?e));
+
     Ok(ApplicationResponse::Json(
         signin_strategy.get_signin_response(&state).await?,
     ))
diff --git a/crates/router/src/services/authentication/blacklist.rs b/crates/router/src/services/authentication/blacklist.rs
index 6fab28433b4..325ef29bad3 100644
--- a/crates/router/src/services/authentication/blacklist.rs
+++ b/crates/router/src/services/authentication/blacklist.rs
@@ -5,6 +5,8 @@ use common_utils::date_time;
 use error_stack::{IntoReport, ResultExt};
 use redis_interface::RedisConnectionPool;
 
+#[cfg(feature = "email")]
+use crate::consts::{EMAIL_TOKEN_BLACKLIST_PREFIX, EMAIL_TOKEN_TIME_IN_SECS};
 use crate::{
     consts::{JWT_TOKEN_TIME_IN_SECS, USER_BLACKLIST_PREFIX},
     core::errors::{ApiErrorResponse, RouterResult},
@@ -47,6 +49,33 @@ pub async fn check_user_in_blacklist<A: AppStateInfo>(
         .map(|timestamp| timestamp.map_or(false, |timestamp| timestamp > token_issued_at))
 }
 
+#[cfg(feature = "email")]
+pub async fn insert_email_token_in_blacklist(state: &AppState, token: &str) -> UserResult<()> {
+    let redis_conn = get_redis_connection(state).change_context(UserErrors::InternalServerError)?;
+    let blacklist_key = format!("{}{token}", EMAIL_TOKEN_BLACKLIST_PREFIX);
+    let expiry =
+        expiry_to_i64(EMAIL_TOKEN_TIME_IN_SECS).change_context(UserErrors::InternalServerError)?;
+    redis_conn
+        .set_key_with_expiry(blacklist_key.as_str(), true, expiry)
+        .await
+        .change_context(UserErrors::InternalServerError)
+}
+
+#[cfg(feature = "email")]
+pub async fn check_email_token_in_blacklist(state: &AppState, token: &str) -> UserResult<()> {
+    let redis_conn = get_redis_connection(state).change_context(UserErrors::InternalServerError)?;
+    let blacklist_key = format!("{}{token}", EMAIL_TOKEN_BLACKLIST_PREFIX);
+    let key_exists = redis_conn
+        .exists::<()>(blacklist_key.as_str())
+        .await
+        .change_context(UserErrors::InternalServerError)?;
+
+    if key_exists {
+        return Err(UserErrors::LinkInvalid.into());
+    }
+    Ok(())
+}
+
 fn get_redis_connection<A: AppStateInfo>(state: &A) -> RouterResult<Arc<RedisConnectionPool>> {
     state
         .store()