feat(users): Add transfer org ownership API (#3603)

Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>

crates/api_models/src/events/user_role.rs

@@ -2,7 +2,7 @@ use common_utils::events::{ApiEventMetric, ApiEventsType};
 
 use crate::user_role::{
     AcceptInvitationRequest, AuthorizationInfoResponse, DeleteUserRoleRequest, GetRoleRequest,
-    ListRolesResponse, RoleInfoResponse, UpdateUserRoleRequest,
+    ListRolesResponse, RoleInfoResponse, TransferOrgOwnershipRequest, UpdateUserRoleRequest,
 };
 
 common_utils::impl_misc_api_event_type!(
@@ -12,5 +12,6 @@ common_utils::impl_misc_api_event_type!(
     AuthorizationInfoResponse,
     UpdateUserRoleRequest,
     AcceptInvitationRequest,
-    DeleteUserRoleRequest
+    DeleteUserRoleRequest,
+    TransferOrgOwnershipRequest
 );

crates/api_models/src/user_role.rs

@@ -108,3 +108,8 @@ pub type AcceptInvitationResponse = DashboardEntryResponse;
 pub struct DeleteUserRoleRequest {
     pub email: pii::Email,
 }
+
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
+pub struct TransferOrgOwnershipRequest {
+    pub email: pii::Email,
+}

crates/diesel_models/src/query/user_role.rs

@@ -54,6 +54,20 @@ impl UserRole {
         .await
     }
 
+    pub async fn update_by_user_id_org_id(
+        conn: &PgPooledConn,
+        user_id: String,
+        org_id: String,
+        update: UserRoleUpdate,
+    ) -> 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)),
+            UserRoleUpdateInternal::from(update),
+        )
+        .await
+    }
+
     pub async fn delete_by_user_id_merchant_id(
         conn: &PgPooledConn,
         user_id: String,

crates/router/src/core/user_role.rs

@@ -1,10 +1,11 @@
-use api_models::user_role as user_role_api;
+use api_models::{user as user_api, user_role as user_role_api};
 use diesel_models::{enums::UserStatus, user_role::UserRoleUpdate};
 use error_stack::ResultExt;
 use masking::ExposeInterface;
 use router_env::logger;
 
 use crate::{
+    consts,
     core::errors::{StorageErrorExt, UserErrors, UserResponse},
     routes::AppState,
     services::{
@@ -135,6 +136,55 @@ pub async fn update_user_role(
     Ok(ApplicationResponse::StatusOk)
 }
 
+pub async fn transfer_org_ownership(
+    state: AppState,
+    user_from_token: auth::UserFromToken,
+    req: user_role_api::TransferOrgOwnershipRequest,
+) -> UserResponse<user_api::DashboardEntryResponse> {
+    if user_from_token.role_id != consts::user_role::ROLE_ID_ORGANIZATION_ADMIN {
+        return Err(UserErrors::InvalidRoleOperation.into()).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(UserErrors::InvalidRoleOperation.into())
+            .attach_printable("User transferring ownership to themselves".to_string());
+    }
+
+    state
+        .store
+        .transfer_org_ownership_between_users(
+            &user_from_token.user_id,
+            user_to_be_updated.get_user_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?;
+
+    let user_from_db = domain::UserFromStorage::from(user_from_token.get_user(&state).await?);
+    let user_role = user_from_db
+        .get_role_from_db_by_merchant_id(&state, &user_from_token.merchant_id)
+        .await
+        .to_not_found_response(UserErrors::InvalidRoleOperation)?;
+
+    let token = utils::user::generate_jwt_auth_token(&state, &user_from_db, &user_role).await?;
+
+    Ok(ApplicationResponse::Json(
+        utils::user::get_dashboard_entry_response(&state, user_from_db, user_role, token)?,
+    ))
+}
+
 pub async fn accept_invitation(
     state: AppState,
     user_token: auth::UserWithoutMerchantFromToken,

crates/router/src/db/kafka_store.rs

@@ -1965,6 +1965,17 @@ impl UserRoleInterface for KafkaStore {
             .await
     }
 
+    async fn update_user_roles_by_user_id_org_id(
+        &self,
+        user_id: &str,
+        org_id: &str,
+        update: user_storage::UserRoleUpdate,
+    ) -> CustomResult<Vec<user_storage::UserRole>, errors::StorageError> {
+        self.diesel_store
+            .update_user_roles_by_user_id_org_id(user_id, org_id, update)
+            .await
+    }
+
     async fn delete_user_role_by_user_id_merchant_id(
         &self,
         user_id: &str,
@@ -1981,6 +1992,17 @@ impl UserRoleInterface for KafkaStore {
     ) -> CustomResult<Vec<user_storage::UserRole>, errors::StorageError> {
         self.diesel_store.list_user_roles_by_user_id(user_id).await
     }
+
+    async fn transfer_org_ownership_between_users(
+        &self,
+        from_user_id: &str,
+        to_user_id: &str,
+        org_id: &str,
+    ) -> CustomResult<(), errors::StorageError> {
+        self.diesel_store
+            .transfer_org_ownership_between_users(from_user_id, to_user_id, org_id)
+            .await
+    }
 }
 
 #[async_trait::async_trait]

crates/router/src/db/user_role.rs

@@ -1,9 +1,12 @@
-use diesel_models::user_role as storage;
+use std::{collections::HashSet, ops::Not};
+
+use async_bb8_diesel::AsyncConnection;
+use diesel_models::{enums, user_role as storage};
 use error_stack::{IntoReport, ResultExt};
 
 use super::MockDb;
 use crate::{
-    connection,
+    connection, consts,
     core::errors::{self, CustomResult},
     services::Store,
 };
@@ -32,6 +35,14 @@ pub trait UserRoleInterface {
         merchant_id: &str,
         update: storage::UserRoleUpdate,
     ) -> CustomResult<storage::UserRole, errors::StorageError>;
+
+    async fn update_user_roles_by_user_id_org_id(
+        &self,
+        user_id: &str,
+        org_id: &str,
+        update: storage::UserRoleUpdate,
+    ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError>;
+
     async fn delete_user_role_by_user_id_merchant_id(
         &self,
         user_id: &str,
@@ -42,6 +53,13 @@ pub trait UserRoleInterface {
         &self,
         user_id: &str,
     ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError>;
+
+    async fn transfer_org_ownership_between_users(
+        &self,
+        from_user_id: &str,
+        to_user_id: &str,
+        org_id: &str,
+    ) -> CustomResult<(), errors::StorageError>;
 }
 
 #[async_trait::async_trait]
@@ -103,6 +121,24 @@ impl UserRoleInterface for Store {
         .into_report()
     }
 
+    async fn update_user_roles_by_user_id_org_id(
+        &self,
+        user_id: &str,
+        org_id: &str,
+        update: storage::UserRoleUpdate,
+    ) -> 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,
+        )
+        .await
+        .map_err(Into::into)
+        .into_report()
+    }
+
     async fn delete_user_role_by_user_id_merchant_id(
         &self,
         user_id: &str,
@@ -129,6 +165,86 @@ impl UserRoleInterface for Store {
             .map_err(Into::into)
             .into_report()
     }
+
+    async fn transfer_org_ownership_between_users(
+        &self,
+        from_user_id: &str,
+        to_user_id: &str,
+        org_id: &str,
+    ) -> 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(),
+                },
+            )
+            .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(),
+                },
+            )
+            .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())
+                .collect::<HashSet<String>>();
+
+            let now = common_utils::date_time::now();
+
+            let missing_new_user_roles =
+                old_org_admin_user_roles.into_iter().filter_map(|old_role| {
+                    new_org_admin_merchant_ids
+                        .contains(&old_role.merchant_id)
+                        .not()
+                        .then_some({
+                            storage::UserRoleNew {
+                                user_id: to_user_id.to_string(),
+                                merchant_id: old_role.merchant_id,
+                                role_id: consts::user_role::ROLE_ID_ORGANIZATION_ADMIN.to_string(),
+                                org_id: org_id.to_string(),
+                                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,
+                            }
+                        })
+                });
+
+            futures::future::try_join_all(missing_new_user_roles.map(|user_role| async {
+                user_role
+                    .insert(&conn)
+                    .await
+                    .map_err(|e| *e.current_context())
+            }))
+            .await?;
+
+            Ok::<_, errors::DatabaseError>(())
+        })
+        .await
+        .into_report()
+        .map_err(Into::into)
+        .into_report()?;
+
+        Ok(())
+    }
 }
 
 #[async_trait::async_trait]
@@ -241,6 +357,107 @@ impl UserRoleInterface for MockDb {
             )
     }
 
+    async fn update_user_roles_by_user_id_org_id(
+        &self,
+        user_id: &str,
+        org_id: &str,
+        update: storage::UserRoleUpdate,
+    ) -> 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() {
+            if user_role.user_id == user_id && user_role.org_id == org_id {
+                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.to_owned();
+                        user_role.last_modified_by = modified_by.to_owned();
+                    }
+                }
+                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: &str,
+    ) -> 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(),
+                },
+            )
+            .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(),
+                },
+            )
+            .await?;
+
+        let new_org_admin_merchant_ids = new_org_admin_user_roles
+            .iter()
+            .map(|user_role| user_role.merchant_id.to_owned())
+            .collect::<HashSet<String>>();
+
+        let now = common_utils::date_time::now();
+
+        let missing_new_user_roles = old_org_admin_user_roles
+            .into_iter()
+            .filter_map(|old_roles| {
+                if !new_org_admin_merchant_ids.contains(&old_roles.merchant_id) {
+                    Some(storage::UserRoleNew {
+                        user_id: to_user_id.to_string(),
+                        merchant_id: old_roles.merchant_id,
+                        role_id: consts::user_role::ROLE_ID_ORGANIZATION_ADMIN.to_string(),
+                        org_id: org_id.to_string(),
+                        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,
+                    })
+                } else {
+                    None
+                }
+            });
+
+        for user_role in missing_new_user_roles {
+            self.insert_user_role(user_role).await?;
+        }
+
+        Ok(())
+    }
+
     async fn delete_user_role_by_user_id_merchant_id(
         &self,
         user_id: &str,

crates/router/src/routes/app.rs

@@ -1026,6 +1026,10 @@ impl User {
                 )
                 .service(web::resource("/invite/accept").route(web::post().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))),
         );
 

crates/router/src/routes/lock_utils.rs

@@ -196,7 +196,8 @@ impl From<Flow> for ApiIdentifier {
             | Flow::UpdateUserRole
             | Flow::GetAuthorizationInfo
             | Flow::AcceptInvitation
-            | Flow::DeleteUserRole => Self::UserRole,
+            | Flow::DeleteUserRole
+            | Flow::TransferOrgOwnership => Self::UserRole,
 
             Flow::GetActionUrl | Flow::SyncOnboardingStatus | Flow::ResetTrackingId => {
                 Self::ConnectorOnboarding

crates/router/src/routes/user_role.rs

@@ -97,6 +97,25 @@ 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,

crates/router_env/src/logger/types.rs

@@ -311,6 +311,8 @@ pub enum Flow {
     GetRoleFromToken,
     /// Update user role
     UpdateUserRole,
+    /// Transfer organization ownership
+    TransferOrgOwnership,
     /// Create merchant account for user in a org
     UserMerchantAccountCreate,
     /// Generate Sample Data