Issue:
Bug: feat(user_role): Transfer ownership api





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 -->
This PR will add a new api to transfer ownership of an organization to other.

### 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).
-->
To allow users to change the ownership of an organization.

## 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)?
-->
Postman
```
curl --location 'http://localhost:8080/user/user/transfer_ownership' \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer JWT of org_admin user' \
--data-raw '{
    "email": "email of the user who is not org_admin"
}'
```
If the api is successful, it will return the following response.
```
{
    "token": "JWT",
    "merchant_id": "merchant_id",
    "name": "name of old org_admin",
    "email": "email of old org_admin",
    "verification_days_left": null,
    "user_role": "merchant_admin"
}
```

## 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:

Postman
```
curl --location 'http://localhost:8080/user/user/transfer_ownership' \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer JWT of org_admin user' \
--data-raw '{
    "email": "email of the user who is not org_admin"
}'
```
If the api is successful, it will return the following response.
```
{
    "token": "JWT",
    "merchant_id": "merchant_id",
    "name": "name of old org_admin",
    "email": "email of old org_admin",
    "verification_days_left": null,
    "user_role": "merchant_admin"
}
```

Patch:
diff --git a/crates/api_models/src/events/user_role.rs b/crates/api_models/src/events/user_role.rs
index 3ec30d6bd97..2b8d0221497 100644
--- a/crates/api_models/src/events/user_role.rs
+++ b/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
 );
diff --git a/crates/api_models/src/user_role.rs b/crates/api_models/src/user_role.rs
index 2672293390e..78df1d6823e 100644
--- a/crates/api_models/src/user_role.rs
+++ b/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,
+}
diff --git a/crates/diesel_models/src/query/user_role.rs b/crates/diesel_models/src/query/user_role.rs
index e67eba64c7c..5e759cf826b 100644
--- a/crates/diesel_models/src/query/user_role.rs
+++ b/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,
diff --git a/crates/router/src/core/user_role.rs b/crates/router/src/core/user_role.rs
index b48b39eea14..14be9bb6991 100644
--- a/crates/router/src/core/user_role.rs
+++ b/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,
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 029b1a57764..a5e5f216a8b 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/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]
diff --git a/crates/router/src/db/user_role.rs b/crates/router/src/db/user_role.rs
index f02e6d60b3b..12816fa006b 100644
--- a/crates/router/src/db/user_role.rs
+++ b/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,
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 651d3c0026f..65caa2b5c04 100644
--- a/crates/router/src/routes/app.rs
+++ b/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))),
         );
 
diff --git a/crates/router/src/routes/lock_utils.rs b/crates/router/src/routes/lock_utils.rs
index 1636ed3a764..02a45408baf 100644
--- a/crates/router/src/routes/lock_utils.rs
+++ b/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
diff --git a/crates/router/src/routes/user_role.rs b/crates/router/src/routes/user_role.rs
index ec05db1d615..f84c158332b 100644
--- a/crates/router/src/routes/user_role.rs
+++ b/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,
diff --git a/crates/router_env/src/logger/types.rs b/crates/router_env/src/logger/types.rs
index 4344acf89ff..2f4d48bea74 100644
--- a/crates/router_env/src/logger/types.rs
+++ b/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