rust/hyperswitch
1
0
Fork 0
mirror of https://github.com/juspay/hyperswitch.git synced 2025-10-29 00:49:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(user_role): Add update by lineage DB function (#5651)

Browse Source
This commit is contained in:
Mani Chandra
2024-08-22 10:30:07 +05:30
committed by GitHub
parent d2754dbc97
commit ca72fedae8
9 changed files with 597 additions and 631 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
crates/diesel_models/src/query/user_role.rs
View File

@ -41,47 +41,6 @@ impl UserRole {
.await
}
pub async fn update_by_user_id_merchant_id(
conn: &PgPooledConn,
user_id: String,
merchant_id: id_type::MerchantId,
update: UserRoleUpdate,
version: UserRoleVersion,
) -> StorageResult<Self> {
generics::generic_update_with_unique_predicate_get_result::<
<Self as HasTable>::Table,
_,
_,
_,
>(
conn,
dsl::user_id
.eq(user_id)
.and(dsl::merchant_id.eq(merchant_id))
.and(dsl::version.eq(version)),
UserRoleUpdateInternal::from(update),
)
.await
}
pub async fn update_by_user_id_org_id(
conn: &PgPooledConn,
user_id: String,
org_id: id_type::OrganizationId,
update: UserRoleUpdate,
version: UserRoleVersion,
) -> StorageResult<Vec<Self>> {
generics::generic_update_with_results::<<Self as HasTable>::Table, _, _, _>(
conn,
dsl::user_id
.eq(user_id)
.and(dsl::org_id.eq(org_id))
.and(dsl::version.eq(version)),
UserRoleUpdateInternal::from(update),
)
.await
}
pub async fn list_by_user_id(
conn: &PgPooledConn,
user_id: String,
@ -147,6 +106,46 @@ impl UserRole {
generics::generic_find_one::<<Self as HasTable>::Table, _, _>(conn, predicate).await
}
pub async fn update_by_user_id_org_id_merchant_id_profile_id(
conn: &PgPooledConn,
user_id: String,
org_id: id_type::OrganizationId,
merchant_id: id_type::MerchantId,
profile_id: Option<String>,
update: UserRoleUpdate,
version: UserRoleVersion,
) -> StorageResult<Self> {
// Checking in user roles, for a user in token hierarchy, only one of the relation will be true, either org level, merchant level or profile level
// (org_id = ? && merchant_id = null && profile_id = null) || (org_id = ? && merchant_id = ? && profile_id = null) || (org_id = ? && merchant_id = ? && profile_id = ?)
let check_lineage = dsl::org_id
.eq(org_id.clone())
.and(dsl::merchant_id.is_null().and(dsl::profile_id.is_null()))
.or(dsl::org_id.eq(org_id.clone()).and(
dsl::merchant_id
.eq(merchant_id.clone())
.and(dsl::profile_id.is_null()),
))
.or(dsl::org_id.eq(org_id).and(
dsl::merchant_id
.eq(merchant_id)
//TODO: In case of None, profile_id = NULL its unexpected behaviour, after V1 profile id will not be option
.and(dsl::profile_id.eq(profile_id)),
));
let predicate = dsl::user_id
.eq(user_id)
.and(check_lineage)
.and(dsl::version.eq(version));
generics::generic_update_with_unique_predicate_get_result::<
<Self as HasTable>::Table,
UserRoleUpdateInternal,
_,
_,
>(conn, predicate, update.into())
.await
}
pub async fn delete_by_user_id_org_id_merchant_id_profile_id(
conn: &PgPooledConn,
user_id: String,

1
crates/diesel_models/src/user_role.rs
View File

@ -51,6 +51,7 @@ pub struct UserRoleUpdateInternal {
last_modified: PrimitiveDateTime,
}
#[derive(Clone)]
pub enum UserRoleUpdate {
UpdateStatus {
status: enums::UserStatus,

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

@ -594,19 +594,58 @@ pub async fn reset_password(
.change_context(UserErrors::InternalServerError)?;
if let Some(inviter_merchant_id) = email_token.get_merchant_id() {
let update_status_result = state
let key_manager_state = &(&state).into();
let key_store = state
.store
.update_user_role_by_user_id_merchant_id(
user.user_id.clone().as_str(),
.get_merchant_key_store_by_merchant_id(
key_manager_state,
inviter_merchant_id,
&state.store.get_master_key().to_vec().into(),
)
.await
.change_context(UserErrors::InternalServerError)
.attach_printable("merchant_key_store not found")?;
let merchant_account = state
.store
.find_merchant_account_by_merchant_id(
key_manager_state,
inviter_merchant_id,
&key_store,
)
.await
.change_context(UserErrors::InternalServerError)
.attach_printable("merchant_account not found")?;
let (update_v1_result, update_v2_result) =
utils::user_role::update_v1_and_v2_user_roles_in_db(
&state,
user.user_id.clone().as_str(),
&merchant_account.organization_id,
inviter_merchant_id,
None,
UserRoleUpdate::UpdateStatus {
status: UserStatus::Active,
modified_by: user.user_id.clone(),
},
UserRoleVersion::V1,
)
.await;
logger::info!(?update_status_result);
if update_v1_result
.as_ref()
.is_err_and(|err| !err.current_context().is_db_not_found())
|| update_v2_result
.as_ref()
.is_err_and(|err| !err.current_context().is_db_not_found())
{
return Err(report!(UserErrors::InternalServerError));
}
if update_v1_result.is_err() && update_v2_result.is_err() {
return Err(report!(UserErrors::InvalidRoleOperation))
.attach_printable("User not found in the organization")?;
}
}
let _ = auth::blacklist::insert_email_token_in_blacklist(&state, &token)
@ -1014,19 +1053,53 @@ pub async fn accept_invite_from_email(
.get_merchant_id()
.ok_or(UserErrors::InternalServerError)?;
let update_status_result = state
let key_manager_state = &(&state).into();
let key_store = state
.store
.update_user_role_by_user_id_merchant_id(
user.get_user_id(),
.get_merchant_key_store_by_merchant_id(
key_manager_state,
merchant_id,
UserRoleUpdate::UpdateStatus {
status: UserStatus::Active,
modified_by: user.get_user_id().to_string(),
},
UserRoleVersion::V1,
&state.store.get_master_key().to_vec().into(),
)
.await
.change_context(UserErrors::InternalServerError)?;
.change_context(UserErrors::InternalServerError)
.attach_printable("merchant_key_store not found")?;
let merchant_account = state
.store
.find_merchant_account_by_merchant_id(key_manager_state, merchant_id, &key_store)
.await
.change_context(UserErrors::InternalServerError)
.attach_printable("merchant_account not found")?;
let (update_v1_result, update_v2_result) = utils::user_role::update_v1_and_v2_user_roles_in_db(
&state,
user.get_user_id(),
&merchant_account.organization_id,
merchant_id,
None,
UserRoleUpdate::UpdateStatus {
status: UserStatus::Active,
modified_by: user.get_user_id().to_string(),
},
)
.await;
if update_v1_result
.as_ref()
.is_err_and(|err| !err.current_context().is_db_not_found())
|| update_v2_result
.as_ref()
.is_err_and(|err| !err.current_context().is_db_not_found())
{
return Err(report!(UserErrors::InternalServerError));
}
if update_v1_result.is_err() && update_v2_result.is_err() {
return Err(report!(UserErrors::InvalidRoleOperation))
.attach_printable("User not found in the organization")?;
}
let _ = auth::blacklist::insert_email_token_in_blacklist(&state, &token)
.await
@ -1039,21 +1112,18 @@ pub async fn accept_invite_from_email(
.change_context(UserErrors::InternalServerError)?
.into();
let token = utils::user::generate_jwt_auth_token_without_profile(
&state,
&user_from_db,
&update_status_result,
)
.await?;
utils::user_role::set_role_permissions_in_cache_by_user_role(&state, &update_status_result)
.await;
let user_role = user_from_db
.get_preferred_or_active_user_role_from_db(&state)
.await
.change_context(UserErrors::InternalServerError)?;
let response = utils::user::get_dashboard_entry_response(
&state,
user_from_db,
update_status_result,
token.clone(),
)?;
let token =
utils::user::generate_jwt_auth_token_without_profile(&state, &user_from_db, &user_role)
.await?;
utils::user_role::set_role_permissions_in_cache_by_user_role(&state, &user_role).await;
let response =
utils::user::get_dashboard_entry_response(&state, user_from_db, user_role, token.clone())?;
auth::cookies::set_cookie_response(response, token)
}
@ -1091,19 +1161,53 @@ pub async fn accept_invite_from_email_token_only_flow(
.get_merchant_id()
.ok_or(UserErrors::LinkInvalid)?;
let user_role = state
let key_manager_state = &(&state).into();
let key_store = state
.store
.update_user_role_by_user_id_merchant_id(
user_from_db.get_user_id(),
.get_merchant_key_store_by_merchant_id(
key_manager_state,
merchant_id,
UserRoleUpdate::UpdateStatus {
status: UserStatus::Active,
modified_by: user_from_db.get_user_id().to_string(),
},
UserRoleVersion::V1,
&state.store.get_master_key().to_vec().into(),
)
.await
.change_context(UserErrors::InternalServerError)?;
.change_context(UserErrors::InternalServerError)
.attach_printable("merchant_key_store not found")?;
let merchant_account = state
.store
.find_merchant_account_by_merchant_id(key_manager_state, merchant_id, &key_store)
.await
.change_context(UserErrors::InternalServerError)
.attach_printable("merchant_account not found")?;
let (update_v1_result, update_v2_result) = utils::user_role::update_v1_and_v2_user_roles_in_db(
&state,
user_from_db.get_user_id(),
&merchant_account.organization_id,
merchant_id,
None,
UserRoleUpdate::UpdateStatus {
status: UserStatus::Active,
modified_by: user_from_db.get_user_id().to_owned(),
},
)
.await;
if update_v1_result
.as_ref()
.is_err_and(|err| !err.current_context().is_db_not_found())
|| update_v2_result
.as_ref()
.is_err_and(|err| !err.current_context().is_db_not_found())
{
return Err(report!(UserErrors::InternalServerError));
}
if update_v1_result.is_err() && update_v2_result.is_err() {
return Err(report!(UserErrors::InvalidRoleOperation))
.attach_printable("User not found in the organization")?;
}
if !user_from_db.is_verified() {
let _ = state
@ -1126,6 +1230,11 @@ pub async fn accept_invite_from_email_token_only_flow(
)?;
let next_flow = current_flow.next(user_from_db.clone(), &state).await?;
let user_role = user_from_db
.get_preferred_or_active_user_role_from_db(&state)
.await
.change_context(UserErrors::InternalServerError)?;
let token = next_flow
.get_token_with_user_role(&state, &user_role)
.await?;

397
crates/router/src/core/user_role.rs
View File

@ -7,10 +7,8 @@ use diesel_models::{
};
use error_stack::{report, ResultExt};
use once_cell::sync::Lazy;
use router_env::logger;
use crate::{
consts,
core::errors::{StorageErrorExt, UserErrors, UserResponse},
routes::{app::ReqState, SessionState},
services::{
@ -115,103 +113,155 @@ pub async fn update_user_role(
.attach_printable("User Changing their own role");
}
let user_role_to_be_updated = user_to_be_updated
.get_role_from_db_by_merchant_id(&state, &user_from_token.merchant_id)
.await
.to_not_found_response(UserErrors::InvalidRoleOperation)?;
let role_to_be_updated = roles::RoleInfo::from_role_id(
let updator_role = roles::RoleInfo::from_role_id(
&state,
&user_role_to_be_updated.role_id,
&user_from_token.role_id,
&user_from_token.merchant_id,
&user_from_token.org_id,
)
.await
.change_context(UserErrors::InternalServerError)?;
if !role_to_be_updated.is_updatable() {
return Err(report!(UserErrors::InvalidRoleOperation)).attach_printable(format!(
"User role cannot be updated from {}",
role_to_be_updated.get_role_id()
));
}
let mut is_updated = false;
state
let v2_user_role_to_be_updated = match state
.store
.update_user_role_by_user_id_merchant_id(
user_to_be_updated.get_user_id(),
&user_role_to_be_updated
.merchant_id
.ok_or(UserErrors::InternalServerError)
.attach_printable("merchant_id not found in user_role")?,
UserRoleUpdate::UpdateRole {
role_id: req.role_id.clone(),
modified_by: user_from_token.user_id,
},
UserRoleVersion::V1,
)
.await
.to_not_found_response(UserErrors::InvalidRoleOperation)
.attach_printable("User with given email is not found in the organization")?;
auth::blacklist::insert_user_in_blacklist(&state, user_to_be_updated.get_user_id()).await?;
Ok(ApplicationResponse::StatusOk)
}
pub async fn transfer_org_ownership(
state: SessionState,
user_from_token: auth::UserFromToken,
req: user_role_api::TransferOrgOwnershipRequest,
_req_state: ReqState,
) -> UserResponse<user_api::DashboardEntryResponse> {
if user_from_token.role_id != consts::user_role::ROLE_ID_ORGANIZATION_ADMIN {
return Err(report!(UserErrors::InvalidRoleOperation)).attach_printable(format!(
"role_id = {} is not org_admin",
user_from_token.role_id
));
}
let user_to_be_updated =
utils::user::get_user_from_db_by_email(&state, domain::UserEmail::try_from(req.email)?)
.await
.to_not_found_response(UserErrors::InvalidRoleOperation)
.attach_printable("User not found in our records".to_string())?;
if user_from_token.user_id == user_to_be_updated.get_user_id() {
return Err(report!(UserErrors::InvalidRoleOperation))
.attach_printable("User transferring ownership to themselves".to_string());
}
state
.store
.transfer_org_ownership_between_users(
&user_from_token.user_id,
.find_user_role_by_user_id_and_lineage(
user_to_be_updated.get_user_id(),
&user_from_token.org_id,
UserRoleVersion::V1,
&user_from_token.merchant_id,
None,
UserRoleVersion::V2,
)
.await
{
Ok(user_role) => Some(user_role),
Err(e) => {
if e.current_context().is_db_not_found() {
None
} else {
return Err(UserErrors::InternalServerError.into());
}
}
};
if let Some(user_role) = v2_user_role_to_be_updated {
let role_to_be_updated = roles::RoleInfo::from_role_id(
&state,
&user_role.role_id,
&user_from_token.merchant_id,
&user_from_token.org_id,
)
.await
.change_context(UserErrors::InternalServerError)?;
auth::blacklist::insert_user_in_blacklist(&state, user_to_be_updated.get_user_id()).await?;
auth::blacklist::insert_user_in_blacklist(&state, &user_from_token.user_id).await?;
if !role_to_be_updated.is_updatable() {
return Err(report!(UserErrors::InvalidRoleOperation)).attach_printable(format!(
"User role cannot be updated from {}",
role_to_be_updated.get_role_id()
));
}
let user_from_db = user_from_token.get_user_from_db(&state).await?;
let user_role = user_from_db
.get_role_from_db_by_merchant_id(&state, &user_from_token.merchant_id)
if updator_role.get_entity_type() < role_to_be_updated.get_entity_type() {
return Err(report!(UserErrors::InvalidRoleOperation)).attach_printable(format!(
"Invalid operation, update requestor = {} cannot update target = {}",
updator_role.get_entity_type(),
role_to_be_updated.get_entity_type()
));
}
state
.store
.update_user_role_by_user_id_and_lineage(
user_to_be_updated.get_user_id(),
&user_from_token.org_id,
&user_from_token.merchant_id,
None,
UserRoleUpdate::UpdateRole {
role_id: req.role_id.clone(),
modified_by: user_from_token.user_id.clone(),
},
UserRoleVersion::V2,
)
.await
.change_context(UserErrors::InternalServerError)?;
is_updated = true;
}
let v1_user_role_to_be_updated = match state
.store
.find_user_role_by_user_id_and_lineage(
user_to_be_updated.get_user_id(),
&user_from_token.org_id,
&user_from_token.merchant_id,
None,
UserRoleVersion::V1,
)
.await
.to_not_found_response(UserErrors::InvalidRoleOperation)?;
{
Ok(user_role) => Some(user_role),
Err(e) => {
if e.current_context().is_db_not_found() {
None
} else {
return Err(UserErrors::InternalServerError.into());
}
}
};
utils::user_role::set_role_permissions_in_cache_by_user_role(&state, &user_role).await;
if let Some(user_role) = v1_user_role_to_be_updated {
let role_to_be_updated = roles::RoleInfo::from_role_id(
&state,
&user_role.role_id,
&user_from_token.merchant_id,
&user_from_token.org_id,
)
.await
.change_context(UserErrors::InternalServerError)?;
let token =
utils::user::generate_jwt_auth_token_without_profile(&state, &user_from_db, &user_role)
.await?;
let response =
utils::user::get_dashboard_entry_response(&state, user_from_db, user_role, token.clone())?;
if !role_to_be_updated.is_updatable() {
return Err(report!(UserErrors::InvalidRoleOperation)).attach_printable(format!(
"User role cannot be updated from {}",
role_to_be_updated.get_role_id()
));
}
auth::cookies::set_cookie_response(response, token)
if updator_role.get_entity_type() < role_to_be_updated.get_entity_type() {
return Err(report!(UserErrors::InvalidRoleOperation)).attach_printable(format!(
"Invalid operation, update requestor = {} cannot update target = {}",
updator_role.get_entity_type(),
role_to_be_updated.get_entity_type()
));
}
state
.store
.update_user_role_by_user_id_and_lineage(
user_to_be_updated.get_user_id(),
&user_from_token.org_id,
&user_from_token.merchant_id,
None,
UserRoleUpdate::UpdateRole {
role_id: req.role_id.clone(),
modified_by: user_from_token.user_id,
},
UserRoleVersion::V1,
)
.await
.change_context(UserErrors::InternalServerError)?;
is_updated = true;
}
if !is_updated {
return Err(report!(UserErrors::InvalidRoleOperation))
.attach_printable("User with given email is not found in the organization")?;
}
auth::blacklist::insert_user_in_blacklist(&state, user_to_be_updated.get_user_id()).await?;
Ok(ApplicationResponse::StatusOk)
}
pub async fn accept_invitation(
@ -219,30 +269,43 @@ pub async fn accept_invitation(
user_token: auth::UserFromToken,
req: user_role_api::AcceptInvitationRequest,
) -> UserResponse<()> {
futures::future::join_all(req.merchant_ids.iter().map(|merchant_id| async {
state
.store
.update_user_role_by_user_id_merchant_id(
user_token.user_id.as_str(),
merchant_id,
UserRoleUpdate::UpdateStatus {
status: UserStatus::Active,
modified_by: user_token.user_id.clone(),
},
UserRoleVersion::V1,
)
.await
.map_err(|e| {
logger::error!("Error while accepting invitation {e:?}");
})
.ok()
}))
.await
.into_iter()
.reduce(Option::or)
.flatten()
.ok_or(UserErrors::MerchantIdNotFound.into())
.map(|_| ApplicationResponse::StatusOk)
let merchant_accounts = state
.store
.list_multiple_merchant_accounts(&(&state).into(), req.merchant_ids)
.await
.change_context(UserErrors::InternalServerError)?;
let update_result =
futures::future::join_all(merchant_accounts.iter().map(|merchant_account| async {
let (update_v1_result, update_v2_result) =
utils::user_role::update_v1_and_v2_user_roles_in_db(
&state,
user_token.user_id.as_str(),
&merchant_account.organization_id,
merchant_account.get_id(),
None,
UserRoleUpdate::UpdateStatus {
status: UserStatus::Active,
modified_by: user_token.user_id.clone(),
},
)
.await;
if update_v1_result.is_err_and(|err| !err.current_context().is_db_not_found())
|| update_v2_result.is_err_and(|err| !err.current_context().is_db_not_found())
{
Err(report!(UserErrors::InternalServerError))
} else {
Ok(())
}
}))
.await;
if update_result.iter().all(Result::is_err) {
return Err(UserErrors::MerchantIdNotFound.into());
}
Ok(ApplicationResponse::StatusOk)
}
pub async fn merchant_select(
@ -250,38 +313,55 @@ pub async fn merchant_select(
user_token: auth::UserFromSinglePurposeToken,
req: user_role_api::MerchantSelectRequest,
) -> UserResponse<user_api::TokenOrPayloadResponse<user_api::DashboardEntryResponse>> {
let user_role = futures::future::join_all(req.merchant_ids.iter().map(|merchant_id| async {
state
.store
.update_user_role_by_user_id_merchant_id(
user_token.user_id.as_str(),
merchant_id,
UserRoleUpdate::UpdateStatus {
status: UserStatus::Active,
modified_by: user_token.user_id.clone(),
},
UserRoleVersion::V1,
)
.await
.map_err(|e| {
logger::error!("Error while accepting invitation {e:?}");
})
.ok()
}))
.await
.into_iter()
.reduce(Option::or)
.flatten()
.ok_or(UserErrors::MerchantIdNotFound)?;
let merchant_accounts = state
.store
.list_multiple_merchant_accounts(&(&state).into(), req.merchant_ids)
.await
.change_context(UserErrors::InternalServerError)?;
let update_result =
futures::future::join_all(merchant_accounts.iter().map(|merchant_account| async {
let (update_v1_result, update_v2_result) =
utils::user_role::update_v1_and_v2_user_roles_in_db(
&state,
user_token.user_id.as_str(),
&merchant_account.organization_id,
merchant_account.get_id(),
None,
UserRoleUpdate::UpdateStatus {
status: UserStatus::Active,
modified_by: user_token.user_id.clone(),
},
)
.await;
if update_v1_result.is_err_and(|err| !err.current_context().is_db_not_found())
|| update_v2_result.is_err_and(|err| !err.current_context().is_db_not_found())
{
Err(report!(UserErrors::InternalServerError))
} else {
Ok(())
}
}))
.await;
if update_result.iter().all(Result::is_err) {
return Err(UserErrors::MerchantIdNotFound.into());
}
if let Some(true) = req.need_dashboard_entry_response {
let user_from_db = state
let user_from_db: domain::UserFromStorage = state
.global_store
.find_user_by_id(user_token.user_id.as_str())
.await
.change_context(UserErrors::InternalServerError)?
.into();
let user_role = user_from_db
.get_preferred_or_active_user_role_from_db(&state)
.await
.change_context(UserErrors::InternalServerError)?;
utils::user_role::set_role_permissions_in_cache_by_user_role(&state, &user_role).await;
let token =
@ -307,29 +387,41 @@ pub async fn merchant_select_token_only_flow(
user_token: auth::UserFromSinglePurposeToken,
req: user_role_api::MerchantSelectRequest,
) -> UserResponse<user_api::TokenOrPayloadResponse<user_api::DashboardEntryResponse>> {
let user_role = futures::future::join_all(req.merchant_ids.iter().map(|merchant_id| async {
state
.store
.update_user_role_by_user_id_merchant_id(
user_token.user_id.as_str(),
merchant_id,
UserRoleUpdate::UpdateStatus {
status: UserStatus::Active,
modified_by: user_token.user_id.clone(),
},
UserRoleVersion::V1,
)
.await
.map_err(|e| {
logger::error!("Error while accepting invitation {e:?}");
})
.ok()
}))
.await
.into_iter()
.reduce(Option::or)
.flatten()
.ok_or(UserErrors::MerchantIdNotFound)?;
let merchant_accounts = state
.store
.list_multiple_merchant_accounts(&(&state).into(), req.merchant_ids)
.await
.change_context(UserErrors::InternalServerError)?;
let update_result =
futures::future::join_all(merchant_accounts.iter().map(|merchant_account| async {
let (update_v1_result, update_v2_result) =
utils::user_role::update_v1_and_v2_user_roles_in_db(
&state,
user_token.user_id.as_str(),
&merchant_account.organization_id,
merchant_account.get_id(),
None,
UserRoleUpdate::UpdateStatus {
status: UserStatus::Active,
modified_by: user_token.user_id.clone(),
},
)
.await;
if update_v1_result.is_err_and(|err| !err.current_context().is_db_not_found())
|| update_v2_result.is_err_and(|err| !err.current_context().is_db_not_found())
{
Err(report!(UserErrors::InternalServerError))
} else {
Ok(())
}
}))
.await;
if update_result.iter().all(Result::is_err) {
return Err(UserErrors::MerchantIdNotFound.into());
}
let user_from_db: domain::UserFromStorage = state
.global_store
@ -338,6 +430,11 @@ pub async fn merchant_select_token_only_flow(
.change_context(UserErrors::InternalServerError)?
.into();
let user_role = user_from_db
.get_preferred_or_active_user_role_from_db(&state)
.await
.change_context(UserErrors::InternalServerError)?;
let current_flow =
domain::CurrentFlow::new(user_token, domain::SPTFlow::MerchantSelect.into())?;
let next_flow = current_flow.next(user_from_db.clone(), &state).await?;

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

@ -2793,30 +2793,6 @@ impl UserRoleInterface for KafkaStore {
.await
}
async fn update_user_role_by_user_id_merchant_id(
&self,
user_id: &str,
merchant_id: &id_type::MerchantId,
update: user_storage::UserRoleUpdate,
version: enums::UserRoleVersion,
) -> CustomResult<user_storage::UserRole, errors::StorageError> {
self.diesel_store
.update_user_role_by_user_id_merchant_id(user_id, merchant_id, update, version)
.await
}
async fn update_user_roles_by_user_id_org_id(
&self,
user_id: &str,
org_id: &id_type::OrganizationId,
update: user_storage::UserRoleUpdate,
version: enums::UserRoleVersion,
) -> CustomResult<Vec<user_storage::UserRole>, errors::StorageError> {
self.diesel_store
.update_user_roles_by_user_id_org_id(user_id, org_id, update, version)
.await
}
async fn list_user_roles_by_user_id(
&self,
user_id: &str,
@ -2846,6 +2822,27 @@ impl UserRoleInterface for KafkaStore {
.await
}
async fn update_user_role_by_user_id_and_lineage(
&self,
user_id: &str,
org_id: &id_type::OrganizationId,
merchant_id: &id_type::MerchantId,
profile_id: Option<&String>,
update: user_storage::UserRoleUpdate,
version: enums::UserRoleVersion,
) -> CustomResult<storage::UserRole, errors::StorageError> {
self.diesel_store
.update_user_role_by_user_id_and_lineage(
user_id,
org_id,
merchant_id,
profile_id,
update,
version,
)
.await
}
async fn delete_user_role_by_user_id_and_lineage(
&self,
user_id: &str,
@ -2865,18 +2862,6 @@ impl UserRoleInterface for KafkaStore {
.await
}
async fn transfer_org_ownership_between_users(
&self,
from_user_id: &str,
to_user_id: &str,
org_id: &id_type::OrganizationId,
version: enums::UserRoleVersion,
) -> CustomResult<(), errors::StorageError> {
self.diesel_store
.transfer_org_ownership_between_users(from_user_id, to_user_id, org_id, version)
.await
}
async fn list_user_roles_by_merchant_id(
&self,
merchant_id: &id_type::MerchantId,

430
crates/router/src/db/user_role.rs
View File

@ -1,6 +1,3 @@
use std::collections::HashSet;
use async_bb8_diesel::AsyncConnection;
use common_utils::id_type;
use diesel_models::{enums, user_role as storage};
use error_stack::{report, ResultExt};
@ -8,7 +5,7 @@ use router_env::{instrument, tracing};
use super::MockDb;
use crate::{
connection, consts,
connection,
core::errors::{self, CustomResult},
services::Store,
};
@ -33,22 +30,6 @@ pub trait UserRoleInterface {
version: enums::UserRoleVersion,
) -> CustomResult<storage::UserRole, errors::StorageError>;
async fn update_user_role_by_user_id_merchant_id(
&self,
user_id: &str,
merchant_id: &id_type::MerchantId,
update: storage::UserRoleUpdate,
version: enums::UserRoleVersion,
) -> CustomResult<storage::UserRole, errors::StorageError>;
async fn update_user_roles_by_user_id_org_id(
&self,
user_id: &str,
org_id: &id_type::OrganizationId,
update: storage::UserRoleUpdate,
version: enums::UserRoleVersion,
) -> CustomResult<Vec<storage::UserRole>, errors::StorageError>;
async fn list_user_roles_by_user_id(
&self,
user_id: &str,
@ -70,6 +51,16 @@ pub trait UserRoleInterface {
version: enums::UserRoleVersion,
) -> CustomResult<storage::UserRole, errors::StorageError>;
async fn update_user_role_by_user_id_and_lineage(
&self,
user_id: &str,
org_id: &id_type::OrganizationId,
merchant_id: &id_type::MerchantId,
profile_id: Option<&String>,
update: storage::UserRoleUpdate,
version: enums::UserRoleVersion,
) -> CustomResult<storage::UserRole, errors::StorageError>;
async fn delete_user_role_by_user_id_and_lineage(
&self,
user_id: &str,
@ -78,14 +69,6 @@ pub trait UserRoleInterface {
profile_id: Option<&String>,
version: enums::UserRoleVersion,
) -> CustomResult<storage::UserRole, errors::StorageError>;
async fn transfer_org_ownership_between_users(
&self,
from_user_id: &str,
to_user_id: &str,
org_id: &id_type::OrganizationId,
version: enums::UserRoleVersion,
) -> CustomResult<(), errors::StorageError>;
}
#[async_trait::async_trait]
@ -132,46 +115,6 @@ impl UserRoleInterface for Store {
.map_err(|error| report!(errors::StorageError::from(error)))
}
#[instrument(skip_all)]
async fn update_user_role_by_user_id_merchant_id(
&self,
user_id: &str,
merchant_id: &id_type::MerchantId,
update: storage::UserRoleUpdate,
version: enums::UserRoleVersion,
) -> CustomResult<storage::UserRole, errors::StorageError> {
let conn = connection::pg_connection_write(self).await?;
storage::UserRole::update_by_user_id_merchant_id(
&conn,
user_id.to_owned(),
merchant_id.to_owned(),
update,
version,
)
.await
.map_err(|error| report!(errors::StorageError::from(error)))
}
#[instrument(skip_all)]
async fn update_user_roles_by_user_id_org_id(
&self,
user_id: &str,
org_id: &id_type::OrganizationId,
update: storage::UserRoleUpdate,
version: enums::UserRoleVersion,
) -> CustomResult<Vec<storage::UserRole>, errors::StorageError> {
let conn = connection::pg_connection_write(self).await?;
storage::UserRole::update_by_user_id_org_id(
&conn,
user_id.to_owned(),
org_id.to_owned(),
update,
version,
)
.await
.map_err(|error| report!(errors::StorageError::from(error)))
}
#[instrument(skip_all)]
async fn list_user_roles_by_user_id(
&self,
@ -218,6 +161,30 @@ impl UserRoleInterface for Store {
.map_err(|error| report!(errors::StorageError::from(error)))
}
#[instrument(skip_all)]
async fn update_user_role_by_user_id_and_lineage(
&self,
user_id: &str,
org_id: &id_type::OrganizationId,
merchant_id: &id_type::MerchantId,
profile_id: Option<&String>,
update: storage::UserRoleUpdate,
version: enums::UserRoleVersion,
) -> CustomResult<storage::UserRole, errors::StorageError> {
let conn = connection::pg_connection_write(self).await?;
storage::UserRole::update_by_user_id_org_id_merchant_id_profile_id(
&conn,
user_id.to_owned(),
org_id.to_owned(),
merchant_id.to_owned(),
profile_id.cloned(),
update,
version,
)
.await
.map_err(|error| report!(errors::StorageError::from(error)))
}
#[instrument(skip_all)]
async fn delete_user_role_by_user_id_and_lineage(
&self,
@ -239,100 +206,6 @@ impl UserRoleInterface for Store {
.await
.map_err(|error| report!(errors::StorageError::from(error)))
}
#[instrument(skip_all)]
async fn transfer_org_ownership_between_users(
&self,
from_user_id: &str,
to_user_id: &str,
org_id: &id_type::OrganizationId,
version: enums::UserRoleVersion,
) -> CustomResult<(), errors::StorageError> {
let conn = connection::pg_connection_write(self)
.await
.change_context(errors::StorageError::DatabaseConnectionError)?;
conn.transaction_async(|conn| async move {
let old_org_admin_user_roles = storage::UserRole::update_by_user_id_org_id(
&conn,
from_user_id.to_owned(),
org_id.to_owned(),
storage::UserRoleUpdate::UpdateRole {
role_id: consts::user_role::ROLE_ID_MERCHANT_ADMIN.to_string(),
modified_by: from_user_id.to_owned(),
},
version,
)
.await
.map_err(|e| *e.current_context())?;
let new_org_admin_user_roles = storage::UserRole::update_by_user_id_org_id(
&conn,
to_user_id.to_owned(),
org_id.to_owned(),
storage::UserRoleUpdate::UpdateRole {
role_id: consts::user_role::ROLE_ID_ORGANIZATION_ADMIN.to_string(),
modified_by: from_user_id.to_owned(),
},
version,
)
.await
.map_err(|e| *e.current_context())?;
let new_org_admin_merchant_ids = new_org_admin_user_roles
.iter()
.map(|user_role| {
user_role
.merchant_id
.to_owned()
.ok_or(errors::DatabaseError::Others)
})
.collect::<Result<HashSet<_>, _>>()?;
let now = common_utils::date_time::now();
let mut missing_new_user_roles = Vec::new();
for old_role in old_org_admin_user_roles {
let Some(old_role_merchant_id) = &old_role.merchant_id else {
return Err(errors::DatabaseError::Others);
};
if !new_org_admin_merchant_ids.contains(old_role_merchant_id) {
missing_new_user_roles.push(storage::UserRoleNew {
user_id: to_user_id.to_string(),
merchant_id: Some(old_role_merchant_id.to_owned()),
role_id: consts::user_role::ROLE_ID_ORGANIZATION_ADMIN.to_string(),
org_id: Some(org_id.to_owned()),
status: enums::UserStatus::Active,
created_by: from_user_id.to_string(),
last_modified_by: from_user_id.to_string(),
created_at: now,
last_modified: now,
profile_id: None,
entity_id: None,
entity_type: None,
version: enums::UserRoleVersion::V1,
});
}
}
futures::future::try_join_all(missing_new_user_roles.into_iter().map(
|user_role| async {
user_role
.insert(&conn)
.await
.map_err(|e| *e.current_context())
},
))
.await?;
Ok::<_, errors::DatabaseError>(())
})
.await
.map_err(|error| report!(errors::StorageError::from(report!(error))))?;
Ok(())
}
}
#[async_trait::async_trait]
@ -422,185 +295,6 @@ impl UserRoleInterface for MockDb {
.into())
}
async fn update_user_role_by_user_id_merchant_id(
&self,
user_id: &str,
merchant_id: &id_type::MerchantId,
update: storage::UserRoleUpdate,
version: enums::UserRoleVersion,
) -> CustomResult<storage::UserRole, errors::StorageError> {
let mut user_roles = self.user_roles.lock().await;
for user_role in user_roles.iter_mut() {
let Some(user_role_merchant_id) = &user_role.merchant_id else {
return Err(errors::StorageError::DatabaseError(
report!(errors::DatabaseError::Others)
.attach_printable("merchant_id not found for user_role"),
)
.into());
};
if user_role.user_id == user_id
&& user_role_merchant_id == merchant_id
&& user_role.version == version
{
match &update {
storage::UserRoleUpdate::UpdateRole {
role_id,
modified_by,
} => {
user_role.role_id = role_id.to_string();
user_role.last_modified_by = modified_by.to_string();
}
storage::UserRoleUpdate::UpdateStatus {
status,
modified_by,
} => {
user_role.status = *status;
user_role.last_modified_by = modified_by.to_string();
}
};
return Ok(user_role.clone());
}
}
Err(errors::StorageError::ValueNotFound(format!(
"No user role available for user_id = {user_id} and merchant_id = {merchant_id:?}"
))
.into())
}
async fn update_user_roles_by_user_id_org_id(
&self,
user_id: &str,
org_id: &id_type::OrganizationId,
update: storage::UserRoleUpdate,
version: enums::UserRoleVersion,
) -> CustomResult<Vec<storage::UserRole>, errors::StorageError> {
let mut user_roles = self.user_roles.lock().await;
let mut updated_user_roles = Vec::new();
for user_role in user_roles.iter_mut() {
let Some(user_role_org_id) = &user_role.org_id else {
return Err(errors::StorageError::DatabaseError(
report!(errors::DatabaseError::Others)
.attach_printable("org_id not found for user_role"),
)
.into());
};
if user_role.user_id == user_id
&& user_role_org_id == org_id
&& user_role.version == version
{
match &update {
storage::UserRoleUpdate::UpdateRole {
role_id,
modified_by,
} => {
user_role.role_id = role_id.to_string();
user_role.last_modified_by = modified_by.to_string();
}
storage::UserRoleUpdate::UpdateStatus {
status,
modified_by,
} => {
status.clone_into(&mut user_role.status);
modified_by.clone_into(&mut user_role.last_modified_by);
}
}
updated_user_roles.push(user_role.to_owned());
}
}
if updated_user_roles.is_empty() {
Err(errors::StorageError::ValueNotFound(format!(
"No user role available for user_id = {user_id} and org_id = {org_id:?}"
))
.into())
} else {
Ok(updated_user_roles)
}
}
async fn transfer_org_ownership_between_users(
&self,
from_user_id: &str,
to_user_id: &str,
org_id: &id_type::OrganizationId,
version: enums::UserRoleVersion,
) -> CustomResult<(), errors::StorageError> {
let old_org_admin_user_roles = self
.update_user_roles_by_user_id_org_id(
from_user_id,
org_id,
storage::UserRoleUpdate::UpdateRole {
role_id: consts::user_role::ROLE_ID_MERCHANT_ADMIN.to_string(),
modified_by: from_user_id.to_string(),
},
version,
)
.await?;
let new_org_admin_user_roles = self
.update_user_roles_by_user_id_org_id(
to_user_id,
org_id,
storage::UserRoleUpdate::UpdateRole {
role_id: consts::user_role::ROLE_ID_ORGANIZATION_ADMIN.to_string(),
modified_by: from_user_id.to_string(),
},
version,
)
.await?;
let new_org_admin_merchant_ids = new_org_admin_user_roles
.iter()
.map(|user_role| {
user_role.merchant_id.to_owned().ok_or(report!(
errors::StorageError::DatabaseError(
report!(errors::DatabaseError::Others)
.attach_printable("merchant_id not found for user_role"),
)
))
})
.collect::<Result<HashSet<_>, _>>()?;
let now = common_utils::date_time::now();
let mut missing_new_user_roles = Vec::new();
for old_roles in old_org_admin_user_roles {
let Some(merchant_id) = &old_roles.merchant_id else {
return Err(errors::StorageError::DatabaseError(
report!(errors::DatabaseError::Others)
.attach_printable("merchant id not found for user role"),
)
.into());
};
if !new_org_admin_merchant_ids.contains(merchant_id) {
let new_user_role = storage::UserRoleNew {
user_id: to_user_id.to_string(),
merchant_id: Some(merchant_id.to_owned()),
role_id: consts::user_role::ROLE_ID_ORGANIZATION_ADMIN.to_string(),
org_id: Some(org_id.to_owned()),
status: enums::UserStatus::Active,
created_by: from_user_id.to_string(),
last_modified_by: from_user_id.to_string(),
created_at: now,
last_modified: now,
profile_id: None,
entity_id: None,
entity_type: None,
version: enums::UserRoleVersion::V1,
};
missing_new_user_roles.push(new_user_role);
}
}
for user_role in missing_new_user_roles {
self.insert_user_role(user_role).await?;
}
Ok(())
}
async fn list_user_roles_by_user_id(
&self,
user_id: &str,
@ -684,6 +378,60 @@ impl UserRoleInterface for MockDb {
.into())
}
async fn update_user_role_by_user_id_and_lineage(
&self,
user_id: &str,
org_id: &id_type::OrganizationId,
merchant_id: &id_type::MerchantId,
profile_id: Option<&String>,
update: storage::UserRoleUpdate,
version: enums::UserRoleVersion,
) -> CustomResult<storage::UserRole, errors::StorageError> {
let mut user_roles = self.user_roles.lock().await;
for user_role in user_roles.iter_mut() {
let org_level_check = user_role.org_id.as_ref() == Some(org_id)
&& user_role.merchant_id.is_none()
&& user_role.profile_id.is_none();
let merchant_level_check = user_role.org_id.as_ref() == Some(org_id)
&& user_role.merchant_id.as_ref() == Some(merchant_id)
&& user_role.profile_id.is_none();
let profile_level_check = user_role.org_id.as_ref() == Some(org_id)
&& user_role.merchant_id.as_ref() == Some(merchant_id)
&& user_role.profile_id.as_ref() == profile_id;
// Check if the user role matches the conditions and the version matches
if user_role.user_id == user_id
&& (org_level_check || merchant_level_check || profile_level_check)
&& user_role.version == version
{
match &update {
storage::UserRoleUpdate::UpdateRole {
role_id,
modified_by,
} => {
user_role.role_id = role_id.to_string();
user_role.last_modified_by = modified_by.to_string();
}
storage::UserRoleUpdate::UpdateStatus {
status,
modified_by,
} => {
user_role.status = *status;
user_role.last_modified_by = modified_by.to_string();
}
}
return Ok(user_role.clone());
}
}
Err(
errors::StorageError::ValueNotFound("Cannot find user role to update".to_string())
.into(),
)
}
async fn delete_user_role_by_user_id_and_lineage(
&self,
user_id: &str,

4
crates/router/src/routes/app.rs
View File

@ -1720,10 +1720,6 @@ impl User {
.route(web::put().to(accept_invitation)),
)
.service(web::resource("/update_role").route(web::post().to(update_user_role)))
.service(
web::resource("/transfer_ownership")
.route(web::post().to(transfer_org_ownership)),
)
.service(web::resource("/delete").route(web::delete().to(delete_user_role))),
);

19
crates/router/src/routes/user_role.rs
View File

@ -186,25 +186,6 @@ pub async fn update_user_role(
.await
}
pub async fn transfer_org_ownership(
state: web::Data<AppState>,
req: HttpRequest,
json_payload: web::Json<user_role_api::TransferOrgOwnershipRequest>,
) -> HttpResponse {
let flow = Flow::TransferOrgOwnership;
let payload = json_payload.into_inner();
Box::pin(api::server_wrap(
flow,
state.clone(),
&req,
payload,
user_role_core::transfer_org_ownership,
&auth::JWTAuth(Permission::UsersWrite),
api_locking::LockAction::NotApplicable,
))
.await
}
pub async fn accept_invitation(
state: web::Data<AppState>,
req: HttpRequest,

56
crates/router/src/utils/user_role.rs
View File

@ -3,9 +3,13 @@ use std::collections::HashSet;
use api_models::user_role as user_role_api;
use common_enums::PermissionGroup;
use common_utils::id_type;
use diesel_models::user_role::UserRole;
use error_stack::{report, ResultExt};
use diesel_models::{
enums::UserRoleVersion,
user_role::{UserRole, UserRoleUpdate},
};
use error_stack::{report, Report, ResultExt};
use router_env::logger;
use storage_impl::errors::StorageError;
use crate::{
consts,
@ -168,7 +172,53 @@ pub async fn get_multiple_role_info_for_user_roles(
.await
.to_not_found_response(UserErrors::InternalServerError)
.attach_printable("Role for user role doesn't exist")?;
Ok::<_, error_stack::Report<UserErrors>>(role)
Ok::<_, Report<UserErrors>>(role)
}))
.await
}
pub async fn update_v1_and_v2_user_roles_in_db(
state: &SessionState,
user_id: &str,
org_id: &id_type::OrganizationId,
merchant_id: &id_type::MerchantId,
profile_id: Option<&String>,
update: UserRoleUpdate,
) -> (
Result<UserRole, Report<StorageError>>,
Result<UserRole, Report<StorageError>>,
) {
let updated_v1_role = state
.store
.update_user_role_by_user_id_and_lineage(
user_id,
org_id,
merchant_id,
profile_id,
update.clone(),
UserRoleVersion::V1,
)
.await
.map_err(|e| {
logger::error!("Error updating user_role {e:?}");
e
});
let updated_v2_role = state
.store
.update_user_role_by_user_id_and_lineage(
user_id,
org_id,
merchant_id,
profile_id,
update,
UserRoleVersion::V2,
)
.await
.map_err(|e| {
logger::error!("Error updating user_role {e:?}");
e
});
(updated_v1_role, updated_v2_role)
}