rust/hyperswitch
1
0
Fork 0
mirror of https://github.com/juspay/hyperswitch.git synced 2025-11-02 12:06:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(users): store and retrieve lineage_context from DB instead of Redis (#7940)

Browse Source 
Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>
Co-authored-by: Gnanasundari24 <118818938+Gnanasundari24@users.noreply.github.com>
This commit is contained in:
Sandeep Kumar
2025-05-13 12:53:10 +05:30
committed by GitHub
parent 40fd473989
commit 6f22a9306c
27 changed files with 186 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/router/src/configs/settings.rs
View File

@ -7,7 +7,7 @@ use std::{
#[cfg(feature = "olap")]
use analytics::{opensearch::OpenSearchConfig, ReportConfig};
use api_models::enums;
use common_utils::{ext_traits::ConfigExt, id_type, types::theme::EmailThemeConfig};
use common_utils::{ext_traits::ConfigExt, id_type, types::user::EmailThemeConfig};
use config::{Environment, File};
use error_stack::ResultExt;
#[cfg(feature = "email")]

3
crates/router/src/consts/user.rs
View File

@ -41,6 +41,3 @@ pub const REDIS_SSO_TTL: i64 = 5 * 60; // 5 minutes
pub const DEFAULT_PROFILE_NAME: &str = "default";
pub const DEFAULT_PRODUCT_TYPE: common_enums::MerchantProductType =
common_enums::MerchantProductType::Orchestration;
pub const LINEAGE_CONTEXT_TIME_EXPIRY_IN_SECS: i64 = 60 * 60 * 24 * 7; // 7 days
pub const LINEAGE_CONTEXT_PREFIX: &str = "LINEAGE_CONTEXT_";

2
crates/router/src/core/recon.rs
View File

@ -1,6 +1,6 @@
use api_models::recon as recon_api;
#[cfg(feature = "email")]
use common_utils::{ext_traits::AsyncExt, types::theme::ThemeLineage};
use common_utils::{ext_traits::AsyncExt, types::user::ThemeLineage};
use error_stack::ResultExt;
#[cfg(feature = "email")]
use masking::{ExposeInterface, PeekInterface, Secret};

35
crates/router/src/core/user.rs
View File

@ -8,7 +8,10 @@ use api_models::{
user::{self as user_api, InviteMultipleUserResponse, NameIdUnit},
};
use common_enums::{EntityType, UserAuthType};
use common_utils::{type_name, types::keymanager::Identifier};
use common_utils::{
type_name,
types::{keymanager::Identifier, user::LineageContext},
};
#[cfg(feature = "email")]
use diesel_models::user_role::UserRoleUpdate;
use diesel_models::{
@ -3237,7 +3240,7 @@ pub async fn switch_org_for_user(
}
};
let lineage_context = domain::LineageContext {
let lineage_context = LineageContext {
user_id: user_from_token.user_id.clone(),
merchant_id: merchant_id.clone(),
role_id: role_id.clone(),
@ -3250,9 +3253,11 @@ pub async fn switch_org_for_user(
.clone(),
};
lineage_context
.try_set_lineage_context_in_cache(&state, user_from_token.user_id.as_str())
.await;
utils::user::spawn_async_lineage_context_update_to_db(
&state,
&user_from_token.user_id,
lineage_context,
);
let token = utils::user::generate_jwt_auth_token_with_attributes(
&state,
@ -3449,7 +3454,7 @@ pub async fn switch_merchant_for_user_in_org(
}
};
let lineage_context = domain::LineageContext {
let lineage_context = LineageContext {
user_id: user_from_token.user_id.clone(),
merchant_id: merchant_id.clone(),
role_id: role_id.clone(),
@ -3462,9 +3467,11 @@ pub async fn switch_merchant_for_user_in_org(
.clone(),
};
lineage_context
.try_set_lineage_context_in_cache(&state, user_from_token.user_id.as_str())
.await;
utils::user::spawn_async_lineage_context_update_to_db(
&state,
&user_from_token.user_id,
lineage_context,
);
let token = utils::user::generate_jwt_auth_token_with_attributes(
&state,
@ -3582,7 +3589,7 @@ pub async fn switch_profile_for_user_in_org_and_merchant(
}
};
let lineage_context = domain::LineageContext {
let lineage_context = LineageContext {
user_id: user_from_token.user_id.clone(),
merchant_id: user_from_token.merchant_id.clone(),
role_id: role_id.clone(),
@ -3595,9 +3602,11 @@ pub async fn switch_profile_for_user_in_org_and_merchant(
.clone(),
};
lineage_context
.try_set_lineage_context_in_cache(&state, user_from_token.user_id.as_str())
.await;
utils::user::spawn_async_lineage_context_update_to_db(
&state,
&user_from_token.user_id,
lineage_context,
);
let token = utils::user::generate_jwt_auth_token_with_attributes(
&state,

2
crates/router/src/core/user/theme.rs
View File

@ -1,7 +1,7 @@
use api_models::user::theme as theme_api;
use common_utils::{
ext_traits::{ByteSliceExt, Encode},
types::theme::ThemeLineage,
types::user::ThemeLineage,
};
use diesel_models::user::theme::ThemeNew;
use error_stack::ResultExt;

2
crates/router/src/db/kafka_store.rs
View File

@ -5,7 +5,7 @@ use common_enums::enums::MerchantStorageScheme;
use common_utils::{
errors::CustomResult,
id_type,
types::{keymanager::KeyManagerState, theme::ThemeLineage},
types::{keymanager::KeyManagerState, user::ThemeLineage},
};
#[cfg(feature = "v2")]
use diesel_models::ephemeral_key::{ClientSecretType, ClientSecretTypeNew};

23
crates/router/src/db/user.rs
View File

@ -163,6 +163,7 @@ impl UserInterface for MockDb {
totp_secret: user_data.totp_secret,
totp_recovery_codes: user_data.totp_recovery_codes,
last_password_modified_at: user_data.last_password_modified_at,
lineage_context: user_data.lineage_context,
};
users.push(user.clone());
Ok(user)
@ -208,17 +209,20 @@ impl UserInterface for MockDb {
update_user: storage::UserUpdate,
) -> CustomResult<storage::User, errors::StorageError> {
let mut users = self.users.lock().await;
let last_modified_at = common_utils::date_time::now();
users
.iter_mut()
.find(|user| user.user_id == user_id)
.map(|user| {
*user = match &update_user {
storage::UserUpdate::VerifyUser => storage::User {
last_modified_at,
is_verified: true,
..user.to_owned()
},
storage::UserUpdate::AccountUpdate { name, is_verified } => storage::User {
name: name.clone().map(Secret::new).unwrap_or(user.name.clone()),
last_modified_at,
is_verified: is_verified.unwrap_or(user.is_verified),
..user.to_owned()
},
@ -227,6 +231,7 @@ impl UserInterface for MockDb {
totp_secret,
totp_recovery_codes,
} => storage::User {
last_modified_at,
totp_status: totp_status.unwrap_or(user.totp_status),
totp_secret: totp_secret.clone().or(user.totp_secret.clone()),
totp_recovery_codes: totp_recovery_codes
@ -239,6 +244,13 @@ impl UserInterface for MockDb {
last_password_modified_at: Some(common_utils::date_time::now()),
..user.to_owned()
},
storage::UserUpdate::LineageContextUpdate { lineage_context } => {
storage::User {
last_modified_at,
lineage_context: Some(lineage_context.clone()),
..user.to_owned()
}
}
};
user.to_owned()
})
@ -256,17 +268,20 @@ impl UserInterface for MockDb {
update_user: storage::UserUpdate,
) -> CustomResult<storage::User, errors::StorageError> {
let mut users = self.users.lock().await;
let last_modified_at = common_utils::date_time::now();
users
.iter_mut()
.find(|user| user.email.eq(user_email.get_inner()))
.map(|user| {
*user = match &update_user {
storage::UserUpdate::VerifyUser => storage::User {
last_modified_at,
is_verified: true,
..user.to_owned()
},
storage::UserUpdate::AccountUpdate { name, is_verified } => storage::User {
name: name.clone().map(Secret::new).unwrap_or(user.name.clone()),
last_modified_at,
is_verified: is_verified.unwrap_or(user.is_verified),
..user.to_owned()
},
@ -275,6 +290,7 @@ impl UserInterface for MockDb {
totp_secret,
totp_recovery_codes,
} => storage::User {
last_modified_at,
totp_status: totp_status.unwrap_or(user.totp_status),
totp_secret: totp_secret.clone().or(user.totp_secret.clone()),
totp_recovery_codes: totp_recovery_codes
@ -287,6 +303,13 @@ impl UserInterface for MockDb {
last_password_modified_at: Some(common_utils::date_time::now()),
..user.to_owned()
},
storage::UserUpdate::LineageContextUpdate { lineage_context } => {
storage::User {
last_modified_at,
lineage_context: Some(lineage_context.clone()),
..user.to_owned()
}
}
};
user.to_owned()
})

2
crates/router/src/db/user/theme.rs
View File

@ -1,4 +1,4 @@
use common_utils::types::theme::ThemeLineage;
use common_utils::types::user::ThemeLineage;
use diesel_models::user::theme as storage;
use error_stack::report;

2
crates/router/src/routes/user/theme.rs
View File

@ -1,7 +1,7 @@
use actix_multipart::form::MultipartForm;
use actix_web::{web, HttpRequest, HttpResponse};
use api_models::user::theme as theme_api;
use common_utils::types::theme::ThemeLineage;
use common_utils::types::user::ThemeLineage;
use masking::Secret;
use router_env::Flow;

2
crates/router/src/services/email/types.rs
View File

@ -1,6 +1,6 @@
use api_models::user::dashboard_metadata::ProdIntent;
use common_enums::{EntityType, MerchantProductType};
use common_utils::{errors::CustomResult, pii, types::theme::EmailThemeConfig};
use common_utils::{errors::CustomResult, pii, types::user::EmailThemeConfig};
use error_stack::ResultExt;
use external_services::email::{EmailContents, EmailData, EmailError};
use masking::{ExposeInterface, Secret};

53
crates/router/src/types/domain/user.rs
View File

@ -23,7 +23,6 @@ use hyperswitch_domain_models::api::ApplicationResponse;
use masking::{ExposeInterface, PeekInterface, Secret};
use once_cell::sync::Lazy;
use rand::distributions::{Alphanumeric, DistString};
use router_env::logger;
use time::PrimitiveDateTime;
use unicode_segmentation::UnicodeSegmentation;
#[cfg(feature = "keymanager_create")]
@ -925,6 +924,7 @@ impl TryFrom<NewUser> for storage_user::UserNew {
last_password_modified_at: value
.password
.and_then(|password_inner| password_inner.is_temporary.not().then_some(now)),
lineage_context: None,
})
}
}
@ -1499,54 +1499,3 @@ where
.change_context(UserErrors::InternalServerError)
}
}
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub struct LineageContext {
pub user_id: String,
pub merchant_id: id_type::MerchantId,
pub role_id: String,
pub org_id: id_type::OrganizationId,
pub profile_id: id_type::ProfileId,
pub tenant_id: id_type::TenantId,
}
impl LineageContext {
pub async fn try_get_lineage_context_from_cache(
state: &SessionState,
user_id: &str,
) -> Option<Self> {
// The errors are not handled here because we don't want to fail the request if the cache operation fails.
// The errors are logged for debugging purposes.
match utils::user::get_lineage_context_from_cache(state, user_id).await {
Ok(Some(ctx)) => Some(ctx),
Ok(None) => {
logger::debug!("Lineage context not found in Redis for user {}", user_id);
None
}
Err(e) => {
logger::error!(
"Failed to retrieve lineage context from Redis for user {}: {:?}",
user_id,
e
);
None
}
}
}
pub async fn try_set_lineage_context_in_cache(&self, state: &SessionState, user_id: &str) {
// The errors are not handled here because we don't want to fail the request if the cache operation fails.
// The errors are logged for debugging purposes.
if let Err(e) =
utils::user::set_lineage_context_in_cache(state, user_id, self.clone()).await
{
logger::error!(
"Failed to set lineage context in Redis for user {}: {:?}",
user_id,
e
);
} else {
logger::debug!("Lineage context cached for user {}", user_id);
}
}
}

42
crates/router/src/types/domain/user/decision_manager.rs
View File

@ -1,11 +1,12 @@
use common_enums::TokenPurpose;
use common_utils::id_type;
use common_utils::{id_type, types::user::LineageContext};
use diesel_models::{
enums::{UserRoleVersion, UserStatus},
user_role::UserRole,
};
use error_stack::ResultExt;
use masking::Secret;
use router_env::logger;
use super::UserFromStorage;
use crate::{
@ -13,7 +14,6 @@ use crate::{
db::user_role::ListUserRolesByUserIdPayload,
routes::SessionState,
services::authentication as auth,
types::domain::LineageContext,
utils,
};
@ -129,13 +129,25 @@ impl JWTFlow {
user_role: &UserRole,
) -> UserResult<Secret<String>> {
let user_id = next_flow.user.get_user_id();
let cached_lineage_context =
LineageContext::try_get_lineage_context_from_cache(state, user_id).await;
// Fetch lineage context from DB
let lineage_context_from_db = state
.global_store
.find_user_by_id(user_id)
.await
.inspect_err(|e| {
logger::error!(
"Failed to fetch lineage context from DB for user {}: {:?}",
user_id,
e
)
})
.ok()
.and_then(|user| user.lineage_context);
let new_lineage_context = match cached_lineage_context {
let new_lineage_context = match lineage_context_from_db {
Some(ctx) => {
let tenant_id = ctx.tenant_id.clone();
let user_role_match_v1 = state
let user_role_match_v2 = state
.global_store
.find_user_role_by_user_id_and_lineage(
&ctx.user_id,
@ -143,15 +155,15 @@ impl JWTFlow {
&ctx.org_id,
&ctx.merchant_id,
&ctx.profile_id,
UserRoleVersion::V1,
UserRoleVersion::V2,
)
.await
.is_ok();
if user_role_match_v1 {
if user_role_match_v2 {
ctx
} else {
let user_role_match_v2 = state
let user_role_match_v1 = state
.global_store
.find_user_role_by_user_id_and_lineage(
&ctx.user_id,
@ -159,12 +171,12 @@ impl JWTFlow {
&ctx.org_id,
&ctx.merchant_id,
&ctx.profile_id,
UserRoleVersion::V2,
UserRoleVersion::V1,
)
.await
.is_ok();
if user_role_match_v2 {
if user_role_match_v1 {
ctx
} else {
// fallback to default lineage if cached context is invalid
@ -179,9 +191,11 @@ impl JWTFlow {
}
};
new_lineage_context
.try_set_lineage_context_in_cache(state, user_id)
.await;
utils::user::spawn_async_lineage_context_update_to_db(
state,
user_id,
new_lineage_context.clone(),
);
auth::AuthToken::new_token(
new_lineage_context.user_id,

81
crates/router/src/utils/user.rs
View File

@ -3,19 +3,19 @@ use std::sync::Arc;
use api_models::user as user_api;
use common_enums::UserAuthType;
use common_utils::{
encryption::Encryption, errors::CustomResult, id_type, type_name, types::keymanager::Identifier,
encryption::Encryption,
errors::CustomResult,
id_type, type_name,
types::{keymanager::Identifier, user::LineageContext},
};
use diesel_models::organization::{self, OrganizationBridge};
use error_stack::ResultExt;
use masking::{ExposeInterface, Secret};
use redis_interface::RedisConnectionPool;
use router_env::env;
use router_env::{env, logger};
use crate::{
consts::user::{
LINEAGE_CONTEXT_PREFIX, LINEAGE_CONTEXT_TIME_EXPIRY_IN_SECS, REDIS_SSO_PREFIX,
REDIS_SSO_TTL,
},
consts::user::{REDIS_SSO_PREFIX, REDIS_SSO_TTL},
core::errors::{StorageError, UserErrors, UserResult},
routes::SessionState,
services::{
@ -23,7 +23,7 @@ use crate::{
authorization::roles::RoleInfo,
},
types::{
domain::{self, LineageContext, MerchantAccount, UserFromStorage},
domain::{self, MerchantAccount, UserFromStorage},
transformers::ForeignFrom,
},
};
@ -341,50 +341,35 @@ pub async fn validate_email_domain_auth_type_using_db(
.ok_or(UserErrors::InvalidUserAuthMethodOperation.into())
}
pub async fn get_lineage_context_from_cache(
state: &SessionState,
user_id: &str,
) -> UserResult<Option<LineageContext>> {
let connection = get_redis_connection_for_global_tenant(state)?;
let key = format!("{}{}", LINEAGE_CONTEXT_PREFIX, user_id);
let lineage_context = connection
.get_key::<Option<String>>(&key.into())
.await
.change_context(UserErrors::InternalServerError)
.attach_printable("Failed to get lineage context from redis")?;
match lineage_context {
Some(json_str) => {
let ctx = serde_json::from_str::<LineageContext>(&json_str)
.change_context(UserErrors::InternalServerError)
.attach_printable("Failed to deserialize LineageContext from JSON")?;
Ok(Some(ctx))
}
None => Ok(None),
}
}
pub async fn set_lineage_context_in_cache(
pub fn spawn_async_lineage_context_update_to_db(
state: &SessionState,
user_id: &str,
lineage_context: LineageContext,
) -> UserResult<()> {
let connection = get_redis_connection_for_global_tenant(state)?;
let key = format!("{}{}", LINEAGE_CONTEXT_PREFIX, user_id);
let serialized_lineage_context: String = serde_json::to_string(&lineage_context)
.change_context(UserErrors::InternalServerError)
.attach_printable("Failed to serialize LineageContext")?;
connection
.set_key_with_expiry(
&key.into(),
serialized_lineage_context,
LINEAGE_CONTEXT_TIME_EXPIRY_IN_SECS,
)
.await
.change_context(UserErrors::InternalServerError)
.attach_printable("Failed to set lineage context in redis")?;
Ok(())
) {
let state = state.clone();
let lineage_context = lineage_context.clone();
let user_id = user_id.to_owned();
tokio::spawn(async move {
match state
.global_store
.update_user_by_user_id(
&user_id,
diesel_models::user::UserUpdate::LineageContextUpdate { lineage_context },
)
.await
{
Ok(_) => {
logger::debug!("Successfully updated lineage context for user {}", user_id);
}
Err(e) => {
logger::error!(
"Failed to update lineage context for user {}: {:?}",
user_id,
e
);
}
}
});
}
pub fn generate_env_specific_merchant_id(value: String) -> UserResult<id_type::MerchantId> {

2
crates/router/src/utils/user/theme.rs
View File

@ -1,7 +1,7 @@
use std::path::PathBuf;
use common_enums::EntityType;
use common_utils::{ext_traits::AsyncExt, id_type, types::theme::ThemeLineage};
use common_utils::{ext_traits::AsyncExt, id_type, types::user::ThemeLineage};
use diesel_models::user::theme::Theme;
use error_stack::ResultExt;
use hyperswitch_domain_models::merchant_key_store::MerchantKeyStore;

2
crates/router/src/workflows/api_key_expiry.rs
View File

@ -1,4 +1,4 @@
use common_utils::{errors::ValidationError, ext_traits::ValueExt, types::theme::ThemeLineage};
use common_utils::{errors::ValidationError, ext_traits::ValueExt, types::user::ThemeLineage};
use diesel_models::{
enums as storage_enums, process_tracker::business_status, ApiKeyExpiryTrackingData,
};