Issue:
Bug: feat(user_roles): Start inserting V2 user_roles in DB



Currently we only have V1 user_roles data. These roles are at merchant level.

To start supporting profile level dashboard, we should start inserting V2 user_roles which will have profile context as well.

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 -->
- From this PR, we will start inserting V2 user_roles.

### Additional Changes

- [ ] This PR modifies the API contract
- [x] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->
Closes #5543.

## 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)?
-->
All the following APIs will insert both V1 and V2 roles in DB.
```curl
curl --location 'https://integ-api.hyperswitch.io/user/signup' \
--header 'Content-Type: application/json' \
--data-raw '{
    "email": "new email",
    "password": "password"
}'
```
```curl
curl --location 'https://integ-api.hyperswitch.io/user/signup?token_only=true' \
--header 'Content-Type: application/json' \
--data '{
    "email": "new email",
    "password": "password"
}'
```
```
curl --location 'https://integ-api.hyperswitch.io/user/connect_account' \
--header 'Content-Type: application/json' \
--data-raw '{
    "email": "new email"
}'
```
```
curl --location 'https://integ-api.hyperswitch.io/user/internal_signup' \
--header 'api-key: test_admin' \
--header 'Content-Type: application/json' \
--data '{
    "name" : "name",
    "email" : "new email",
    "password": "password"
}'
```
```
curl --location 'https://integ-api.hyperswitch.io/user/user/invite_multiple' \
--header 'Content-Type: application/json' \
--data-raw '[
    {
        "email": "email",
        "name": "name",
        "role_id": "merchant_view_only"
    },
    {
        "email": "email 2",
        "name": "name",
        "role_id": "merchant_admin"
    }
]'
```
This will only insert V1 role in DB.
```
curl --location 'https://integ-api.hyperswitch.io/user/create_merchant' \
--header 'Content-Type: application/json' \
--header 'Authorization: ••••••' \
--data '{
    "company_name": "Juspay"
}'
```


## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible


Testing:

All the following APIs will insert both V1 and V2 roles in DB.
```curl
curl --location 'https://integ-api.hyperswitch.io/user/signup' \
--header 'Content-Type: application/json' \
--data-raw '{
    "email": "new email",
    "password": "password"
}'
```
```curl
curl --location 'https://integ-api.hyperswitch.io/user/signup?token_only=true' \
--header 'Content-Type: application/json' \
--data '{
    "email": "new email",
    "password": "password"
}'
```
```
curl --location 'https://integ-api.hyperswitch.io/user/connect_account' \
--header 'Content-Type: application/json' \
--data-raw '{
    "email": "new email"
}'
```
```
curl --location 'https://integ-api.hyperswitch.io/user/internal_signup' \
--header 'api-key: test_admin' \
--header 'Content-Type: application/json' \
--data '{
    "name" : "name",
    "email" : "new email",
    "password": "password"
}'
```
```
curl --location 'https://integ-api.hyperswitch.io/user/user/invite_multiple' \
--header 'Content-Type: application/json' \
--data-raw '[
    {
        "email": "email",
        "name": "name",
        "role_id": "merchant_view_only"
    },
    {
        "email": "email 2",
        "name": "name",
        "role_id": "merchant_admin"
    }
]'
```
This will only insert V1 role in DB.
```
curl --location 'https://integ-api.hyperswitch.io/user/create_merchant' \
--header 'Content-Type: application/json' \
--header 'Authorization: ••••••' \
--data '{
    "company_name": "Juspay"
}'
```

Patch:
diff --git a/crates/diesel_models/src/query/user_role.rs b/crates/diesel_models/src/query/user_role.rs
index 85f866f2fc2..fde146408a9 100644
--- a/crates/diesel_models/src/query/user_role.rs
+++ b/crates/diesel_models/src/query/user_role.rs
@@ -5,6 +5,7 @@ use diesel::{
     BoolExpressionMethods, ExpressionMethods, QueryDsl,
 };
 use error_stack::{report, ResultExt};
+use router_env::logger;
 
 use crate::{
     enums::UserRoleVersion, errors, query::generics, schema::user_roles::dsl, user_role::*,
@@ -18,6 +19,21 @@ impl UserRoleNew {
 }
 
 impl UserRole {
+    pub async fn insert_multiple_user_roles(
+        conn: &PgPooledConn,
+        user_roles: Vec<UserRoleNew>,
+    ) -> StorageResult<Vec<Self>> {
+        let query = diesel::insert_into(<Self>::table()).values(user_roles);
+
+        logger::debug!(query = %debug_query::<Pg,_>(&query).to_string());
+
+        query
+            .get_results_async(conn)
+            .await
+            .change_context(errors::DatabaseError::Others)
+            .attach_printable("Error while inserting user_roles")
+    }
+
     pub async fn find_by_user_id(
         conn: &PgPooledConn,
         user_id: String,
diff --git a/crates/router/src/core/user.rs b/crates/router/src/core/user.rs
index 17ebd00a994..ace8c3babc6 100644
--- a/crates/router/src/core/user.rs
+++ b/crates/router/src/core/user.rs
@@ -1,4 +1,7 @@
-use std::collections::{HashMap, HashSet};
+use std::{
+    collections::{HashMap, HashSet},
+    ops::Not,
+};
 
 use api_models::{
     payments::RedirectionResponse,
@@ -13,7 +16,6 @@ use diesel_models::{
     organization::OrganizationBridge,
     user as storage_user,
     user_authentication_method::{UserAuthenticationMethodNew, UserAuthenticationMethodUpdate},
-    user_role::UserRoleNew,
 };
 use error_stack::{report, ResultExt};
 #[cfg(feature = "email")]
@@ -60,10 +62,11 @@ pub async fn signup_with_merchant_id(
         .await?;
 
     let user_role = new_user
-        .insert_user_role_in_db(
+        .insert_org_level_user_role_in_db(
             state.clone(),
             common_utils::consts::ROLE_ID_ORGANIZATION_ADMIN.to_string(),
             UserStatus::Active,
+            None,
         )
         .await?;
 
@@ -132,10 +135,11 @@ pub async fn signup(
         .insert_user_and_merchant_in_db(state.clone())
         .await?;
     let user_role = new_user
-        .insert_user_role_in_db(
+        .insert_org_level_user_role_in_db(
             state.clone(),
             common_utils::consts::ROLE_ID_ORGANIZATION_ADMIN.to_string(),
             UserStatus::Active,
+            None,
         )
         .await?;
     utils::user_role::set_role_permissions_in_cache_by_user_role(&state, &user_role).await;
@@ -163,10 +167,11 @@ pub async fn signup_token_only_flow(
         .insert_user_and_merchant_in_db(state.clone())
         .await?;
     let user_role = new_user
-        .insert_user_role_in_db(
+        .insert_org_level_user_role_in_db(
             state.clone(),
             common_utils::consts::ROLE_ID_ORGANIZATION_ADMIN.to_string(),
             UserStatus::Active,
+            None,
         )
         .await?;
 
@@ -314,10 +319,11 @@ pub async fn connect_account(
             .insert_user_and_merchant_in_db(state.clone())
             .await?;
         let user_role = new_user
-            .insert_user_role_in_db(
+            .insert_org_level_user_role_in_db(
                 state.clone(),
                 common_utils::consts::ROLE_ID_ORGANIZATION_ADMIN.to_string(),
                 UserStatus::Active,
+                None,
             )
             .await?;
 
@@ -773,37 +779,60 @@ async fn handle_existing_user_invitation(
     auth_id: &Option<String>,
 ) -> UserResult<InviteMultipleUserResponse> {
     let now = common_utils::date_time::now();
-    state
+
+    if state
         .store
-        .insert_user_role(UserRoleNew {
-            user_id: invitee_user_from_db.get_user_id().to_owned(),
-            merchant_id: Some(user_from_token.merchant_id.clone()),
-            role_id: request.role_id.clone(),
-            org_id: Some(user_from_token.org_id.clone()),
-            status: {
-                if cfg!(feature = "email") {
-                    UserStatus::InvitationSent
-                } else {
-                    UserStatus::Active
-                }
-            },
-            created_by: user_from_token.user_id.clone(),
-            last_modified_by: user_from_token.user_id.clone(),
-            created_at: now,
-            last_modified: now,
-            profile_id: None,
-            entity_id: None,
-            entity_type: None,
-            version: UserRoleVersion::V1,
-        })
+        .find_user_role_by_user_id_and_lineage(
+            invitee_user_from_db.get_user_id(),
+            &user_from_token.org_id,
+            &user_from_token.merchant_id,
+            user_from_token.profile_id.as_ref(),
+            UserRoleVersion::V1,
+        )
         .await
-        .map_err(|e| {
-            if e.current_context().is_db_unique_violation() {
-                e.change_context(UserErrors::UserExists)
+        .is_err_and(|err| err.current_context().is_db_not_found())
+        .not()
+    {
+        return Err(UserErrors::UserExists.into());
+    }
+
+    if state
+        .store
+        .find_user_role_by_user_id_and_lineage(
+            invitee_user_from_db.get_user_id(),
+            &user_from_token.org_id,
+            &user_from_token.merchant_id,
+            user_from_token.profile_id.as_ref(),
+            UserRoleVersion::V2,
+        )
+        .await
+        .is_err_and(|err| err.current_context().is_db_not_found())
+        .not()
+    {
+        return Err(UserErrors::UserExists.into());
+    }
+
+    let user_role = domain::NewUserRole {
+        user_id: invitee_user_from_db.get_user_id().to_owned(),
+        role_id: request.role_id.clone(),
+        status: {
+            if cfg!(feature = "email") {
+                UserStatus::InvitationSent
             } else {
-                e.change_context(UserErrors::InternalServerError)
+                UserStatus::Active
             }
-        })?;
+        },
+        created_by: user_from_token.user_id.clone(),
+        last_modified_by: user_from_token.user_id.clone(),
+        created_at: now,
+        last_modified: now,
+        entity: domain::MerchantLevel {
+            org_id: user_from_token.org_id.clone(),
+            merchant_id: user_from_token.merchant_id.clone(),
+        },
+    }
+    .insert_in_v1_and_v2(state)
+    .await?;
 
     let is_email_sent;
     #[cfg(feature = "email")]
@@ -865,31 +894,22 @@ async fn handle_new_user_invitation(
     };
 
     let now = common_utils::date_time::now();
-    state
-        .store
-        .insert_user_role(UserRoleNew {
-            user_id: new_user.get_user_id().to_owned(),
-            merchant_id: Some(user_from_token.merchant_id.clone()),
-            role_id: request.role_id.clone(),
-            org_id: Some(user_from_token.org_id.clone()),
-            status: invitation_status,
-            created_by: user_from_token.user_id.clone(),
-            last_modified_by: user_from_token.user_id.clone(),
-            created_at: now,
-            last_modified: now,
-            profile_id: None,
-            entity_id: None,
-            entity_type: None,
-            version: UserRoleVersion::V1,
-        })
-        .await
-        .map_err(|e| {
-            if e.current_context().is_db_unique_violation() {
-                e.change_context(UserErrors::UserExists)
-            } else {
-                e.change_context(UserErrors::InternalServerError)
-            }
-        })?;
+
+    let user_role = domain::NewUserRole {
+        user_id: new_user.get_user_id().to_owned(),
+        role_id: request.role_id.clone(),
+        status: invitation_status,
+        created_by: user_from_token.user_id.clone(),
+        last_modified_by: user_from_token.user_id.clone(),
+        created_at: now,
+        last_modified: now,
+        entity: domain::MerchantLevel {
+            merchant_id: user_from_token.merchant_id.clone(),
+            org_id: user_from_token.org_id.clone(),
+        },
+    }
+    .insert_in_v1_and_v2(state)
+    .await?;
 
     let is_email_sent;
     // TODO: Adding this to avoid clippy lints, remove this once the token only flow is being used
@@ -1289,7 +1309,7 @@ pub async fn create_internal_user(
             }
         })?;
 
-    let new_user = domain::NewUser::try_from((request, internal_merchant.organization_id))?;
+    let new_user = domain::NewUser::try_from((request, internal_merchant.organization_id.clone()))?;
 
     let mut store_user: storage_user::UserNew = new_user.clone().try_into()?;
     store_user.set_is_verified(true);
@@ -1308,12 +1328,16 @@ pub async fn create_internal_user(
         .map(domain::user::UserFromStorage::from)?;
 
     new_user
-        .insert_user_role_in_db(
-            state,
+        .get_no_level_user_role(
             common_utils::consts::ROLE_ID_INTERNAL_VIEW_ONLY_USER.to_string(),
             UserStatus::Active,
         )
-        .await?;
+        .add_entity(domain::InternalLevel {
+            org_id: internal_merchant.organization_id,
+        })
+        .insert_in_v1_and_v2(&state)
+        .await
+        .change_context(UserErrors::InternalServerError)?;
 
     Ok(ApplicationResponse::StatusOk)
 }
@@ -1448,10 +1472,11 @@ pub async fn create_merchant_account(
         .await?;
 
     let role_insertion_res = new_user
-        .insert_user_role_in_db(
+        .insert_org_level_user_role_in_db(
             state.clone(),
             common_utils::consts::ROLE_ID_ORGANIZATION_ADMIN.to_string(),
             UserStatus::Active,
+            Some(UserRoleVersion::V1),
         )
         .await;
     if let Err(e) = role_insertion_res {
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 0d8b753463b..3a7a2c44f31 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -35,7 +35,7 @@ use super::{
     user::{sample_data::BatchSampleDataInterface, UserInterface},
     user_authentication_method::UserAuthenticationMethodInterface,
     user_key_store::UserKeyStoreInterface,
-    user_role::{ListUserRolesByOrgIdPayload, UserRoleInterface},
+    user_role::{InsertUserRolePayload, ListUserRolesByOrgIdPayload, UserRoleInterface},
 };
 #[cfg(feature = "payouts")]
 use crate::services::kafka::payout::KafkaPayout;
@@ -2766,8 +2766,8 @@ impl RedisConnInterface for KafkaStore {
 impl UserRoleInterface for KafkaStore {
     async fn insert_user_role(
         &self,
-        user_role: user_storage::UserRoleNew,
-    ) -> CustomResult<user_storage::UserRole, errors::StorageError> {
+        user_role: InsertUserRolePayload,
+    ) -> CustomResult<Vec<user_storage::UserRole>, errors::StorageError> {
         self.diesel_store.insert_user_role(user_role).await
     }
 
diff --git a/crates/router/src/db/user_role.rs b/crates/router/src/db/user_role.rs
index 8d07f53e12d..57090632725 100644
--- a/crates/router/src/db/user_role.rs
+++ b/crates/router/src/db/user_role.rs
@@ -10,12 +10,35 @@ use crate::{
     services::Store,
 };
 
+pub enum InsertUserRolePayload {
+    OnlyV1(storage::UserRoleNew),
+    OnlyV2(storage::UserRoleNew),
+    V1AndV2(Box<[storage::UserRoleNew; 2]>),
+}
+
+impl InsertUserRolePayload {
+    fn convert_to_vec(self) -> Vec<storage::UserRoleNew> {
+        match self {
+            Self::OnlyV1(user_role) | Self::OnlyV2(user_role) => vec![user_role],
+            Self::V1AndV2(user_roles) => user_roles.to_vec(),
+        }
+    }
+}
+
+pub struct ListUserRolesByOrgIdPayload<'a> {
+    pub user_id: Option<&'a String>,
+    pub org_id: &'a id_type::OrganizationId,
+    pub merchant_id: Option<&'a id_type::MerchantId>,
+    pub profile_id: Option<&'a id_type::ProfileId>,
+    pub version: Option<enums::UserRoleVersion>,
+}
+
 #[async_trait::async_trait]
 pub trait UserRoleInterface {
     async fn insert_user_role(
         &self,
-        user_role: storage::UserRoleNew,
-    ) -> CustomResult<storage::UserRole, errors::StorageError>;
+        user_role: InsertUserRolePayload,
+    ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError>;
 
     async fn find_user_role_by_user_id(
         &self,
@@ -86,24 +109,16 @@ pub trait UserRoleInterface {
     ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError>;
 }
 
-pub struct ListUserRolesByOrgIdPayload<'a> {
-    pub user_id: Option<&'a String>,
-    pub org_id: &'a id_type::OrganizationId,
-    pub merchant_id: Option<&'a id_type::MerchantId>,
-    pub profile_id: Option<&'a id_type::ProfileId>,
-    pub version: Option<enums::UserRoleVersion>,
-}
-
 #[async_trait::async_trait]
 impl UserRoleInterface for Store {
     #[instrument(skip_all)]
     async fn insert_user_role(
         &self,
-        user_role: storage::UserRoleNew,
-    ) -> CustomResult<storage::UserRole, errors::StorageError> {
+        user_role: InsertUserRolePayload,
+    ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError> {
         let conn = connection::pg_connection_write(self).await?;
-        user_role
-            .insert(&conn)
+
+        storage::UserRole::insert_multiple_user_roles(&conn, user_role.convert_to_vec())
             .await
             .map_err(|error| report!(errors::StorageError::from(error)))
     }
@@ -138,7 +153,6 @@ impl UserRoleInterface for Store {
         .map_err(|error| report!(errors::StorageError::from(error)))
     }
 
-    #[instrument(skip_all)]
     async fn list_user_roles_by_user_id_and_version(
         &self,
         user_id: &str,
@@ -275,37 +289,44 @@ impl UserRoleInterface for Store {
 impl UserRoleInterface for MockDb {
     async fn insert_user_role(
         &self,
-        user_role: storage::UserRoleNew,
-    ) -> CustomResult<storage::UserRole, errors::StorageError> {
-        let mut user_roles = self.user_roles.lock().await;
-        if user_roles
-            .iter()
-            .any(|user_role_inner| user_role_inner.user_id == user_role.user_id)
-        {
-            Err(errors::StorageError::DuplicateValue {
-                entity: "user_id",
-                key: None,
-            })?
-        }
-        let user_role = storage::UserRole {
-            id: i32::try_from(user_roles.len())
-                .change_context(errors::StorageError::MockDbError)?,
-            user_id: user_role.user_id,
-            merchant_id: user_role.merchant_id,
-            role_id: user_role.role_id,
-            status: user_role.status,
-            created_by: user_role.created_by,
-            created_at: user_role.created_at,
-            last_modified: user_role.last_modified,
-            last_modified_by: user_role.last_modified_by,
-            org_id: user_role.org_id,
-            profile_id: None,
-            entity_id: None,
-            entity_type: None,
-            version: enums::UserRoleVersion::V1,
-        };
-        user_roles.push(user_role.clone());
-        Ok(user_role)
+        user_role: InsertUserRolePayload,
+    ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError> {
+        let mut db_user_roles = self.user_roles.lock().await;
+
+        user_role
+            .convert_to_vec()
+            .into_iter()
+            .map(|user_role| {
+                if db_user_roles
+                    .iter()
+                    .any(|user_role_inner| user_role_inner.user_id == user_role.user_id)
+                {
+                    Err(errors::StorageError::DuplicateValue {
+                        entity: "user_id",
+                        key: None,
+                    })?
+                }
+                let user_role = storage::UserRole {
+                    id: i32::try_from(db_user_roles.len())
+                        .change_context(errors::StorageError::MockDbError)?,
+                    user_id: user_role.user_id,
+                    merchant_id: user_role.merchant_id,
+                    role_id: user_role.role_id,
+                    status: user_role.status,
+                    created_by: user_role.created_by,
+                    created_at: user_role.created_at,
+                    last_modified: user_role.last_modified,
+                    last_modified_by: user_role.last_modified_by,
+                    org_id: user_role.org_id,
+                    profile_id: None,
+                    entity_id: None,
+                    entity_type: None,
+                    version: enums::UserRoleVersion::V1,
+                };
+                db_user_roles.push(user_role.clone());
+                Ok(user_role)
+            })
+            .collect::<Result<Vec<_>, _>>()
     }
 
     async fn find_user_role_by_user_id(
diff --git a/crates/router/src/types/domain/user.rs b/crates/router/src/types/domain/user.rs
index b252e618536..42cb71458d5 100644
--- a/crates/router/src/types/domain/user.rs
+++ b/crates/router/src/types/domain/user.rs
@@ -3,7 +3,7 @@ use std::{collections::HashSet, ops, str::FromStr};
 use api_models::{
     admin as admin_api, organization as api_org, user as user_api, user_role as user_role_api,
 };
-use common_enums::TokenPurpose;
+use common_enums::{EntityType, TokenPurpose};
 use common_utils::{
     crypto::Encryptable, errors::CustomResult, id_type, new_type::MerchantName, pii, type_name,
     types::keymanager::Identifier,
@@ -19,6 +19,7 @@ use masking::{ExposeInterface, PeekInterface, Secret};
 use once_cell::sync::Lazy;
 use rand::distributions::{Alphanumeric, DistString};
 use router_env::env;
+use time::PrimitiveDateTime;
 use unicode_segmentation::UnicodeSegmentation;
 #[cfg(feature = "keymanager_create")]
 use {base64::Engine, common_utils::types::keymanager::EncryptionTransferRequest};
@@ -29,9 +30,13 @@ use crate::{
         admin,
         errors::{self, UserErrors, UserResult},
     },
-    db::GlobalStorageInterface,
+    db::{user_role::InsertUserRolePayload, GlobalStorageInterface},
     routes::SessionState,
-    services::{self, authentication as auth, authentication::UserFromToken, authorization::info},
+    services::{
+        self,
+        authentication::{self as auth, UserFromToken},
+        authorization::info,
+    },
     types::transformers::ForeignFrom,
     utils::{self, user::password},
 };
@@ -638,38 +643,51 @@ impl NewUser {
         created_user
     }
 
-    pub async fn insert_user_role_in_db(
+    pub fn get_no_level_user_role(
         self,
-        state: SessionState,
         role_id: String,
         user_status: UserStatus,
-    ) -> UserResult<UserRole> {
+    ) -> NewUserRole<NoLevel> {
         let now = common_utils::date_time::now();
         let user_id = self.get_user_id();
 
-        state
-            .store
-            .insert_user_role(UserRoleNew {
-                merchant_id: Some(self.get_new_merchant().get_merchant_id()),
-                status: user_status,
-                created_by: user_id.clone(),
-                last_modified_by: user_id.clone(),
-                user_id,
-                role_id,
-                created_at: now,
-                last_modified: now,
-                org_id: Some(
-                    self.get_new_merchant()
-                        .get_new_organization()
-                        .get_organization_id(),
-                ),
-                profile_id: None,
-                entity_id: None,
-                entity_type: None,
-                version: UserRoleVersion::V1,
-            })
-            .await
-            .change_context(UserErrors::InternalServerError)
+        NewUserRole {
+            status: user_status,
+            created_by: user_id.clone(),
+            last_modified_by: user_id.clone(),
+            user_id,
+            role_id,
+            created_at: now,
+            last_modified: now,
+            entity: NoLevel,
+        }
+    }
+
+    pub async fn insert_org_level_user_role_in_db(
+        self,
+        state: SessionState,
+        role_id: String,
+        user_status: UserStatus,
+        version: Option<UserRoleVersion>,
+    ) -> UserResult<UserRole> {
+        let org_id = self
+            .get_new_merchant()
+            .get_new_organization()
+            .get_organization_id();
+        let merchant_id = self.get_new_merchant().get_merchant_id();
+
+        let org_user_role = self
+            .get_no_level_user_role(role_id, user_status)
+            .add_entity(OrganizationLevel {
+                org_id,
+                merchant_id,
+            });
+
+        match version {
+            Some(UserRoleVersion::V1) => org_user_role.insert_in_v1(&state).await,
+            Some(UserRoleVersion::V2) => org_user_role.insert_in_v2(&state).await,
+            None => org_user_role.insert_in_v1_and_v2(&state).await,
+        }
     }
 }
 
@@ -1282,3 +1300,252 @@ impl RecoveryCodes {
         self.0
     }
 }
+
+// This is for easier construction
+#[derive(Clone)]
+pub struct NoLevel;
+
+#[derive(Clone)]
+pub struct OrganizationLevel {
+    pub org_id: id_type::OrganizationId,
+    // Keeping this to allow insertion of org_admins in V1
+    pub merchant_id: id_type::MerchantId,
+}
+
+#[derive(Clone)]
+pub struct MerchantLevel {
+    pub org_id: id_type::OrganizationId,
+    pub merchant_id: id_type::MerchantId,
+}
+
+#[derive(Clone)]
+pub struct ProfileLevel {
+    pub org_id: id_type::OrganizationId,
+    pub merchant_id: id_type::MerchantId,
+    pub profile_id: id_type::ProfileId,
+}
+
+#[derive(Clone)]
+pub struct InternalLevel {
+    pub org_id: id_type::OrganizationId,
+}
+
+#[derive(Clone)]
+pub struct NewUserRole<E: Clone> {
+    pub user_id: String,
+    pub role_id: String,
+    pub status: UserStatus,
+    pub created_by: String,
+    pub last_modified_by: String,
+    pub created_at: PrimitiveDateTime,
+    pub last_modified: PrimitiveDateTime,
+    pub entity: E,
+}
+
+impl NewUserRole<NoLevel> {
+    pub fn add_entity<T>(self, entity: T) -> NewUserRole<T>
+    where
+        T: Clone,
+    {
+        NewUserRole {
+            entity,
+            user_id: self.user_id,
+            role_id: self.role_id,
+            status: self.status,
+            created_by: self.created_by,
+            last_modified_by: self.last_modified_by,
+            created_at: self.created_at,
+            last_modified: self.last_modified,
+        }
+    }
+}
+
+pub struct EntityInfo {
+    org_id: id_type::OrganizationId,
+    merchant_id: Option<id_type::MerchantId>,
+    profile_id: Option<id_type::ProfileId>,
+    entity_id: String,
+    entity_type: EntityType,
+}
+
+impl<E> NewUserRole<E>
+where
+    E: Clone,
+{
+    fn convert_to_new_v1_role(
+        self,
+        org_id: id_type::OrganizationId,
+        merchant_id: id_type::MerchantId,
+    ) -> UserRoleNew {
+        UserRoleNew {
+            user_id: self.user_id,
+            role_id: self.role_id,
+            status: self.status,
+            created_by: self.created_by,
+            last_modified_by: self.last_modified_by,
+            created_at: self.created_at,
+            last_modified: self.last_modified,
+            org_id: Some(org_id),
+            merchant_id: Some(merchant_id),
+            profile_id: None,
+            entity_id: None,
+            entity_type: None,
+            version: UserRoleVersion::V1,
+        }
+    }
+
+    fn convert_to_new_v2_role(self, entity: EntityInfo) -> UserRoleNew {
+        UserRoleNew {
+            user_id: self.user_id,
+            role_id: self.role_id,
+            status: self.status,
+            created_by: self.created_by,
+            last_modified_by: self.last_modified_by,
+            created_at: self.created_at,
+            last_modified: self.last_modified,
+            org_id: Some(entity.org_id),
+            merchant_id: entity.merchant_id,
+            profile_id: entity.profile_id,
+            entity_id: Some(entity.entity_id),
+            entity_type: Some(entity.entity_type),
+            version: UserRoleVersion::V2,
+        }
+    }
+
+    async fn insert_v1_and_v2_in_db_and_get_v1(
+        state: &SessionState,
+        v1_role: UserRoleNew,
+        v2_role: UserRoleNew,
+    ) -> UserResult<UserRole> {
+        let inserted_roles = state
+            .store
+            .insert_user_role(InsertUserRolePayload::V1AndV2(Box::new([v1_role, v2_role])))
+            .await
+            .change_context(UserErrors::InternalServerError)?;
+
+        // Returning v1 role so other code which was not migrated doesn't break
+        inserted_roles
+            .into_iter()
+            .find(|role| role.version == UserRoleVersion::V1)
+            .ok_or(report!(UserErrors::InternalServerError))
+    }
+}
+
+impl NewUserRole<OrganizationLevel> {
+    pub async fn insert_in_v1(self, state: &SessionState) -> UserResult<UserRole> {
+        let entity = self.entity.clone();
+
+        let new_v1_role = self
+            .clone()
+            .convert_to_new_v1_role(entity.org_id.clone(), entity.merchant_id.clone());
+
+        state
+            .store
+            .insert_user_role(InsertUserRolePayload::OnlyV1(new_v1_role))
+            .await
+            .change_context(UserErrors::InternalServerError)?
+            .pop()
+            .ok_or(report!(UserErrors::InternalServerError))
+    }
+
+    pub async fn insert_in_v2(self, state: &SessionState) -> UserResult<UserRole> {
+        let entity = self.entity.clone();
+
+        let new_v2_role = self.convert_to_new_v2_role(EntityInfo {
+            org_id: entity.org_id.clone(),
+            merchant_id: None,
+            profile_id: None,
+            entity_id: entity.org_id.get_string_repr().to_owned(),
+            entity_type: EntityType::Organization,
+        });
+        state
+            .store
+            .insert_user_role(InsertUserRolePayload::OnlyV2(new_v2_role))
+            .await
+            .change_context(UserErrors::InternalServerError)?
+            .pop()
+            .ok_or(report!(UserErrors::InternalServerError))
+    }
+
+    pub async fn insert_in_v1_and_v2(self, state: &SessionState) -> UserResult<UserRole> {
+        let entity = self.entity.clone();
+
+        let new_v1_role = self
+            .clone()
+            .convert_to_new_v1_role(entity.org_id.clone(), entity.merchant_id.clone());
+
+        let new_v2_role = self.clone().convert_to_new_v2_role(EntityInfo {
+            org_id: entity.org_id.clone(),
+            merchant_id: None,
+            profile_id: None,
+            entity_id: entity.org_id.get_string_repr().to_owned(),
+            entity_type: EntityType::Organization,
+        });
+
+        Self::insert_v1_and_v2_in_db_and_get_v1(state, new_v1_role, new_v2_role).await
+    }
+}
+
+impl NewUserRole<MerchantLevel> {
+    pub async fn insert_in_v1_and_v2(self, state: &SessionState) -> UserResult<UserRole> {
+        let entity = self.entity.clone();
+
+        let new_v1_role = self
+            .clone()
+            .convert_to_new_v1_role(entity.org_id.clone(), entity.merchant_id.clone());
+
+        let new_v2_role = self.clone().convert_to_new_v2_role(EntityInfo {
+            org_id: entity.org_id.clone(),
+            merchant_id: Some(entity.merchant_id.clone()),
+            profile_id: None,
+            entity_id: entity.merchant_id.get_string_repr().to_owned(),
+            entity_type: EntityType::Merchant,
+        });
+
+        Self::insert_v1_and_v2_in_db_and_get_v1(state, new_v1_role, new_v2_role).await
+    }
+}
+
+impl NewUserRole<InternalLevel> {
+    pub async fn insert_in_v1_and_v2(self, state: &SessionState) -> UserResult<UserRole> {
+        let entity = self.entity.clone();
+        let internal_merchant_id = id_type::MerchantId::get_internal_user_merchant_id(
+            consts::user_role::INTERNAL_USER_MERCHANT_ID,
+        );
+
+        let new_v1_role = self
+            .clone()
+            .convert_to_new_v1_role(entity.org_id.clone(), internal_merchant_id.clone());
+
+        let new_v2_role = self.convert_to_new_v2_role(EntityInfo {
+            org_id: entity.org_id.clone(),
+            merchant_id: Some(internal_merchant_id.clone()),
+            profile_id: None,
+            entity_id: internal_merchant_id.get_string_repr().to_owned(),
+            entity_type: EntityType::Internal,
+        });
+
+        Self::insert_v1_and_v2_in_db_and_get_v1(state, new_v1_role, new_v2_role).await
+    }
+}
+
+impl NewUserRole<ProfileLevel> {
+    pub async fn insert_in_v2(self, state: &SessionState) -> UserResult<UserRole> {
+        let entity = self.entity.clone();
+
+        let new_v2_role = self.convert_to_new_v2_role(EntityInfo {
+            org_id: entity.org_id.clone(),
+            merchant_id: Some(entity.merchant_id.clone()),
+            profile_id: Some(entity.profile_id.clone()),
+            entity_id: entity.profile_id.get_string_repr().to_owned(),
+            entity_type: EntityType::Profile,
+        });
+        state
+            .store
+            .insert_user_role(InsertUserRolePayload::OnlyV2(new_v2_role))
+            .await
+            .change_context(UserErrors::InternalServerError)?
+            .pop()
+            .ok_or(report!(UserErrors::InternalServerError))
+    }
+}
diff --git a/migrations/2024-08-06-103905_drop_user_id_merchant_id_unique_in_user_roles/down.sql b/migrations/2024-08-06-103905_drop_user_id_merchant_id_unique_in_user_roles/down.sql
new file mode 100644
index 00000000000..3a210e27da9
--- /dev/null
+++ b/migrations/2024-08-06-103905_drop_user_id_merchant_id_unique_in_user_roles/down.sql
@@ -0,0 +1,2 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE user_roles ADD CONSTRAINT user_merchant_unique UNIQUE (user_id, merchant_id);
diff --git a/migrations/2024-08-06-103905_drop_user_id_merchant_id_unique_in_user_roles/up.sql b/migrations/2024-08-06-103905_drop_user_id_merchant_id_unique_in_user_roles/up.sql
new file mode 100644
index 00000000000..33c04e8dbde
--- /dev/null
+++ b/migrations/2024-08-06-103905_drop_user_id_merchant_id_unique_in_user_roles/up.sql
@@ -0,0 +1,2 @@
+-- Your SQL goes here
+ALTER TABLE user_roles DROP CONSTRAINT user_merchant_unique;