Issue:
Bug: feat: Create a decision starter API for email flows



As we have split the API flows, there is no API to trigger the email specific flows like Verify email, Accept invite from email, etc...

For this API to work correctly, email token should have the information about the origin.

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
<!-- Describe your changes in detail -->
- Create a new API `from_email` which will take email token and give the token for the next flow.
- Add `flow` field in email token which will help `from_email` API to decide the next flow. 

### Additional Changes

- [x] 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).
-->
Closes #4532 

## 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)?
-->
```
curl --location 'http://localhost:8080/user/from_email' \
--header 'Content-Type: application/json' \
--data '{
    "token": "email token"
}'
```
Response should be like this
```
{
    "token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiMDYxNjE1YjAtZjI5Yi00NWIyLWJmNmItODczNTYyYWU1ODhlIiwicHVycG9zZSI6InRvdHAiLCJvcmlnaW4iOiJtYWdpY19saW5rIiwiZXhwIjoxNzE0ODExOTAwfQ.CDErXJ89a1o3ROyAKvm6TFW1drHDnlcqg43OoBQjY6A",
    "token_type": "totp"
}
```

## 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:

```
curl --location 'http://localhost:8080/user/from_email' \
--header 'Content-Type: application/json' \
--data '{
    "token": "email token"
}'
```
Response should be like this
```
{
    "token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiMDYxNjE1YjAtZjI5Yi00NWIyLWJmNmItODczNTYyYWU1ODhlIiwicHVycG9zZSI6InRvdHAiLCJvcmlnaW4iOiJtYWdpY19saW5rIiwiZXhwIjoxNzE0ODExOTAwfQ.CDErXJ89a1o3ROyAKvm6TFW1drHDnlcqg43OoBQjY6A",
    "token_type": "totp"
}
```

Patch:
diff --git a/crates/api_models/src/events/user.rs b/crates/api_models/src/events/user.rs
index 594b60b5816..14676656d0f 100644
--- a/crates/api_models/src/events/user.rs
+++ b/crates/api_models/src/events/user.rs
@@ -15,8 +15,8 @@ use crate::user::{
     GetUserDetailsResponse, GetUserRoleDetailsRequest, GetUserRoleDetailsResponse,
     InviteUserRequest, ListUsersResponse, ReInviteUserRequest, ResetPasswordRequest,
     SendVerifyEmailRequest, SignInResponse, SignInWithTokenResponse, SignUpRequest,
-    SignUpWithMerchantIdRequest, SwitchMerchantIdRequest, UpdateUserAccountDetailsRequest,
-    UserMerchantCreate, VerifyEmailRequest,
+    SignUpWithMerchantIdRequest, SwitchMerchantIdRequest, TokenResponse,
+    UpdateUserAccountDetailsRequest, UserFromEmailRequest, UserMerchantCreate, VerifyEmailRequest,
 };
 
 impl ApiEventMetric for DashboardEntryResponse {
@@ -64,7 +64,9 @@ common_utils::impl_misc_api_event_type!(
     GetUserDetailsResponse,
     SignInWithTokenResponse,
     GetUserRoleDetailsRequest,
-    GetUserRoleDetailsResponse
+    GetUserRoleDetailsResponse,
+    TokenResponse,
+    UserFromEmailRequest
 );
 
 #[cfg(feature = "dummy_connector")]
diff --git a/crates/api_models/src/user.rs b/crates/api_models/src/user.rs
index b4d53a92c1a..a6d5cf36358 100644
--- a/crates/api_models/src/user.rs
+++ b/crates/api_models/src/user.rs
@@ -231,3 +231,8 @@ pub enum SignInWithTokenResponse {
     Token(TokenResponse),
     SignInResponse(SignInResponse),
 }
+
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
+pub struct UserFromEmailRequest {
+    pub token: Secret<String>,
+}
diff --git a/crates/router/src/core/user.rs b/crates/router/src/core/user.rs
index 0ae1b162e0e..58a6452c9bb 100644
--- a/crates/router/src/core/user.rs
+++ b/crates/router/src/core/user.rs
@@ -183,21 +183,7 @@ pub async fn signin_token_only_flow(
     let next_flow =
         domain::NextFlow::from_origin(domain::Origin::SignIn, user_from_db.clone(), &state).await?;
 
-    let token = match next_flow.get_flow() {
-        domain::UserFlow::SPTFlow(spt_flow) => spt_flow.generate_spt(&state, &next_flow).await,
-        domain::UserFlow::JWTFlow(jwt_flow) => {
-            #[cfg(feature = "email")]
-            {
-                user_from_db.get_verification_days_left(&state)?;
-            }
-
-            let user_role = user_from_db
-                .get_preferred_or_active_user_role_from_db(&state)
-                .await
-                .to_not_found_response(UserErrors::InternalServerError)?;
-            jwt_flow.generate_jwt(&state, &next_flow, &user_role).await
-        }
-    }?;
+    let token = next_flow.get_token(&state).await?;
 
     let response = user_api::SignInWithTokenResponse::Token(user_api::TokenResponse {
         token: token.clone(),
@@ -1323,3 +1309,38 @@ pub async fn update_user_details(
 
     Ok(ApplicationResponse::StatusOk)
 }
+
+#[cfg(feature = "email")]
+pub async fn user_from_email(
+    state: AppState,
+    req: user_api::UserFromEmailRequest,
+) -> UserResponse<user_api::TokenResponse> {
+    let token = req.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 user_from_db: domain::UserFromStorage = state
+        .store
+        .find_user_by_email(
+            &email_token
+                .get_email()
+                .change_context(UserErrors::InternalServerError)?,
+        )
+        .await
+        .change_context(UserErrors::InternalServerError)?
+        .into();
+
+    let next_flow =
+        domain::NextFlow::from_origin(email_token.get_flow(), user_from_db.clone(), &state).await?;
+
+    let token = next_flow.get_token(&state).await?;
+
+    let response = user_api::TokenResponse {
+        token: token.clone(),
+        token_type: next_flow.get_flow().into(),
+    };
+    auth::cookies::set_cookie_response(response, token)
+}
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 19a632bf895..0f3673ee206 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -1201,6 +1201,7 @@ impl User {
         #[cfg(feature = "email")]
         {
             route = route
+                .service(web::resource("/from_email").route(web::post().to(user_from_email)))
                 .service(
                     web::resource("/connect_account").route(web::post().to(user_connect_account)),
                 )
diff --git a/crates/router/src/routes/lock_utils.rs b/crates/router/src/routes/lock_utils.rs
index faef85f3901..4d1121a8eab 100644
--- a/crates/router/src/routes/lock_utils.rs
+++ b/crates/router/src/routes/lock_utils.rs
@@ -219,7 +219,8 @@ impl From<Flow> for ApiIdentifier {
             | Flow::DeleteUserRole
             | Flow::TransferOrgOwnership
             | Flow::CreateRole
-            | Flow::UpdateRole => Self::UserRole,
+            | Flow::UpdateRole
+            | Flow::UserFromEmail => Self::UserRole,
 
             Flow::GetActionUrl | Flow::SyncOnboardingStatus | Flow::ResetTrackingId => {
                 Self::ConnectorOnboarding
diff --git a/crates/router/src/routes/user.rs b/crates/router/src/routes/user.rs
index 4d841913bc0..ede6edbceb8 100644
--- a/crates/router/src/routes/user.rs
+++ b/crates/router/src/routes/user.rs
@@ -512,3 +512,22 @@ pub async fn update_user_account_details(
     ))
     .await
 }
+
+#[cfg(feature = "email")]
+pub async fn user_from_email(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    json_payload: web::Json<user_api::UserFromEmailRequest>,
+) -> HttpResponse {
+    let flow = Flow::UserFromEmail;
+    Box::pin(api::server_wrap(
+        flow,
+        state.clone(),
+        &req,
+        json_payload.into_inner(),
+        |state, _: (), req_body, _| user_core::user_from_email(state, req_body),
+        &auth::NoAuth,
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
diff --git a/crates/router/src/services/email/types.rs b/crates/router/src/services/email/types.rs
index 323f98a56df..ee51d976b40 100644
--- a/crates/router/src/services/email/types.rs
+++ b/crates/router/src/services/email/types.rs
@@ -151,6 +151,7 @@ Email         : {user_email}
 pub struct EmailToken {
     email: String,
     merchant_id: Option<String>,
+    flow: domain::Origin,
     exp: u64,
 }
 
@@ -158,6 +159,7 @@ impl EmailToken {
     pub async fn new_token(
         email: domain::UserEmail,
         merchant_id: Option<String>,
+        flow: domain::Origin,
         settings: &configs::Settings,
     ) -> CustomResult<String, UserErrors> {
         let expiration_duration = std::time::Duration::from_secs(consts::EMAIL_TOKEN_TIME_IN_SECS);
@@ -165,6 +167,7 @@ impl EmailToken {
         let token_payload = Self {
             email: email.get_secret().expose(),
             merchant_id,
+            flow,
             exp,
         };
         jwt::generate_jwt(&token_payload, settings).await
@@ -177,6 +180,10 @@ impl EmailToken {
     pub fn get_merchant_id(&self) -> Option<&str> {
         self.merchant_id.as_deref()
     }
+
+    pub fn get_flow(&self) -> domain::Origin {
+        self.flow.clone()
+    }
 }
 
 pub fn get_link_with_token(
@@ -197,9 +204,14 @@ pub struct VerifyEmail {
 #[async_trait::async_trait]
 impl EmailData for VerifyEmail {
     async fn get_email_data(&self) -> CustomResult<EmailContents, EmailError> {
-        let token = EmailToken::new_token(self.recipient_email.clone(), None, &self.settings)
-            .await
-            .change_context(EmailError::TokenGenerationFailure)?;
+        let token = EmailToken::new_token(
+            self.recipient_email.clone(),
+            None,
+            domain::Origin::VerifyEmail,
+            &self.settings,
+        )
+        .await
+        .change_context(EmailError::TokenGenerationFailure)?;
 
         let verify_email_link =
             get_link_with_token(&self.settings.email.base_url, token, "verify_email");
@@ -226,9 +238,14 @@ pub struct ResetPassword {
 #[async_trait::async_trait]
 impl EmailData for ResetPassword {
     async fn get_email_data(&self) -> CustomResult<EmailContents, EmailError> {
-        let token = EmailToken::new_token(self.recipient_email.clone(), None, &self.settings)
-            .await
-            .change_context(EmailError::TokenGenerationFailure)?;
+        let token = EmailToken::new_token(
+            self.recipient_email.clone(),
+            None,
+            domain::Origin::ResetPassword,
+            &self.settings,
+        )
+        .await
+        .change_context(EmailError::TokenGenerationFailure)?;
 
         let reset_password_link =
             get_link_with_token(&self.settings.email.base_url, token, "set_password");
@@ -256,9 +273,14 @@ pub struct MagicLink {
 #[async_trait::async_trait]
 impl EmailData for MagicLink {
     async fn get_email_data(&self) -> CustomResult<EmailContents, EmailError> {
-        let token = EmailToken::new_token(self.recipient_email.clone(), None, &self.settings)
-            .await
-            .change_context(EmailError::TokenGenerationFailure)?;
+        let token = EmailToken::new_token(
+            self.recipient_email.clone(),
+            None,
+            domain::Origin::MagicLink,
+            &self.settings,
+        )
+        .await
+        .change_context(EmailError::TokenGenerationFailure)?;
 
         let magic_link_login =
             get_link_with_token(&self.settings.email.base_url, token, "verify_email");
@@ -276,6 +298,7 @@ impl EmailData for MagicLink {
     }
 }
 
+// TODO: Deprecate this and use InviteRegisteredUser for new invites
 pub struct InviteUser {
     pub recipient_email: domain::UserEmail,
     pub user_name: domain::UserName,
@@ -290,6 +313,7 @@ impl EmailData for InviteUser {
         let token = EmailToken::new_token(
             self.recipient_email.clone(),
             Some(self.merchant_id.clone()),
+            domain::Origin::ResetPassword,
             &self.settings,
         )
         .await
@@ -310,6 +334,7 @@ impl EmailData for InviteUser {
         })
     }
 }
+
 pub struct InviteRegisteredUser {
     pub recipient_email: domain::UserEmail,
     pub user_name: domain::UserName,
@@ -324,6 +349,7 @@ impl EmailData for InviteRegisteredUser {
         let token = EmailToken::new_token(
             self.recipient_email.clone(),
             Some(self.merchant_id.clone()),
+            domain::Origin::AcceptInvitationFromEmail,
             &self.settings,
         )
         .await
diff --git a/crates/router/src/types/domain/user/decision_manager.rs b/crates/router/src/types/domain/user/decision_manager.rs
index b5aff779106..e460bc74649 100644
--- a/crates/router/src/types/domain/user/decision_manager.rs
+++ b/crates/router/src/types/domain/user/decision_manager.rs
@@ -4,7 +4,7 @@ use masking::Secret;
 
 use super::UserFromStorage;
 use crate::{
-    core::errors::{UserErrors, UserResult},
+    core::errors::{StorageErrorExt, UserErrors, UserResult},
     routes::AppState,
     services::authentication as auth,
 };
@@ -225,6 +225,25 @@ impl NextFlow {
     pub fn get_flow(&self) -> UserFlow {
         self.next_flow
     }
+
+    pub async fn get_token(&self, state: &AppState) -> UserResult<Secret<String>> {
+        match self.next_flow {
+            UserFlow::SPTFlow(spt_flow) => spt_flow.generate_spt(state, self).await,
+            UserFlow::JWTFlow(jwt_flow) => {
+                #[cfg(feature = "email")]
+                {
+                    self.user.get_verification_days_left(state)?;
+                }
+
+                let user_role = self
+                    .user
+                    .get_preferred_or_active_user_role_from_db(state)
+                    .await
+                    .to_not_found_response(UserErrors::InternalServerError)?;
+                jwt_flow.generate_jwt(state, self, &user_role).await
+            }
+        }
+    }
 }
 
 impl From<UserFlow> for TokenPurpose {
diff --git a/crates/router_env/src/logger/types.rs b/crates/router_env/src/logger/types.rs
index 26cb619d74c..78f570f647e 100644
--- a/crates/router_env/src/logger/types.rs
+++ b/crates/router_env/src/logger/types.rs
@@ -394,6 +394,8 @@ pub enum Flow {
     CreateRole,
     /// Update Role
     UpdateRole,
+    /// User email flow start
+    UserFromEmail,
     /// List initial webhook delivery attempts
     WebhookEventInitialDeliveryAttemptList,
     /// List delivery attempts for a webhook event