Issue:
Bug: refactor: migrate user roles table to support v2



- Make org_id and merchant_id nullable.
- Add new nullable columns profile_id, entity_id , entity_type, version

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [ ] New feature
- [ ] Enhancement
- [x] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
### DB changes in user_roles
- New Columns
  - `version`
  - `entity_id`
  - `entity_type`
  - `profile_id`
- Changes
  - `merchant_id` is now nullable
  - `org_id` is now nullable

### 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
Closes [#5352](https://github.com/juspay/hyperswitch/issues/5352)


## How did you test it?
All these APIs should work without giving 500.
- User - Signup w/ merchant id
- User - Signin
- User - Signin Token Only
- User - Connect Account
- User - Reset Password
- User - Invite Multiple
- User - Resend Invite
- User - Accept Invite from Mail
- User - Accept Invite from Mail Token only
- User - Switch Merchant
- User - Switch List
- User - Merchants List
- User - Merchants Select List
- User - List
- User - Verify Email
- User - Verify Email Token only
- User - Update User details

- User Role - Update
- User Role - Transfer Ownership
- User Role - Accept Invite - POST
- User Role - Accept Invite - PUT
- User Role - Delete






## 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 these APIs should work without giving 500.
- User - Signup w/ merchant id
- User - Signin
- User - Signin Token Only
- User - Connect Account
- User - Reset Password
- User - Invite Multiple
- User - Resend Invite
- User - Accept Invite from Mail
- User - Accept Invite from Mail Token only
- User - Switch Merchant
- User - Switch List
- User - Merchants List
- User - Merchants Select List
- User - List
- User - Verify Email
- User - Verify Email Token only
- User - Update User details

- User Role - Update
- User Role - Transfer Ownership
- User Role - Accept Invite - POST
- User Role - Accept Invite - PUT
- User Role - Delete

Patch:
diff --git a/crates/diesel_models/src/enums.rs b/crates/diesel_models/src/enums.rs
index 96678a7a3e1..f0059994cb6 100644
--- a/crates/diesel_models/src/enums.rs
+++ b/crates/diesel_models/src/enums.rs
@@ -19,7 +19,8 @@ pub mod diesel_exports {
         DbRequestIncrementalAuthorization as RequestIncrementalAuthorization,
         DbRoleScope as RoleScope, DbRoutingAlgorithmKind as RoutingAlgorithmKind,
         DbTotpStatus as TotpStatus, DbTransactionType as TransactionType,
-        DbUserStatus as UserStatus, DbWebhookDeliveryAttempt as WebhookDeliveryAttempt,
+        DbUserRoleVersion as UserRoleVersion, DbUserStatus as UserStatus,
+        DbWebhookDeliveryAttempt as WebhookDeliveryAttempt,
     };
 }
 pub use common_enums::*;
@@ -298,3 +299,24 @@ pub enum TotpStatus {
     #[default]
     NotSet,
 }
+
+#[derive(
+    Clone,
+    Copy,
+    Debug,
+    Default,
+    Eq,
+    PartialEq,
+    serde::Serialize,
+    serde::Deserialize,
+    strum::EnumString,
+    strum::Display,
+)]
+#[diesel_enum(storage_type = "db_enum")]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum UserRoleVersion {
+    #[default]
+    V1,
+    V2,
+}
diff --git a/crates/diesel_models/src/query/user_role.rs b/crates/diesel_models/src/query/user_role.rs
index b3fa3994793..2f2923db000 100644
--- a/crates/diesel_models/src/query/user_role.rs
+++ b/crates/diesel_models/src/query/user_role.rs
@@ -1,7 +1,10 @@
 use common_utils::id_type;
 use diesel::{associations::HasTable, BoolExpressionMethods, ExpressionMethods};
 
-use crate::{query::generics, schema::user_roles::dsl, user_role::*, PgPooledConn, StorageResult};
+use crate::{
+    enums::UserRoleVersion, query::generics, schema::user_roles::dsl, user_role::*, PgPooledConn,
+    StorageResult,
+};
 
 impl UserRoleNew {
     pub async fn insert(self, conn: &PgPooledConn) -> StorageResult<UserRole> {
@@ -10,10 +13,14 @@ impl UserRoleNew {
 }
 
 impl UserRole {
-    pub async fn find_by_user_id(conn: &PgPooledConn, user_id: String) -> StorageResult<Self> {
+    pub async fn find_by_user_id(
+        conn: &PgPooledConn,
+        user_id: String,
+        version: UserRoleVersion,
+    ) -> StorageResult<Self> {
         generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
             conn,
-            dsl::user_id.eq(user_id),
+            dsl::user_id.eq(user_id).and(dsl::version.eq(version)),
         )
         .await
     }
@@ -22,12 +29,14 @@ impl UserRole {
         conn: &PgPooledConn,
         user_id: String,
         merchant_id: id_type::MerchantId,
+        version: UserRoleVersion,
     ) -> StorageResult<Self> {
         generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
             conn,
             dsl::user_id
                 .eq(user_id)
-                .and(dsl::merchant_id.eq(merchant_id)),
+                .and(dsl::merchant_id.eq(merchant_id))
+                .and(dsl::version.eq(version)),
         )
         .await
     }
@@ -37,6 +46,7 @@ impl UserRole {
         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,
@@ -47,7 +57,8 @@ impl UserRole {
             conn,
             dsl::user_id
                 .eq(user_id)
-                .and(dsl::merchant_id.eq(merchant_id)),
+                .and(dsl::merchant_id.eq(merchant_id))
+                .and(dsl::version.eq(version)),
             UserRoleUpdateInternal::from(update),
         )
         .await
@@ -58,10 +69,14 @@ impl UserRole {
         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)),
+            dsl::user_id
+                .eq(user_id)
+                .and(dsl::org_id.eq(org_id))
+                .and(dsl::version.eq(version)),
             UserRoleUpdateInternal::from(update),
         )
         .await
@@ -71,20 +86,26 @@ impl UserRole {
         conn: &PgPooledConn,
         user_id: String,
         merchant_id: id_type::MerchantId,
+        version: UserRoleVersion,
     ) -> StorageResult<Self> {
         generics::generic_delete_one_with_result::<<Self as HasTable>::Table, _, _>(
             conn,
             dsl::user_id
                 .eq(user_id)
-                .and(dsl::merchant_id.eq(merchant_id)),
+                .and(dsl::merchant_id.eq(merchant_id))
+                .and(dsl::version.eq(version)),
         )
         .await
     }
 
-    pub async fn list_by_user_id(conn: &PgPooledConn, user_id: String) -> StorageResult<Vec<Self>> {
+    pub async fn list_by_user_id(
+        conn: &PgPooledConn,
+        user_id: String,
+        version: UserRoleVersion,
+    ) -> StorageResult<Vec<Self>> {
         generics::generic_filter::<<Self as HasTable>::Table, _, _, _>(
             conn,
-            dsl::user_id.eq(user_id),
+            dsl::user_id.eq(user_id).and(dsl::version.eq(version)),
             None,
             None,
             Some(dsl::created_at.asc()),
@@ -95,10 +116,13 @@ impl UserRole {
     pub async fn list_by_merchant_id(
         conn: &PgPooledConn,
         merchant_id: id_type::MerchantId,
+        version: UserRoleVersion,
     ) -> StorageResult<Vec<Self>> {
         generics::generic_filter::<<Self as HasTable>::Table, _, _, _>(
             conn,
-            dsl::merchant_id.eq(merchant_id),
+            dsl::merchant_id
+                .eq(merchant_id)
+                .and(dsl::version.eq(version)),
             None,
             None,
             Some(dsl::created_at.asc()),
diff --git a/crates/diesel_models/src/schema.rs b/crates/diesel_models/src/schema.rs
index 544a99bccb5..185677c4c75 100644
--- a/crates/diesel_models/src/schema.rs
+++ b/crates/diesel_models/src/schema.rs
@@ -1243,16 +1243,16 @@ diesel::table! {
     use diesel::sql_types::*;
     use crate::enums::diesel_exports::*;
 
-    user_roles (user_id, merchant_id) {
+    user_roles (id) {
         id -> Int4,
         #[max_length = 64]
         user_id -> Varchar,
         #[max_length = 64]
-        merchant_id -> Varchar,
+        merchant_id -> Nullable<Varchar>,
         #[max_length = 64]
         role_id -> Varchar,
         #[max_length = 64]
-        org_id -> Varchar,
+        org_id -> Nullable<Varchar>,
         status -> UserStatus,
         #[max_length = 64]
         created_by -> Varchar,
@@ -1260,6 +1260,13 @@ diesel::table! {
         last_modified_by -> Varchar,
         created_at -> Timestamp,
         last_modified -> Timestamp,
+        #[max_length = 64]
+        profile_id -> Nullable<Varchar>,
+        #[max_length = 64]
+        entity_id -> Nullable<Varchar>,
+        #[max_length = 64]
+        entity_type -> Nullable<Varchar>,
+        version -> UserRoleVersion,
     }
 }
 
diff --git a/crates/diesel_models/src/schema_v2.rs b/crates/diesel_models/src/schema_v2.rs
index e60e2da0a47..28937b48db4 100644
--- a/crates/diesel_models/src/schema_v2.rs
+++ b/crates/diesel_models/src/schema_v2.rs
@@ -1252,16 +1252,16 @@ diesel::table! {
     use diesel::sql_types::*;
     use crate::enums::diesel_exports::*;
 
-    user_roles (user_id, merchant_id) {
+    user_roles (id) {
         id -> Int4,
         #[max_length = 64]
         user_id -> Varchar,
         #[max_length = 64]
-        merchant_id -> Varchar,
+        merchant_id -> Nullable<Varchar>,
         #[max_length = 64]
         role_id -> Varchar,
         #[max_length = 64]
-        org_id -> Varchar,
+        org_id -> Nullable<Varchar>,
         status -> UserStatus,
         #[max_length = 64]
         created_by -> Varchar,
@@ -1269,6 +1269,13 @@ diesel::table! {
         last_modified_by -> Varchar,
         created_at -> Timestamp,
         last_modified -> Timestamp,
+        #[max_length = 64]
+        profile_id -> Nullable<Varchar>,
+        #[max_length = 64]
+        entity_id -> Nullable<Varchar>,
+        #[max_length = 64]
+        entity_type -> Nullable<Varchar>,
+        version -> UserRoleVersion,
     }
 }
 
diff --git a/crates/diesel_models/src/user_role.rs b/crates/diesel_models/src/user_role.rs
index 7f7c8484e8f..d484db2c6ba 100644
--- a/crates/diesel_models/src/user_role.rs
+++ b/crates/diesel_models/src/user_role.rs
@@ -9,28 +9,36 @@ use crate::{enums, schema::user_roles};
 pub struct UserRole {
     pub id: i32,
     pub user_id: String,
-    pub merchant_id: id_type::MerchantId,
+    pub merchant_id: Option<id_type::MerchantId>,
     pub role_id: String,
-    pub org_id: id_type::OrganizationId,
+    pub org_id: Option<id_type::OrganizationId>,
     pub status: enums::UserStatus,
     pub created_by: String,
     pub last_modified_by: String,
     pub created_at: PrimitiveDateTime,
     pub last_modified: PrimitiveDateTime,
+    pub profile_id: Option<String>,
+    pub entity_id: Option<String>,
+    pub entity_type: Option<String>,
+    pub version: enums::UserRoleVersion,
 }
 
 #[derive(router_derive::Setter, Clone, Debug, Insertable, router_derive::DebugAsDisplay)]
 #[diesel(table_name = user_roles)]
 pub struct UserRoleNew {
     pub user_id: String,
-    pub merchant_id: id_type::MerchantId,
+    pub merchant_id: Option<id_type::MerchantId>,
     pub role_id: String,
-    pub org_id: id_type::OrganizationId,
+    pub org_id: Option<id_type::OrganizationId>,
     pub status: enums::UserStatus,
     pub created_by: String,
     pub last_modified_by: String,
     pub created_at: PrimitiveDateTime,
     pub last_modified: PrimitiveDateTime,
+    pub profile_id: Option<String>,
+    pub entity_id: Option<String>,
+    pub entity_type: Option<String>,
+    pub version: enums::UserRoleVersion,
 }
 
 #[derive(Clone, Debug, AsChangeset, router_derive::DebugAsDisplay)]
diff --git a/crates/router/src/core/user.rs b/crates/router/src/core/user.rs
index 9389240c972..ba85a382dbf 100644
--- a/crates/router/src/core/user.rs
+++ b/crates/router/src/core/user.rs
@@ -8,7 +8,7 @@ use common_utils::types::keymanager::Identifier;
 #[cfg(feature = "email")]
 use diesel_models::user_role::UserRoleUpdate;
 use diesel_models::{
-    enums::{TotpStatus, UserStatus},
+    enums::{TotpStatus, UserRoleVersion, UserStatus},
     user as storage_user,
     user_authentication_method::{UserAuthenticationMethodNew, UserAuthenticationMethodUpdate},
     user_role::UserRoleNew,
@@ -81,11 +81,16 @@ pub async fn signup_with_merchant_id(
         )
         .await;
 
+    let Some(merchant_id) = user_role.merchant_id else {
+        return Err(report!(UserErrors::InternalServerError)
+            .attach_printable("merchant_id not found for user_role"));
+    };
+
     logger::info!(?send_email_result);
     Ok(ApplicationResponse::Json(user_api::AuthorizeResponse {
         is_email_sent: send_email_result.is_ok(),
         user_id: user_from_db.get_user_id().to_string(),
-        merchant_id: user_role.merchant_id,
+        merchant_id,
     }))
 }
 
@@ -203,7 +208,7 @@ pub async fn signin(
                 .attach_printable("User role with preferred_merchant_id not found")?;
             domain::SignInWithRoleStrategyType::SingleRole(domain::SignInWithSingleRoleStrategy {
                 user: user_from_db,
-                user_role: preferred_role,
+                user_role: Box::new(preferred_role),
             })
         } else {
             let user_roles = user_from_db.get_roles_from_db(&state).await?;
@@ -272,13 +277,17 @@ pub async fn connect_account(
             )
             .await;
 
+        let Some(merchant_id) = user_role.merchant_id else {
+            return Err(report!(UserErrors::InternalServerError)
+                .attach_printable("merchant_id not found for user_role"));
+        };
         logger::info!(?send_email_result);
 
         return Ok(ApplicationResponse::Json(
             user_api::ConnectAccountResponse {
                 is_email_sent: send_email_result.is_ok(),
                 user_id: user_from_db.get_user_id().to_string(),
-                merchant_id: user_role.merchant_id,
+                merchant_id,
             },
         ));
     } else if find_user
@@ -323,13 +332,18 @@ pub async fn connect_account(
             )
             .await;
 
+        let Some(merchant_id) = user_role.merchant_id else {
+            return Err(report!(UserErrors::InternalServerError)
+                .attach_printable("merchant_id not found for user_role"));
+        };
+
         logger::info!(?send_email_result);
 
         return Ok(ApplicationResponse::Json(
             user_api::ConnectAccountResponse {
                 is_email_sent: send_email_result.is_ok(),
                 user_id: user_from_db.get_user_id().to_string(),
-                merchant_id: user_role.merchant_id,
+                merchant_id,
             },
         ));
     } else {
@@ -587,6 +601,7 @@ pub async fn reset_password(
                     status: UserStatus::Active,
                     modified_by: user.user_id.clone(),
                 },
+                UserRoleVersion::V1,
             )
             .await;
         logger::info!(?update_status_result);
@@ -719,9 +734,9 @@ async fn handle_existing_user_invitation(
         .store
         .insert_user_role(UserRoleNew {
             user_id: invitee_user_from_db.get_user_id().to_owned(),
-            merchant_id: user_from_token.merchant_id.clone(),
+            merchant_id: Some(user_from_token.merchant_id.clone()),
             role_id: request.role_id.clone(),
-            org_id: user_from_token.org_id.clone(),
+            org_id: Some(user_from_token.org_id.clone()),
             status: {
                 if cfg!(feature = "email") {
                     UserStatus::InvitationSent
@@ -733,6 +748,10 @@ async fn handle_existing_user_invitation(
             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| {
@@ -807,14 +826,18 @@ async fn handle_new_user_invitation(
         .store
         .insert_user_role(UserRoleNew {
             user_id: new_user.get_user_id().to_owned(),
-            merchant_id: user_from_token.merchant_id.clone(),
+            merchant_id: Some(user_from_token.merchant_id.clone()),
             role_id: request.role_id.clone(),
-            org_id: user_from_token.org_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| {
@@ -916,7 +939,11 @@ pub async fn resend_invite(
         .into();
     let user_role = state
         .store
-        .find_user_role_by_user_id_merchant_id(user.get_user_id(), &user_from_token.merchant_id)
+        .find_user_role_by_user_id_merchant_id(
+            user.get_user_id(),
+            &user_from_token.merchant_id,
+            UserRoleVersion::V1,
+        )
         .await
         .map_err(|e| {
             if e.current_context().is_db_not_found() {
@@ -993,6 +1020,7 @@ pub async fn accept_invite_from_email(
                 status: UserStatus::Active,
                 modified_by: user.get_user_id().to_string(),
             },
+            UserRoleVersion::V1,
         )
         .await
         .change_context(UserErrors::InternalServerError)?;
@@ -1065,6 +1093,7 @@ pub async fn accept_invite_from_email_token_only_flow(
                 status: UserStatus::Active,
                 modified_by: user_from_db.get_user_id().to_string(),
             },
+            UserRoleVersion::V1,
         )
         .await
         .change_context(UserErrors::InternalServerError)?;
@@ -1241,7 +1270,7 @@ pub async fn switch_merchant_id(
     } else {
         let user_roles = state
             .store
-            .list_user_roles_by_user_id(&user_from_token.user_id)
+            .list_user_roles_by_user_id(&user_from_token.user_id, UserRoleVersion::V1)
             .await
             .change_context(UserErrors::InternalServerError)?;
 
@@ -1252,7 +1281,17 @@ pub async fn switch_merchant_id(
 
         let user_role = active_user_roles
             .iter()
-            .find(|role| role.merchant_id == request.merchant_id)
+            .find_map(|role| {
+                let Some(ref merchant_id) = role.merchant_id else {
+                    return Some(Err(report!(UserErrors::InternalServerError)));
+                };
+                if merchant_id == &request.merchant_id {
+                    Some(Ok(role))
+                } else {
+                    None
+                }
+            })
+            .transpose()?
             .ok_or(report!(UserErrors::InvalidRoleOperation))
             .attach_printable("User doesn't have access to switch")?;
 
@@ -1312,7 +1351,7 @@ pub async fn list_merchants_for_user(
 ) -> UserResponse<Vec<user_api::UserMerchantAccount>> {
     let user_roles = state
         .store
-        .list_user_roles_by_user_id(user_from_token.user_id.as_str())
+        .list_user_roles_by_user_id(user_from_token.user_id.as_str(), UserRoleVersion::V1)
         .await
         .change_context(UserErrors::InternalServerError)?;
 
@@ -1322,8 +1361,12 @@ pub async fn list_merchants_for_user(
             &(&state).into(),
             user_roles
                 .iter()
-                .map(|role| role.merchant_id.clone())
-                .collect(),
+                .map(|role| {
+                    role.merchant_id
+                        .clone()
+                        .ok_or(UserErrors::InternalServerError)
+                })
+                .collect::<Result<Vec<_>, _>>()?,
         )
         .await
         .change_context(UserErrors::InternalServerError)?;
@@ -1355,6 +1398,7 @@ pub async fn get_user_details_in_merchant_account(
         .find_user_role_by_user_id_merchant_id(
             required_user.get_user_id(),
             &user_from_token.merchant_id,
+            UserRoleVersion::V1,
         )
         .await
         .to_not_found_response(UserErrors::InvalidRoleOperation)
@@ -1390,7 +1434,7 @@ pub async fn list_users_for_merchant_account(
 ) -> UserResponse<user_api::ListUsersResponse> {
     let user_roles: HashMap<String, _> = state
         .store
-        .list_user_roles_by_merchant_id(&user_from_token.merchant_id)
+        .list_user_roles_by_merchant_id(&user_from_token.merchant_id, UserRoleVersion::V1)
         .await
         .change_context(UserErrors::InternalServerError)
         .attach_printable("No user roles for given merchant id")?
@@ -1422,8 +1466,14 @@ pub async fn list_users_for_merchant_account(
                 roles::RoleInfo::from_role_id(
                     &state,
                     &user_role.role_id.clone(),
-                    &user_role.merchant_id,
-                    &user_role.org_id,
+                    user_role
+                        .merchant_id
+                        .as_ref()
+                        .ok_or(UserErrors::InternalServerError)?,
+                    user_role
+                        .org_id
+                        .as_ref()
+                        .ok_or(UserErrors::InternalServerError)?,
                 )
                 .await
                 .map(|role_info| (user, user_role, role_info))
@@ -1491,7 +1541,7 @@ pub async fn verify_email(
                 .attach_printable("User role with preferred_merchant_id not found")?;
             domain::SignInWithRoleStrategyType::SingleRole(domain::SignInWithSingleRoleStrategy {
                 user: user_from_db,
-                user_role: preferred_role,
+                user_role: Box::new(preferred_role),
             })
         } else {
             let user_roles = user_from_db.get_roles_from_db(&state).await?;
@@ -1624,10 +1674,11 @@ pub async fn verify_token(
         })?;
     let merchant_id = state
         .store
-        .find_user_role_by_user_id(&req.user_id)
+        .find_user_role_by_user_id(&req.user_id, UserRoleVersion::V1)
         .await
         .change_context(UserErrors::InternalServerError)?
-        .merchant_id;
+        .merchant_id
+        .ok_or(UserErrors::InternalServerError)?;
 
     Ok(ApplicationResponse::Json(user_api::VerifyTokenResponse {
         merchant_id: merchant_id.to_owned(),
@@ -1653,7 +1704,11 @@ pub async fn update_user_details(
     if let Some(ref preferred_merchant_id) = req.preferred_merchant_id {
         let _ = state
             .store
-            .find_user_role_by_user_id_merchant_id(user.get_user_id(), preferred_merchant_id)
+            .find_user_role_by_user_id_merchant_id(
+                user.get_user_id(),
+                preferred_merchant_id,
+                UserRoleVersion::V1,
+            )
             .await
             .map_err(|e| {
                 if e.current_context().is_db_not_found() {
diff --git a/crates/router/src/core/user_role.rs b/crates/router/src/core/user_role.rs
index 5d93c00909d..a728836857c 100644
--- a/crates/router/src/core/user_role.rs
+++ b/crates/router/src/core/user_role.rs
@@ -1,5 +1,8 @@
 use api_models::{user as user_api, user_role as user_role_api};
-use diesel_models::{enums::UserStatus, user_role::UserRoleUpdate};
+use diesel_models::{
+    enums::{UserRoleVersion, UserStatus},
+    user_role::UserRoleUpdate,
+};
 use error_stack::{report, ResultExt};
 use router_env::logger;
 
@@ -101,11 +104,15 @@ pub async fn update_user_role(
         .store
         .update_user_role_by_user_id_merchant_id(
             user_to_be_updated.get_user_id(),
-            &user_role_to_be_updated.merchant_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)
@@ -146,6 +153,7 @@ pub async fn transfer_org_ownership(
             &user_from_token.user_id,
             user_to_be_updated.get_user_id(),
             &user_from_token.org_id,
+            UserRoleVersion::V1,
         )
         .await
         .change_context(UserErrors::InternalServerError)?;
@@ -183,6 +191,7 @@ pub async fn accept_invitation(
                     status: UserStatus::Active,
                     modified_by: user_token.user_id.clone(),
                 },
+                UserRoleVersion::V1,
             )
             .await
             .map_err(|e| {
@@ -213,6 +222,7 @@ pub async fn merchant_select(
                     status: UserStatus::Active,
                     modified_by: user_token.user_id.clone(),
                 },
+                UserRoleVersion::V1,
             )
             .await
             .map_err(|e| {
@@ -267,6 +277,7 @@ pub async fn merchant_select_token_only_flow(
                     status: UserStatus::Active,
                     modified_by: user_token.user_id.clone(),
                 },
+                UserRoleVersion::V1,
             )
             .await
             .map_err(|e| {
@@ -329,15 +340,17 @@ pub async fn delete_user_role(
 
     let user_roles = state
         .store
-        .list_user_roles_by_user_id(user_from_db.get_user_id())
+        .list_user_roles_by_user_id(user_from_db.get_user_id(), UserRoleVersion::V1)
         .await
         .change_context(UserErrors::InternalServerError)?;
 
-    match user_roles
-        .iter()
-        .find(|&role| role.merchant_id == user_from_token.merchant_id)
-    {
-        Some(user_role) => {
+    for user_role in user_roles.iter() {
+        let Some(merchant_id) = user_role.merchant_id.as_ref() else {
+            return Err(report!(UserErrors::InternalServerError))
+                .attach_printable("merchant_id not found for user_role");
+        };
+
+        if merchant_id == &user_from_token.merchant_id {
             let role_info = roles::RoleInfo::from_role_id(
                 &state,
                 &user_role.role_id,
@@ -350,12 +363,11 @@ pub async fn delete_user_role(
                 return Err(report!(UserErrors::InvalidDeleteOperation))
                     .attach_printable(format!("role_id = {} is not deletable", user_role.role_id));
             }
-        }
-        None => {
+        } else {
             return Err(report!(UserErrors::InvalidDeleteOperation))
                 .attach_printable("User is not associated with the merchant");
         }
-    };
+    }
 
     let deleted_user_role = if user_roles.len() > 1 {
         state
@@ -363,6 +375,7 @@ pub async fn delete_user_role(
             .delete_user_role_by_user_id_merchant_id(
                 user_from_db.get_user_id(),
                 &user_from_token.merchant_id,
+                UserRoleVersion::V1,
             )
             .await
             .change_context(UserErrors::InternalServerError)
@@ -380,6 +393,7 @@ pub async fn delete_user_role(
             .delete_user_role_by_user_id_merchant_id(
                 user_from_db.get_user_id(),
                 &user_from_token.merchant_id,
+                UserRoleVersion::V1,
             )
             .await
             .change_context(UserErrors::InternalServerError)
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 003f3ff4845..55fb7096a3a 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -2535,17 +2535,21 @@ impl UserRoleInterface for KafkaStore {
     async fn find_user_role_by_user_id(
         &self,
         user_id: &str,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<user_storage::UserRole, errors::StorageError> {
-        self.diesel_store.find_user_role_by_user_id(user_id).await
+        self.diesel_store
+            .find_user_role_by_user_id(user_id, version)
+            .await
     }
 
     async fn find_user_role_by_user_id_merchant_id(
         &self,
         user_id: &str,
         merchant_id: &id_type::MerchantId,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<user_storage::UserRole, errors::StorageError> {
         self.diesel_store
-            .find_user_role_by_user_id_merchant_id(user_id, merchant_id)
+            .find_user_role_by_user_id_merchant_id(user_id, merchant_id, version)
             .await
     }
 
@@ -2554,9 +2558,10 @@ impl UserRoleInterface for KafkaStore {
         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)
+            .update_user_role_by_user_id_merchant_id(user_id, merchant_id, update, version)
             .await
     }
 
@@ -2565,9 +2570,10 @@ impl UserRoleInterface for KafkaStore {
         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)
+            .update_user_roles_by_user_id_org_id(user_id, org_id, update, version)
             .await
     }
 
@@ -2575,17 +2581,21 @@ impl UserRoleInterface for KafkaStore {
         &self,
         user_id: &str,
         merchant_id: &id_type::MerchantId,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<user_storage::UserRole, errors::StorageError> {
         self.diesel_store
-            .delete_user_role_by_user_id_merchant_id(user_id, merchant_id)
+            .delete_user_role_by_user_id_merchant_id(user_id, merchant_id, version)
             .await
     }
 
     async fn list_user_roles_by_user_id(
         &self,
         user_id: &str,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<Vec<user_storage::UserRole>, errors::StorageError> {
-        self.diesel_store.list_user_roles_by_user_id(user_id).await
+        self.diesel_store
+            .list_user_roles_by_user_id(user_id, version)
+            .await
     }
 
     async fn transfer_org_ownership_between_users(
@@ -2593,18 +2603,20 @@ impl UserRoleInterface for KafkaStore {
         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)
+            .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,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<Vec<user_storage::UserRole>, errors::StorageError> {
         self.diesel_store
-            .list_user_roles_by_merchant_id(merchant_id)
+            .list_user_roles_by_merchant_id(merchant_id, version)
             .await
     }
 }
diff --git a/crates/router/src/db/user_role.rs b/crates/router/src/db/user_role.rs
index 7f16cedd570..cc3e2460967 100644
--- a/crates/router/src/db/user_role.rs
+++ b/crates/router/src/db/user_role.rs
@@ -1,4 +1,4 @@
-use std::{collections::HashSet, ops::Not};
+use std::collections::HashSet;
 
 use async_bb8_diesel::AsyncConnection;
 use common_utils::id_type;
@@ -23,12 +23,14 @@ pub trait UserRoleInterface {
     async fn find_user_role_by_user_id(
         &self,
         user_id: &str,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<storage::UserRole, errors::StorageError>;
 
     async fn find_user_role_by_user_id_merchant_id(
         &self,
         user_id: &str,
         merchant_id: &id_type::MerchantId,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<storage::UserRole, errors::StorageError>;
 
     async fn update_user_role_by_user_id_merchant_id(
@@ -36,6 +38,7 @@ pub trait UserRoleInterface {
         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(
@@ -43,22 +46,26 @@ pub trait UserRoleInterface {
         user_id: &str,
         org_id: &id_type::OrganizationId,
         update: storage::UserRoleUpdate,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError>;
 
     async fn delete_user_role_by_user_id_merchant_id(
         &self,
         user_id: &str,
         merchant_id: &id_type::MerchantId,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<storage::UserRole, errors::StorageError>;
 
     async fn list_user_roles_by_user_id(
         &self,
         user_id: &str,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError>;
 
     async fn list_user_roles_by_merchant_id(
         &self,
         merchant_id: &id_type::MerchantId,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError>;
 
     async fn transfer_org_ownership_between_users(
@@ -66,6 +73,7 @@ pub trait UserRoleInterface {
         from_user_id: &str,
         to_user_id: &str,
         org_id: &id_type::OrganizationId,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<(), errors::StorageError>;
 }
 
@@ -87,9 +95,10 @@ impl UserRoleInterface for Store {
     async fn find_user_role_by_user_id(
         &self,
         user_id: &str,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<storage::UserRole, errors::StorageError> {
         let conn = connection::pg_connection_write(self).await?;
-        storage::UserRole::find_by_user_id(&conn, user_id.to_owned())
+        storage::UserRole::find_by_user_id(&conn, user_id.to_owned(), version)
             .await
             .map_err(|error| report!(errors::StorageError::from(error)))
     }
@@ -99,12 +108,14 @@ impl UserRoleInterface for Store {
         &self,
         user_id: &str,
         merchant_id: &id_type::MerchantId,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<storage::UserRole, errors::StorageError> {
         let conn = connection::pg_connection_write(self).await?;
         storage::UserRole::find_by_user_id_merchant_id(
             &conn,
             user_id.to_owned(),
             merchant_id.to_owned(),
+            version,
         )
         .await
         .map_err(|error| report!(errors::StorageError::from(error)))
@@ -116,6 +127,7 @@ impl UserRoleInterface for Store {
         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(
@@ -123,6 +135,7 @@ impl UserRoleInterface for Store {
             user_id.to_owned(),
             merchant_id.to_owned(),
             update,
+            version,
         )
         .await
         .map_err(|error| report!(errors::StorageError::from(error)))
@@ -134,6 +147,7 @@ impl UserRoleInterface for Store {
         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(
@@ -141,6 +155,7 @@ impl UserRoleInterface for Store {
             user_id.to_owned(),
             org_id.to_owned(),
             update,
+            version,
         )
         .await
         .map_err(|error| report!(errors::StorageError::from(error)))
@@ -151,6 +166,7 @@ impl UserRoleInterface for Store {
         &self,
         user_id: &str,
         merchant_id: &id_type::MerchantId,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<storage::UserRole, errors::StorageError> {
         let conn = connection::pg_connection_write(self).await?;
 
@@ -158,6 +174,7 @@ impl UserRoleInterface for Store {
             &conn,
             user_id.to_owned(),
             merchant_id.to_owned(),
+            version,
         )
         .await
         .map_err(|error| report!(errors::StorageError::from(error)))
@@ -167,9 +184,10 @@ impl UserRoleInterface for Store {
     async fn list_user_roles_by_user_id(
         &self,
         user_id: &str,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError> {
         let conn = connection::pg_connection_write(self).await?;
-        storage::UserRole::list_by_user_id(&conn, user_id.to_owned())
+        storage::UserRole::list_by_user_id(&conn, user_id.to_owned(), version)
             .await
             .map_err(|error| report!(errors::StorageError::from(error)))
     }
@@ -178,9 +196,10 @@ impl UserRoleInterface for Store {
     async fn list_user_roles_by_merchant_id(
         &self,
         merchant_id: &id_type::MerchantId,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError> {
         let conn = connection::pg_connection_write(self).await?;
-        storage::UserRole::list_by_merchant_id(&conn, merchant_id.to_owned())
+        storage::UserRole::list_by_merchant_id(&conn, merchant_id.to_owned(), version)
             .await
             .map_err(|error| report!(errors::StorageError::from(error)))
     }
@@ -191,6 +210,7 @@ impl UserRoleInterface for Store {
         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
@@ -205,6 +225,7 @@ impl UserRoleInterface for Store {
                     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())?;
@@ -217,43 +238,56 @@ impl UserRoleInterface for Store {
                     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())
-                .collect::<HashSet<_>>();
+                .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 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_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,
-                            }
-                        })
-                });
-
-            futures::future::try_join_all(missing_new_user_roles.map(|user_role| async {
-                user_role
-                    .insert(&conn)
-                    .await
-                    .map_err(|e| *e.current_context())
-            }))
+            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>(())
@@ -293,6 +327,10 @@ impl UserRoleInterface for MockDb {
             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)
@@ -301,11 +339,12 @@ impl UserRoleInterface for MockDb {
     async fn find_user_role_by_user_id(
         &self,
         user_id: &str,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<storage::UserRole, errors::StorageError> {
         let user_roles = self.user_roles.lock().await;
         user_roles
             .iter()
-            .find(|user_role| user_role.user_id == user_id)
+            .find(|user_role| user_role.user_id == user_id && user_role.version == version)
             .cloned()
             .ok_or(
                 errors::StorageError::ValueNotFound(format!(
@@ -319,18 +358,31 @@ impl UserRoleInterface for MockDb {
         &self,
         user_id: &str,
         merchant_id: &id_type::MerchantId,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<storage::UserRole, errors::StorageError> {
         let user_roles = self.user_roles.lock().await;
-        user_roles
-            .iter()
-            .find(|user_role| user_role.user_id == user_id && user_role.merchant_id == *merchant_id)
-            .cloned()
-            .ok_or(
-                errors::StorageError::ValueNotFound(format!(
-                    "No user role available for user_id = {user_id} and merchant_id = {merchant_id:?}"
-                ))
-                .into(),
-            )
+
+        for user_role in user_roles.iter() {
+            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
+            {
+                return Ok(user_role.clone());
+            }
+        }
+
+        Err(errors::StorageError::ValueNotFound(format!(
+            "No user role available for user_id = {} and merchant_id = {}",
+            user_id, merchant_id
+        ))
+        .into())
     }
 
     async fn update_user_role_by_user_id_merchant_id(
@@ -338,38 +390,46 @@ impl UserRoleInterface for MockDb {
         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;
-        user_roles
-            .iter_mut()
-            .find(|user_role| user_role.user_id == user_id && user_role.merchant_id == *merchant_id)
-            .map(|user_role| {
-                *user_role = match &update {
+
+        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,
-                    } => storage::UserRole {
-                        role_id: role_id.to_string(),
-                        last_modified_by: modified_by.to_string(),
-                        ..user_role.to_owned()
-                    },
+                    } => {
+                        user_role.role_id = role_id.to_string();
+                        user_role.last_modified_by = modified_by.to_string();
+                    }
                     storage::UserRoleUpdate::UpdateStatus {
                         status,
                         modified_by,
-                    } => storage::UserRole {
-                        status: status.to_owned(),
-                        last_modified_by: modified_by.to_owned(),
-                        ..user_role.to_owned()
-                    },
+                    } => {
+                        user_role.status = *status;
+                        user_role.last_modified_by = modified_by.to_string();
+                    }
                 };
-                user_role.to_owned()
-            })
-            .ok_or(
-                errors::StorageError::ValueNotFound(format!(
-                    "No user role available for user_id = {user_id} and merchant_id = {merchant_id:?}"
-                ))
-                .into(),
-            )
+                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(
@@ -377,11 +437,22 @@ impl UserRoleInterface for MockDb {
         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() {
-            if user_role.user_id == user_id && user_role.org_id == *org_id {
+            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,
@@ -416,6 +487,7 @@ impl UserRoleInterface for MockDb {
         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(
@@ -425,6 +497,7 @@ impl UserRoleInterface for MockDb {
                     role_id: consts::user_role::ROLE_ID_MERCHANT_ADMIN.to_string(),
                     modified_by: from_user_id.to_string(),
                 },
+                version,
             )
             .await?;
 
@@ -436,35 +509,53 @@ impl UserRoleInterface for MockDb {
                     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())
-            .collect::<HashSet<_>>();
+            .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,
+                };
 
-        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_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,
-                    })
-                } else {
-                    None
-                }
-            });
+                missing_new_user_roles.push(new_user_role);
+            }
+        }
 
         for user_role in missing_new_user_roles {
             self.insert_user_role(user_role).await?;
@@ -477,14 +568,21 @@ impl UserRoleInterface for MockDb {
         &self,
         user_id: &str,
         merchant_id: &id_type::MerchantId,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<storage::UserRole, errors::StorageError> {
         let mut user_roles = self.user_roles.lock().await;
 
-        match user_roles
-            .iter()
-            .position(|role| role.user_id == user_id && role.merchant_id == *merchant_id)
-        {
-            Some(index) => Ok(user_roles.remove(index)),
+        let index = user_roles.iter().position(|role| {
+            role.user_id == user_id
+                && role.version == version
+                && match role.merchant_id {
+                    Some(ref mid) => mid == merchant_id,
+                    None => false,
+                }
+        });
+
+        match index {
+            Some(idx) => Ok(user_roles.remove(idx)),
             None => Err(errors::StorageError::ValueNotFound(
                 "Cannot find user role to delete".to_string(),
             )
@@ -495,6 +593,7 @@ impl UserRoleInterface for MockDb {
     async fn list_user_roles_by_user_id(
         &self,
         user_id: &str,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError> {
         let user_roles = self.user_roles.lock().await;
 
@@ -502,7 +601,7 @@ impl UserRoleInterface for MockDb {
             .iter()
             .cloned()
             .filter_map(|ele| {
-                if ele.user_id == user_id {
+                if ele.user_id == user_id && ele.version == version {
                     return Some(ele);
                 }
                 None
@@ -513,19 +612,26 @@ impl UserRoleInterface for MockDb {
     async fn list_user_roles_by_merchant_id(
         &self,
         merchant_id: &id_type::MerchantId,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError> {
         let user_roles = self.user_roles.lock().await;
 
-        Ok(user_roles
+        let filtered_roles: Vec<_> = user_roles
             .iter()
-            .cloned()
-            .filter_map(|ele| {
-                if ele.merchant_id == *merchant_id {
-                    return Some(ele);
+            .filter_map(|role| {
+                if let Some(role_merchant_id) = &role.merchant_id {
+                    if role_merchant_id == merchant_id && role.version == version {
+                        Some(role.clone())
+                    } else {
+                        None
+                    }
+                } else {
+                    None
                 }
-                None
             })
-            .collect())
+            .collect();
+
+        Ok(filtered_roles)
     }
 }
 
@@ -551,8 +657,11 @@ impl UserRoleInterface for super::KafkaStore {
     async fn find_user_role_by_user_id(
         &self,
         user_id: &str,
+        version: enums::UserRoleVersion,
     ) -> CustomResult<storage::UserRole, errors::StorageError> {
-        self.diesel_store.find_user_role_by_user_id(user_id).await
+        self.diesel_store
+            .find_user_role_by_user_id(user_id, version)
+            .await
     }
     async fn delete_user_role_by_user_id_merchant_id(
         &self,
diff --git a/crates/router/src/routes/recon.rs b/crates/router/src/routes/recon.rs
index be489a8ea06..910e2e8f96b 100644
--- a/crates/router/src/routes/recon.rs
+++ b/crates/router/src/routes/recon.rs
@@ -1,5 +1,6 @@
 use actix_web::{web, HttpRequest, HttpResponse};
 use api_models::recon as recon_api;
+use diesel_models::enums::UserRoleVersion;
 use error_stack::ResultExt;
 use masking::{ExposeInterface, PeekInterface, Secret};
 use router_env::Flow;
@@ -81,10 +82,11 @@ pub async fn send_recon_request(
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)?;
     let merchant_id = db
-        .find_user_role_by_user_id(&user.user_id)
+        .find_user_role_by_user_id(&user.user_id, UserRoleVersion::V1)
         .await
         .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?
-        .merchant_id;
+        .merchant_id
+        .ok_or(errors::ApiErrorResponse::InternalServerError)?;
     let key_manager_state = &(&state).into();
     let key_store = db
         .get_merchant_key_store_by_merchant_id(
diff --git a/crates/router/src/types/domain/user.rs b/crates/router/src/types/domain/user.rs
index 95349cb4393..8b973a8cf44 100644
--- a/crates/router/src/types/domain/user.rs
+++ b/crates/router/src/types/domain/user.rs
@@ -9,9 +9,8 @@ use common_utils::{
     types::keymanager::Identifier,
 };
 use diesel_models::{
-    enums::{TotpStatus, UserStatus},
-    organization as diesel_org,
-    organization::Organization,
+    enums::{TotpStatus, UserRoleVersion, UserStatus},
+    organization::{self as diesel_org, Organization},
     user as storage_user,
     user_role::{UserRole, UserRoleNew},
 };
@@ -655,7 +654,7 @@ impl NewUser {
         state
             .store
             .insert_user_role(UserRoleNew {
-                merchant_id: self.get_new_merchant().get_merchant_id(),
+                merchant_id: Some(self.get_new_merchant().get_merchant_id()),
                 status: user_status,
                 created_by: user_id.clone(),
                 last_modified_by: user_id.clone(),
@@ -663,10 +662,15 @@ impl NewUser {
                 role_id,
                 created_at: now,
                 last_modified: now,
-                org_id: self
-                    .get_new_merchant()
-                    .get_new_organization()
-                    .get_organization_id(),
+                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)
@@ -860,7 +864,7 @@ impl UserFromStorage {
     pub async fn get_role_from_db(&self, state: SessionState) -> UserResult<UserRole> {
         state
             .store
-            .find_user_role_by_user_id(&self.0.user_id)
+            .find_user_role_by_user_id(&self.0.user_id, UserRoleVersion::V1)
             .await
             .change_context(UserErrors::InternalServerError)
     }
@@ -868,7 +872,7 @@ impl UserFromStorage {
     pub async fn get_roles_from_db(&self, state: &SessionState) -> UserResult<Vec<UserRole>> {
         state
             .store
-            .list_user_roles_by_user_id(&self.0.user_id)
+            .list_user_roles_by_user_id(&self.0.user_id, UserRoleVersion::V1)
             .await
             .change_context(UserErrors::InternalServerError)
     }
@@ -931,7 +935,11 @@ impl UserFromStorage {
     ) -> CustomResult<UserRole, errors::StorageError> {
         state
             .store
-            .find_user_role_by_user_id_merchant_id(self.get_user_id(), merchant_id)
+            .find_user_role_by_user_id_merchant_id(
+                self.get_user_id(),
+                merchant_id,
+                UserRoleVersion::V1,
+            )
             .await
     }
 
@@ -945,7 +953,7 @@ impl UserFromStorage {
         } else {
             state
                 .store
-                .list_user_roles_by_user_id(&self.0.user_id)
+                .list_user_roles_by_user_id(&self.0.user_id, UserRoleVersion::V1)
                 .await?
                 .into_iter()
                 .find(|role| role.status == UserStatus::Active)
@@ -1109,7 +1117,7 @@ impl SignInWithRoleStrategyType {
         {
             Ok(Self::SingleRole(SignInWithSingleRoleStrategy {
                 user,
-                user_role: user_role.clone(),
+                user_role: Box::new(user_role.clone()),
             }))
         } else {
             Ok(Self::MultipleRoles(SignInWithMultipleRolesStrategy {
@@ -1132,7 +1140,7 @@ impl SignInWithRoleStrategyType {
 
 pub struct SignInWithSingleRoleStrategy {
     pub user: UserFromStorage,
-    pub user_role: UserRole,
+    pub user_role: Box<UserRole>,
 }
 
 impl SignInWithSingleRoleStrategy {
@@ -1145,7 +1153,7 @@ impl SignInWithSingleRoleStrategy {
         utils::user_role::set_role_permissions_in_cache_by_user_role(state, &self.user_role).await;
 
         let dashboard_entry_response =
-            utils::user::get_dashboard_entry_response(state, self.user, self.user_role, token)?;
+            utils::user::get_dashboard_entry_response(state, self.user, *self.user_role, token)?;
 
         Ok(user_api::SignInResponse::DashboardEntry(
             dashboard_entry_response,
@@ -1169,8 +1177,12 @@ impl SignInWithMultipleRolesStrategy {
                 &state.into(),
                 self.user_roles
                     .iter()
-                    .map(|role| role.merchant_id.clone())
-                    .collect(),
+                    .map(|role| {
+                        role.merchant_id
+                            .clone()
+                            .ok_or(UserErrors::InternalServerError)
+                    })
+                    .collect::<Result<Vec<_>, _>>()?,
             )
             .await
             .change_context(UserErrors::InternalServerError)?;
diff --git a/crates/router/src/types/domain/user/decision_manager.rs b/crates/router/src/types/domain/user/decision_manager.rs
index 97fb69074a6..cffdd05448a 100644
--- a/crates/router/src/types/domain/user/decision_manager.rs
+++ b/crates/router/src/types/domain/user/decision_manager.rs
@@ -1,5 +1,6 @@
 use common_enums::TokenPurpose;
 use diesel_models::{enums::UserStatus, user_role::UserRole};
+use error_stack::report;
 use masking::Secret;
 
 use super::UserFromStorage;
@@ -108,10 +109,16 @@ impl JWTFlow {
     ) -> UserResult<Secret<String>> {
         auth::AuthToken::new_token(
             next_flow.user.get_user_id().to_string(),
-            user_role.merchant_id.clone(),
+            user_role
+                .merchant_id
+                .clone()
+                .ok_or(report!(UserErrors::InternalServerError))?,
             user_role.role_id.clone(),
             &state.conf,
-            user_role.org_id.clone(),
+            user_role
+                .org_id
+                .clone()
+                .ok_or(report!(UserErrors::InternalServerError))?,
         )
         .await
         .map(|token| token.into())
diff --git a/crates/router/src/utils/user.rs b/crates/router/src/utils/user.rs
index 58215bf1bce..72d2b83c6b4 100644
--- a/crates/router/src/utils/user.rs
+++ b/crates/router/src/utils/user.rs
@@ -6,7 +6,7 @@ use common_utils::{
     encryption::Encryption, errors::CustomResult, id_type, types::keymanager::Identifier,
 };
 use diesel_models::{enums::UserStatus, user_role::UserRole};
-use error_stack::ResultExt;
+use error_stack::{report, ResultExt};
 use masking::{ExposeInterface, Secret};
 use redis_interface::RedisConnectionPool;
 
@@ -88,10 +88,20 @@ pub async fn generate_jwt_auth_token(
 ) -> UserResult<Secret<String>> {
     let token = AuthToken::new_token(
         user.get_user_id().to_string(),
-        user_role.merchant_id.clone(),
+        user_role
+            .merchant_id
+            .as_ref()
+            .ok_or(report!(UserErrors::InternalServerError))
+            .attach_printable("merchant_id not found for user_role")?
+            .clone(),
         user_role.role_id.clone(),
         &state.conf,
-        user_role.org_id.clone(),
+        user_role
+            .org_id
+            .as_ref()
+            .ok_or(report!(UserErrors::InternalServerError))
+            .attach_printable("org_id not found for user_role")?
+            .clone(),
     )
     .await?;
     Ok(Secret::new(token))
@@ -124,7 +134,10 @@ pub fn get_dashboard_entry_response(
     let verification_days_left = get_verification_days_left(state, &user)?;
 
     Ok(user_api::DashboardEntryResponse {
-        merchant_id: user_role.merchant_id,
+        merchant_id: user_role.merchant_id.ok_or(
+            report!(UserErrors::InternalServerError)
+                .attach_printable("merchant_id not found for user_role"),
+        )?,
         token,
         name: user.get_name(),
         email: user.get_email(),
@@ -163,8 +176,16 @@ pub fn get_multiple_merchant_details_with_status(
     user_roles
         .into_iter()
         .map(|user_role| {
+            let Some(merchant_id) = &user_role.merchant_id else {
+                return Err(report!(UserErrors::InternalServerError))
+                    .attach_printable("merchant_id not found for user_role");
+            };
+            let Some(org_id) = &user_role.org_id else {
+                return Err(report!(UserErrors::InternalServerError)
+                    .attach_printable("org_id not found in user_role"));
+            };
             let merchant_account = merchant_account_map
-                .get(&user_role.merchant_id)
+                .get(merchant_id)
                 .ok_or(UserErrors::InternalServerError)
                 .attach_printable("Merchant account for user role doesn't exist")?;
 
@@ -174,12 +195,12 @@ pub fn get_multiple_merchant_details_with_status(
                 .attach_printable("Role info for user role doesn't exist")?;
 
             Ok(user_api::UserMerchantAccount {
-                merchant_id: user_role.merchant_id,
+                merchant_id: merchant_id.to_owned(),
                 merchant_name: merchant_account.merchant_name.clone(),
                 is_active: user_role.status == UserStatus::Active,
                 role_id: user_role.role_id,
                 role_name: role_info.get_role_name().to_string(),
-                org_id: user_role.org_id,
+                org_id: org_id.to_owned(),
             })
         })
         .collect()
diff --git a/crates/router/src/utils/user_role.rs b/crates/router/src/utils/user_role.rs
index 99c020e0ee1..4fcc03b1904 100644
--- a/crates/router/src/utils/user_role.rs
+++ b/crates/router/src/utils/user_role.rs
@@ -106,11 +106,18 @@ pub async fn set_role_permissions_in_cache_by_user_role(
     state: &SessionState,
     user_role: &UserRole,
 ) -> bool {
+    let Some(ref merchant_id) = user_role.merchant_id else {
+        return false;
+    };
+
+    let Some(ref org_id) = user_role.org_id else {
+        return false;
+    };
     set_role_permissions_in_cache_if_required(
         state,
         user_role.role_id.as_str(),
-        &user_role.merchant_id,
-        &user_role.org_id,
+        merchant_id,
+        org_id,
     )
     .await
     .map_err(|e| logger::error!("Error setting permissions in cache {:?}", e))
@@ -149,15 +156,18 @@ pub async fn get_multiple_role_info_for_user_roles(
     user_roles: &[UserRole],
 ) -> UserResult<Vec<roles::RoleInfo>> {
     futures::future::try_join_all(user_roles.iter().map(|user_role| async {
-        let role = roles::RoleInfo::from_role_id(
-            state,
-            &user_role.role_id,
-            &user_role.merchant_id,
-            &user_role.org_id,
-        )
-        .await
-        .to_not_found_response(UserErrors::InternalServerError)
-        .attach_printable("Role for user role doesn't exist")?;
+        let Some(merchant_id) = &user_role.merchant_id else {
+            return Err(report!(UserErrors::InternalServerError))
+                .attach_printable("merchant_id not found for user_role");
+        };
+        let Some(org_id) = &user_role.org_id else {
+            return Err(report!(UserErrors::InternalServerError)
+                .attach_printable("org_id not found in user_role"));
+        };
+        let role = roles::RoleInfo::from_role_id(state, &user_role.role_id, merchant_id, org_id)
+            .await
+            .to_not_found_response(UserErrors::InternalServerError)
+            .attach_printable("Role for user role doesn't exist")?;
         Ok::<_, error_stack::Report<UserErrors>>(role)
     }))
     .await
diff --git a/migrations/2024-07-23-100214_make_org_and_merchant_id_nullable_user_roles/down.sql b/migrations/2024-07-23-100214_make_org_and_merchant_id_nullable_user_roles/down.sql
new file mode 100644
index 00000000000..29d51b1eb42
--- /dev/null
+++ b/migrations/2024-07-23-100214_make_org_and_merchant_id_nullable_user_roles/down.sql
@@ -0,0 +1,13 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE user_roles DROP CONSTRAINT user_roles_pkey;
+ALTER TABLE user_roles ADD PRIMARY KEY (user_id, merchant_id);
+
+ALTER TABLE user_roles ALTER COLUMN org_id SET NOT NULL;
+ALTER TABLE user_roles ALTER COLUMN merchant_id SET NOT NULL;
+
+ALTER TABLE user_roles DROP COLUMN profile_id;
+ALTER TABLE user_roles DROP COLUMN entity_id;
+ALTER TABLE user_roles DROP COLUMN entity_type;
+
+ALTER TABLE user_roles DROP COLUMN version;
+DROP TYPE IF EXISTS "UserRoleVersion";
diff --git a/migrations/2024-07-23-100214_make_org_and_merchant_id_nullable_user_roles/up.sql b/migrations/2024-07-23-100214_make_org_and_merchant_id_nullable_user_roles/up.sql
new file mode 100644
index 00000000000..a281c172ca0
--- /dev/null
+++ b/migrations/2024-07-23-100214_make_org_and_merchant_id_nullable_user_roles/up.sql
@@ -0,0 +1,20 @@
+-- Your SQL goes here
+-- The below query will lock the user_roles table
+-- Running this query is not necessary on higher environments
+-- as the application will work fine without these queries being run
+-- This query should be run after the new version of application is deployed
+ALTER TABLE user_roles DROP CONSTRAINT user_roles_pkey;
+-- Use the `id` column as primary key
+-- This is serial and a not null column
+-- So this query should not fail for not null or duplicate value reasons
+ALTER TABLE user_roles ADD PRIMARY KEY (id);
+
+ALTER TABLE user_roles ALTER COLUMN org_id DROP NOT NULL;
+ALTER TABLE user_roles ALTER COLUMN merchant_id DROP NOT NULL;
+
+ALTER TABLE user_roles ADD COLUMN profile_id VARCHAR(64);
+ALTER TABLE user_roles ADD COLUMN entity_id VARCHAR(64);
+ALTER TABLE user_roles ADD COLUMN entity_type VARCHAR(64);
+
+CREATE TYPE "UserRoleVersion" AS ENUM('v1', 'v2');
+ALTER TABLE user_roles ADD COLUMN version "UserRoleVersion" DEFAULT 'v1' NOT NULL;