feat(user): setup user tables (#2803)

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Sahkal Poddar <sahkalplanet@gmail.com>
Co-authored-by: Sahkal Poddar <sahkal.poddar@juspay.in>
Co-authored-by: Sai Harsha Vardhan <56996463+sai-harsha-vardhan@users.noreply.github.com>
Co-authored-by: Venkatesh <inventvenkat@gmail.com>
Co-authored-by: venkatesh.devendran <venkatesh.devendran@juspay.in>
Co-authored-by: Abhishek Marrivagu <68317979+Abhicodes-crypto@users.noreply.github.com>

crates/diesel_models/src/enums.rs

@@ -401,3 +401,25 @@ pub enum FraudCheckLastStep {
     TransactionOrRecordRefund,
     Fulfillment,
 }
+
+#[derive(
+    Clone,
+    Copy,
+    Debug,
+    Default,
+    Eq,
+    PartialEq,
+    serde::Serialize,
+    serde::Deserialize,
+    strum::Display,
+    strum::EnumString,
+    frunk::LabelledGeneric,
+)]
+#[router_derive::diesel_enum(storage_type = "text")]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum UserStatus {
+    Active,
+    #[default]
+    InvitationSent,
+}

crates/diesel_models/src/lib.rs

@@ -38,6 +38,8 @@ pub mod reverse_lookup;
 pub mod routing_algorithm;
 #[allow(unused_qualifications)]
 pub mod schema;
+pub mod user;
+pub mod user_role;
 
 use diesel_impl::{DieselArray, OptionalDieselArray};
 

crates/diesel_models/src/query.rs

@@ -28,3 +28,5 @@ pub mod process_tracker;
 pub mod refund;
 pub mod reverse_lookup;
 pub mod routing_algorithm;
+pub mod user;
+pub mod user_role;

crates/diesel_models/src/query/user.rs

@@ -0,0 +1,62 @@
+use diesel::{associations::HasTable, ExpressionMethods};
+use error_stack::report;
+use router_env::tracing::{self, instrument};
+
+use crate::{
+    errors::{self},
+    query::generics,
+    schema::users::dsl,
+    user::*,
+    PgPooledConn, StorageResult,
+};
+
+impl UserNew {
+    #[instrument(skip(conn))]
+    pub async fn insert(self, conn: &PgPooledConn) -> StorageResult<User> {
+        generics::generic_insert(conn, self).await
+    }
+}
+
+impl User {
+    pub async fn find_by_user_email(conn: &PgPooledConn, user_email: &str) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::email.eq(user_email.to_owned()),
+        )
+        .await
+    }
+
+    pub async fn find_by_user_id(conn: &PgPooledConn, user_id: &str) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::user_id.eq(user_id.to_owned()),
+        )
+        .await
+    }
+
+    pub async fn update_by_user_id(
+        conn: &PgPooledConn,
+        user_id: &str,
+        user: UserUpdate,
+    ) -> StorageResult<Self> {
+        generics::generic_update_with_results::<<Self as HasTable>::Table, _, _, _>(
+            conn,
+            dsl::user_id.eq(user_id.to_owned()),
+            UserUpdateInternal::from(user),
+        )
+        .await?
+        .first()
+        .cloned()
+        .ok_or_else(|| {
+            report!(errors::DatabaseError::NotFound).attach_printable("Error while updating user")
+        })
+    }
+
+    pub async fn delete_by_user_id(conn: &PgPooledConn, user_id: &str) -> StorageResult<bool> {
+        generics::generic_delete::<<Self as HasTable>::Table, _>(
+            conn,
+            dsl::user_id.eq(user_id.to_owned()),
+        )
+        .await
+    }
+}

crates/diesel_models/src/query/user_role.rs

@@ -0,0 +1,58 @@
+use diesel::{associations::HasTable, BoolExpressionMethods, ExpressionMethods};
+use router_env::tracing::{self, instrument};
+
+use crate::{query::generics, schema::user_roles::dsl, user_role::*, PgPooledConn, StorageResult};
+
+impl UserRoleNew {
+    #[instrument(skip(conn))]
+    pub async fn insert(self, conn: &PgPooledConn) -> StorageResult<UserRole> {
+        generics::generic_insert(conn, self).await
+    }
+}
+
+impl UserRole {
+    pub async fn find_by_user_id(conn: &PgPooledConn, user_id: String) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::user_id.eq(user_id),
+        )
+        .await
+    }
+
+    pub async fn update_by_user_id_merchant_id(
+        conn: &PgPooledConn,
+        user_id: String,
+        merchant_id: String,
+        update: UserRoleUpdate,
+    ) -> StorageResult<Self> {
+        generics::generic_update_with_unique_predicate_get_result::<
+            <Self as HasTable>::Table,
+            _,
+            _,
+            _,
+        >(
+            conn,
+            dsl::user_id
+                .eq(user_id)
+                .and(dsl::merchant_id.eq(merchant_id)),
+            UserRoleUpdateInternal::from(update),
+        )
+        .await
+    }
+
+    pub async fn delete_by_user_id(conn: &PgPooledConn, user_id: String) -> StorageResult<bool> {
+        generics::generic_delete::<<Self as HasTable>::Table, _>(conn, dsl::user_id.eq(user_id))
+            .await
+    }
+
+    pub async fn list_by_user_id(conn: &PgPooledConn, user_id: String) -> StorageResult<Vec<Self>> {
+        generics::generic_filter::<<Self as HasTable>::Table, _, _, _>(
+            conn,
+            dsl::user_id.eq(user_id),
+            None,
+            None,
+            Some(dsl::created_at.asc()),
+        )
+        .await
+    }
+}

crates/diesel_models/src/schema.rs

@@ -900,6 +900,51 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    use diesel::sql_types::*;
+    use crate::enums::diesel_exports::*;
+
+    user_roles (id) {
+        id -> Int4,
+        #[max_length = 64]
+        user_id -> Varchar,
+        #[max_length = 64]
+        merchant_id -> Varchar,
+        #[max_length = 64]
+        role_id -> Varchar,
+        #[max_length = 64]
+        org_id -> Varchar,
+        #[max_length = 64]
+        status -> Varchar,
+        #[max_length = 64]
+        created_by -> Varchar,
+        #[max_length = 64]
+        last_modified_by -> Varchar,
+        created_at -> Timestamp,
+        last_modified_at -> Timestamp,
+    }
+}
+
+diesel::table! {
+    use diesel::sql_types::*;
+    use crate::enums::diesel_exports::*;
+
+    users (id) {
+        id -> Int4,
+        #[max_length = 64]
+        user_id -> Varchar,
+        #[max_length = 255]
+        email -> Varchar,
+        #[max_length = 255]
+        name -> Varchar,
+        #[max_length = 255]
+        password -> Varchar,
+        is_verified -> Bool,
+        created_at -> Timestamp,
+        last_modified_at -> Timestamp,
+    }
+}
+
 diesel::allow_tables_to_appear_in_same_query!(
     address,
     api_keys,
@@ -929,4 +974,6 @@ diesel::allow_tables_to_appear_in_same_query!(
     refund,
     reverse_lookup,
     routing_algorithm,
+    user_roles,
+    users,
 );

crates/diesel_models/src/user.rs

@@ -0,0 +1,76 @@
+use common_utils::pii;
+use diesel::{AsChangeset, Identifiable, Insertable, Queryable};
+use masking::Secret;
+use time::PrimitiveDateTime;
+
+use crate::schema::users;
+
+#[derive(Clone, Debug, Identifiable, Queryable)]
+#[diesel(table_name = users)]
+pub struct User {
+    pub id: i32,
+    pub user_id: String,
+    pub email: pii::Email,
+    pub name: Secret<String>,
+    pub password: Secret<String>,
+    pub is_verified: bool,
+    pub created_at: PrimitiveDateTime,
+    pub last_modified_at: PrimitiveDateTime,
+}
+
+#[derive(
+    router_derive::Setter, Clone, Debug, Default, Insertable, router_derive::DebugAsDisplay,
+)]
+#[diesel(table_name = users)]
+pub struct UserNew {
+    pub user_id: String,
+    pub email: pii::Email,
+    pub name: Secret<String>,
+    pub password: Secret<String>,
+    pub is_verified: bool,
+    pub created_at: Option<PrimitiveDateTime>,
+    pub last_modified_at: Option<PrimitiveDateTime>,
+}
+
+#[derive(Clone, Debug, AsChangeset, router_derive::DebugAsDisplay)]
+#[diesel(table_name = users)]
+pub struct UserUpdateInternal {
+    name: Option<String>,
+    password: Option<Secret<String>>,
+    is_verified: Option<bool>,
+    last_modified_at: PrimitiveDateTime,
+}
+
+#[derive(Debug)]
+pub enum UserUpdate {
+    VerifyUser,
+    AccountUpdate {
+        name: Option<String>,
+        password: Option<Secret<String>>,
+        is_verified: Option<bool>,
+    },
+}
+
+impl From<UserUpdate> for UserUpdateInternal {
+    fn from(user_update: UserUpdate) -> Self {
+        let last_modified_at = common_utils::date_time::now();
+        match user_update {
+            UserUpdate::VerifyUser => Self {
+                name: None,
+                password: None,
+                is_verified: Some(true),
+                last_modified_at,
+            },
+            UserUpdate::AccountUpdate {
+                name,
+                password,
+                is_verified,
+            } => Self {
+                name,
+                password,
+                is_verified,
+                last_modified_at,
+            },
+        }
+    }
+}

crates/diesel_models/src/user_role.rs

@@ -0,0 +1,79 @@
+use diesel::{AsChangeset, Identifiable, Insertable, Queryable};
+use time::PrimitiveDateTime;
+
+use crate::{enums, schema::user_roles};
+
+#[derive(Clone, Debug, Identifiable, Queryable)]
+#[diesel(table_name = user_roles)]
+pub struct UserRole {
+    pub id: i32,
+    pub user_id: String,
+    pub merchant_id: String,
+    pub role_id: String,
+    pub org_id: String,
+    pub status: enums::UserStatus,
+    pub created_by: String,
+    pub last_modified_by: String,
+    pub created_at: PrimitiveDateTime,
+    pub last_modified_at: PrimitiveDateTime,
+}
+
+#[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: String,
+    pub role_id: String,
+    pub org_id: String,
+    pub status: enums::UserStatus,
+    pub created_by: String,
+    pub last_modified_by: String,
+    pub created_at: PrimitiveDateTime,
+    pub last_modified_at: PrimitiveDateTime,
+}
+
+#[derive(Clone, Debug, AsChangeset, router_derive::DebugAsDisplay)]
+#[diesel(table_name = user_roles)]
+pub struct UserRoleUpdateInternal {
+    role_id: Option<String>,
+    status: Option<enums::UserStatus>,
+    last_modified_by: Option<String>,
+    last_modified_at: PrimitiveDateTime,
+}
+
+pub enum UserRoleUpdate {
+    UpdateStatus {
+        status: enums::UserStatus,
+        modified_by: String,
+    },
+    UpdateRole {
+        role_id: String,
+        modified_by: String,
+    },
+}
+
+impl From<UserRoleUpdate> for UserRoleUpdateInternal {
+    fn from(value: UserRoleUpdate) -> Self {
+        let last_modified_at = common_utils::date_time::now();
+        match value {
+            UserRoleUpdate::UpdateRole {
+                role_id,
+                modified_by,
+            } => Self {
+                role_id: Some(role_id),
+                last_modified_by: Some(modified_by),
+                status: None,
+                last_modified_at,
+            },
+            UserRoleUpdate::UpdateStatus {
+                status,
+                modified_by,
+            } => Self {
+                status: Some(status),
+                last_modified_at,
+                last_modified_by: Some(modified_by),
+                role_id: None,
+            },
+        }
+    }
+}

crates/router/src/db.rs

@@ -25,6 +25,8 @@ pub mod payouts;
 pub mod refund;
 pub mod reverse_lookup;
 pub mod routing_algorithm;
+pub mod user;
+pub mod user_role;
 
 use data_models::payments::{
     payment_attempt::PaymentAttemptInterface, payment_intent::PaymentIntentInterface,
@@ -80,6 +82,8 @@ pub trait StorageInterface:
     + organization::OrganizationInterface
     + routing_algorithm::RoutingAlgorithmInterface
     + gsm::GsmInterface
+    + user::UserInterface
+    + user_role::UserRoleInterface
     + 'static
 {
     fn get_scheduler_db(&self) -> Box<dyn scheduler::SchedulerInterface>;

crates/router/src/db/user.rs

@@ -0,0 +1,265 @@
+use diesel_models::user as storage;
+use error_stack::{IntoReport, ResultExt};
+use masking::Secret;
+
+use super::MockDb;
+use crate::{
+    connection,
+    core::errors::{self, CustomResult},
+    services::Store,
+};
+
+#[async_trait::async_trait]
+pub trait UserInterface {
+    async fn insert_user(
+        &self,
+        user_data: storage::UserNew,
+    ) -> CustomResult<storage::User, errors::StorageError>;
+
+    async fn find_user_by_email(
+        &self,
+        user_email: &str,
+    ) -> CustomResult<storage::User, errors::StorageError>;
+
+    async fn find_user_by_id(
+        &self,
+        user_id: &str,
+    ) -> CustomResult<storage::User, errors::StorageError>;
+
+    async fn update_user_by_user_id(
+        &self,
+        user_id: &str,
+        user: storage::UserUpdate,
+    ) -> CustomResult<storage::User, errors::StorageError>;
+
+    async fn delete_user_by_user_id(
+        &self,
+        user_id: &str,
+    ) -> CustomResult<bool, errors::StorageError>;
+}
+
+#[async_trait::async_trait]
+impl UserInterface for Store {
+    async fn insert_user(
+        &self,
+        user_data: storage::UserNew,
+    ) -> CustomResult<storage::User, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        user_data
+            .insert(&conn)
+            .await
+            .map_err(Into::into)
+            .into_report()
+    }
+
+    async fn find_user_by_email(
+        &self,
+        user_email: &str,
+    ) -> CustomResult<storage::User, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        storage::User::find_by_user_email(&conn, user_email)
+            .await
+            .map_err(Into::into)
+            .into_report()
+    }
+
+    async fn find_user_by_id(
+        &self,
+        user_id: &str,
+    ) -> CustomResult<storage::User, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        storage::User::find_by_user_id(&conn, user_id)
+            .await
+            .map_err(Into::into)
+            .into_report()
+    }
+
+    async fn update_user_by_user_id(
+        &self,
+        user_id: &str,
+        user: storage::UserUpdate,
+    ) -> CustomResult<storage::User, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        storage::User::update_by_user_id(&conn, user_id, user)
+            .await
+            .map_err(Into::into)
+            .into_report()
+    }
+
+    async fn delete_user_by_user_id(
+        &self,
+        user_id: &str,
+    ) -> CustomResult<bool, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        storage::User::delete_by_user_id(&conn, user_id)
+            .await
+            .map_err(Into::into)
+            .into_report()
+    }
+}
+
+#[async_trait::async_trait]
+impl UserInterface for MockDb {
+    async fn insert_user(
+        &self,
+        user_data: storage::UserNew,
+    ) -> CustomResult<storage::User, errors::StorageError> {
+        let mut users = self.users.lock().await;
+        if users
+            .iter()
+            .any(|user| user.email == user_data.email || user.user_id == user_data.user_id)
+        {
+            Err(errors::StorageError::DuplicateValue {
+                entity: "email or user_id",
+                key: None,
+            })?
+        }
+        let time_now = common_utils::date_time::now();
+        let user = storage::User {
+            id: users
+                .len()
+                .try_into()
+                .into_report()
+                .change_context(errors::StorageError::MockDbError)?,
+            user_id: user_data.user_id,
+            email: user_data.email,
+            name: user_data.name,
+            password: user_data.password,
+            is_verified: user_data.is_verified,
+            created_at: user_data.created_at.unwrap_or(time_now),
+            last_modified_at: user_data.created_at.unwrap_or(time_now),
+        };
+        users.push(user.clone());
+        Ok(user)
+    }
+
+    async fn find_user_by_email(
+        &self,
+        user_email: &str,
+    ) -> CustomResult<storage::User, errors::StorageError> {
+        let users = self.users.lock().await;
+        let user_email_pii: common_utils::pii::Email = user_email
+            .to_string()
+            .try_into()
+            .map_err(|_| errors::StorageError::MockDbError)?;
+        users
+            .iter()
+            .find(|user| user.email == user_email_pii)
+            .cloned()
+            .ok_or(
+                errors::StorageError::ValueNotFound(format!(
+                    "No user available for email = {user_email}"
+                ))
+                .into(),
+            )
+    }
+
+    async fn find_user_by_id(
+        &self,
+        user_id: &str,
+    ) -> CustomResult<storage::User, errors::StorageError> {
+        let users = self.users.lock().await;
+        users
+            .iter()
+            .find(|user| user.user_id == user_id)
+            .cloned()
+            .ok_or(
+                errors::StorageError::ValueNotFound(format!(
+                    "No user available for user_id = {user_id}"
+                ))
+                .into(),
+            )
+    }
+
+    async fn update_user_by_user_id(
+        &self,
+        user_id: &str,
+        update_user: storage::UserUpdate,
+    ) -> CustomResult<storage::User, errors::StorageError> {
+        let mut users = self.users.lock().await;
+        users
+            .iter_mut()
+            .find(|user| user.user_id == user_id)
+            .map(|user| {
+                *user = match &update_user {
+                    storage::UserUpdate::VerifyUser => storage::User {
+                        is_verified: true,
+                        ..user.to_owned()
+                    },
+                    storage::UserUpdate::AccountUpdate {
+                        name,
+                        password,
+                        is_verified,
+                    } => storage::User {
+                        name: name.clone().map(Secret::new).unwrap_or(user.name.clone()),
+                        password: password.clone().unwrap_or(user.password.clone()),
+                        is_verified: is_verified.unwrap_or(user.is_verified),
+                        ..user.to_owned()
+                    },
+                };
+                user.to_owned()
+            })
+            .ok_or(
+                errors::StorageError::ValueNotFound(format!(
+                    "No user available for user_id = {user_id}"
+                ))
+                .into(),
+            )
+    }
+
+    async fn delete_user_by_user_id(
+        &self,
+        user_id: &str,
+    ) -> CustomResult<bool, errors::StorageError> {
+        let mut users = self.users.lock().await;
+        let user_index = users
+            .iter()
+            .position(|user| user.user_id == user_id)
+            .ok_or(errors::StorageError::ValueNotFound(format!(
+                "No user available for user_id = {user_id}"
+            )))?;
+        users.remove(user_index);
+        Ok(true)
+    }
+}
+#[cfg(feature = "kafka_events")]
+#[async_trait::async_trait]
+impl UserInterface for super::KafkaStore {
+    async fn insert_user(
+        &self,
+        user_data: storage::UserNew,
+    ) -> CustomResult<storage::User, errors::StorageError> {
+        self.diesel_store.insert_user(user_data).await
+    }
+
+    async fn find_user_by_email(
+        &self,
+        user_email: &str,
+    ) -> CustomResult<storage::User, errors::StorageError> {
+        self.diesel_store.find_user_by_email(user_email).await
+    }
+
+    async fn find_user_by_id(
+        &self,
+        user_id: &str,
+    ) -> CustomResult<storage::User, errors::StorageError> {
+        self.diesel_store.find_user_by_id(user_id).await
+    }
+
+    async fn update_user_by_user_id(
+        &self,
+        user_id: &str,
+        user: storage::UserUpdate,
+    ) -> CustomResult<storage::User, errors::StorageError> {
+        self.diesel_store
+            .update_user_by_user_id(user_id, user)
+            .await
+    }
+
+    async fn delete_user_by_user_id(
+        &self,
+        user_id: &str,
+    ) -> CustomResult<bool, errors::StorageError> {
+        self.diesel_store.delete_user_by_user_id(user_id).await
+    }
+}

crates/router/src/db/user_role.rs

@@ -0,0 +1,255 @@
+use diesel_models::user_role as storage;
+use error_stack::{IntoReport, ResultExt};
+
+use super::MockDb;
+use crate::{
+    connection,
+    core::errors::{self, CustomResult},
+    services::Store,
+};
+
+#[async_trait::async_trait]
+pub trait UserRoleInterface {
+    async fn insert_user_role(
+        &self,
+        user_role: storage::UserRoleNew,
+    ) -> CustomResult<storage::UserRole, errors::StorageError>;
+    async fn find_user_role_by_user_id(
+        &self,
+        user_id: &str,
+    ) -> CustomResult<storage::UserRole, errors::StorageError>;
+    async fn update_user_role_by_user_id_merchant_id(
+        &self,
+        user_id: &str,
+        merchant_id: &str,
+        update: storage::UserRoleUpdate,
+    ) -> CustomResult<storage::UserRole, errors::StorageError>;
+    async fn delete_user_role(&self, user_id: &str) -> CustomResult<bool, errors::StorageError>;
+
+    async fn list_user_roles_by_user_id(
+        &self,
+        user_id: &str,
+    ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError>;
+}
+
+#[async_trait::async_trait]
+impl UserRoleInterface for Store {
+    async fn insert_user_role(
+        &self,
+        user_role: storage::UserRoleNew,
+    ) -> CustomResult<storage::UserRole, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        user_role
+            .insert(&conn)
+            .await
+            .map_err(Into::into)
+            .into_report()
+    }
+
+    async fn find_user_role_by_user_id(
+        &self,
+        user_id: &str,
+    ) -> CustomResult<storage::UserRole, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        storage::UserRole::find_by_user_id(&conn, user_id.to_owned())
+            .await
+            .map_err(Into::into)
+            .into_report()
+    }
+
+    async fn update_user_role_by_user_id_merchant_id(
+        &self,
+        user_id: &str,
+        merchant_id: &str,
+        update: storage::UserRoleUpdate,
+    ) -> CustomResult<storage::UserRole, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        storage::UserRole::update_by_user_id_merchant_id(
+            &conn,
+            user_id.to_owned(),
+            merchant_id.to_owned(),
+            update,
+        )
+        .await
+        .map_err(Into::into)
+        .into_report()
+    }
+
+    async fn delete_user_role(&self, user_id: &str) -> CustomResult<bool, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        storage::UserRole::delete_by_user_id(&conn, user_id.to_owned())
+            .await
+            .map_err(Into::into)
+            .into_report()
+    }
+
+    async fn list_user_roles_by_user_id(
+        &self,
+        user_id: &str,
+    ) -> 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())
+            .await
+            .map_err(Into::into)
+            .into_report()
+    }
+}
+
+#[async_trait::async_trait]
+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: user_roles
+                .len()
+                .try_into()
+                .into_report()
+                .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_at: user_role.last_modified_at,
+            last_modified_by: user_role.last_modified_by,
+            org_id: user_role.org_id,
+        };
+        user_roles.push(user_role.clone());
+        Ok(user_role)
+    }
+
+    async fn find_user_role_by_user_id(
+        &self,
+        user_id: &str,
+    ) -> 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)
+            .cloned()
+            .ok_or(
+                errors::StorageError::ValueNotFound(format!(
+                    "No user role available for user_id = {user_id}"
+                ))
+                .into(),
+            )
+    }
+
+    async fn update_user_role_by_user_id_merchant_id(
+        &self,
+        user_id: &str,
+        merchant_id: &str,
+        update: storage::UserRoleUpdate,
+    ) -> 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 {
+                    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()
+                    },
+                    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.to_owned()
+            })
+            .ok_or(
+                errors::StorageError::ValueNotFound(format!(
+                    "No user role available for user_id = {user_id} and merchant_id = {merchant_id}"
+                ))
+                .into(),
+            )
+    }
+
+    async fn delete_user_role(&self, user_id: &str) -> CustomResult<bool, errors::StorageError> {
+        let mut user_roles = self.user_roles.lock().await;
+        let user_role_index = user_roles
+            .iter()
+            .position(|user_role| user_role.user_id == user_id)
+            .ok_or(errors::StorageError::ValueNotFound(format!(
+                "No user available for user_id = {user_id}"
+            )))?;
+        user_roles.remove(user_role_index);
+        Ok(true)
+    }
+
+    async fn list_user_roles_by_user_id(
+        &self,
+        user_id: &str,
+    ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError> {
+        let user_roles = self.user_roles.lock().await;
+
+        Ok(user_roles
+            .iter()
+            .cloned()
+            .filter_map(|ele| {
+                if ele.user_id == user_id {
+                    return Some(ele);
+                }
+                None
+            })
+            .collect())
+    }
+}
+
+#[cfg(feature = "kafka_events")]
+#[async_trait::async_trait]
+impl UserRoleInterface for super::KafkaStore {
+    async fn insert_user_role(
+        &self,
+        user_role: storage::UserRoleNew,
+    ) -> CustomResult<storage::UserRole, errors::StorageError> {
+        self.diesel_store.insert_user_role(user_role).await
+    }
+    async fn update_user_role_by_user_id_merchant_id(
+        &self,
+        user_id: &str,
+        merchant_id: &str,
+        update: storage::UserRoleUpdate,
+    ) -> CustomResult<storage::UserRole, errors::StorageError> {
+        self.diesel_store
+            .update_user_role_by_user_id_merchant_id(user_id, merchant_id, update)
+            .await
+    }
+    async fn find_user_role_by_user_id(
+        &self,
+        user_id: &str,
+    ) -> CustomResult<storage::UserRole, errors::StorageError> {
+        self.diesel_store.find_user_role_by_user_id(user_id).await
+    }
+    async fn delete_user_role(&self, user_id: &str) -> CustomResult<bool, errors::StorageError> {
+        self.diesel_store.delete_user_role(user_id).await
+    }
+    async fn list_user_roles_by_user_id(
+        &self,
+        user_id: &str,
+    ) -> CustomResult<Vec<storage::UserRole>, errors::StorageError> {
+        self.diesel_store.list_user_roles_by_user_id(user_id).await
+    }
+}

crates/router/src/types/storage.rs

@@ -32,6 +32,8 @@ pub mod payout_attempt;
 pub mod payouts;
 mod query;
 pub mod refund;
+pub mod user;
+pub mod user_role;
 
 pub use data_models::payments::{
     payment_attempt::{PaymentAttempt, PaymentAttemptNew, PaymentAttemptUpdate},
@@ -44,7 +46,7 @@ pub use self::{
     ephemeral_key::*, events::*, file::*, gsm::*, locker_mock_up::*, mandate::*,
     merchant_account::*, merchant_connector_account::*, merchant_key_store::*, payment_link::*,
     payment_method::*, payout_attempt::*, payouts::*, process_tracker::*, refund::*,
-    reverse_lookup::*, routing_algorithm::*,
+    reverse_lookup::*, routing_algorithm::*, user::*, user_role::*,
 };
 use crate::types::api::routing;
 

crates/router/src/types/storage/user.rs

@@ -0,0 +1 @@
+pub use diesel_models::user::*;

crates/router/src/types/storage/user_role.rs

@@ -0,0 +1 @@
+pub use diesel_models::user_role::*;

crates/storage_impl/src/mock_db.rs

@@ -41,6 +41,8 @@ pub struct MockDb {
     pub reverse_lookups: Arc<Mutex<Vec<store::ReverseLookup>>>,
     pub payment_link: Arc<Mutex<Vec<store::payment_link::PaymentLink>>>,
     pub organizations: Arc<Mutex<Vec<store::organization::Organization>>>,
+    pub users: Arc<Mutex<Vec<store::user::User>>>,
+    pub user_roles: Arc<Mutex<Vec<store::user_role::UserRole>>>,
 }
 
 impl MockDb {
@@ -74,6 +76,8 @@ impl MockDb {
             reverse_lookups: Default::default(),
             payment_link: Default::default(),
             organizations: Default::default(),
+            users: Default::default(),
+            user_roles: Default::default(),
         })
     }
 }

migrations/2023-11-06-110233_create_user_table/down.sql

@@ -0,0 +1,2 @@
+-- This file should undo anything in `up.sql`
+DROP TABLE users;

migrations/2023-11-06-110233_create_user_table/up.sql

@@ -0,0 +1,14 @@
+-- Your SQL goes here
+CREATE TABLE IF NOT EXISTS users (
+    id SERIAL PRIMARY KEY,
+    user_id VARCHAR(64) NOT NULL UNIQUE,
+    email VARCHAR(255) NOT NULL UNIQUE,
+    name VARCHAR(255) NOT NULL,
+    password VARCHAR(255) NOT NULL,
+    is_verified bool NOT NULL DEFAULT false,
+    created_at TIMESTAMP NOT NULL DEFAULT now(),
+    last_modified_at TIMESTAMP NOT NULL DEFAULT now()
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS user_id_index ON users (user_id);
+CREATE UNIQUE INDEX IF NOT EXISTS user_email_index ON users (email);

migrations/2023-11-06-113726_create_user_roles_table/down.sql

@@ -0,0 +1,4 @@
+-- This file should undo anything in `up.sql`
+
+-- Drop the table
+DROP TABLE IF EXISTS user_roles;

migrations/2023-11-06-113726_create_user_roles_table/up.sql

@@ -0,0 +1,18 @@
+-- Your SQL goes here
+CREATE TABLE IF NOT EXISTS user_roles (
+	id SERIAL PRIMARY KEY,
+	user_id VARCHAR(64) NOT NULL,
+	merchant_id VARCHAR(64) NOT NULL,
+	role_id VARCHAR(64) NOT NULL,
+	org_id VARCHAR(64) NOT NULL, 
+	status VARCHAR(64) NOT NULL, 
+	created_by VARCHAR(64) NOT NULL,
+	last_modified_by VARCHAR(64) NOT NULL,
+	created_at TIMESTAMP NOT NULL DEFAULT now(),
+	last_modified_at TIMESTAMP NOT NULL DEFAULT now(),
+	CONSTRAINT user_merchant_unique UNIQUE (user_id, merchant_id)
+);
+
+
+CREATE INDEX IF NOT EXISTS  user_id_roles_index ON user_roles (user_id);
+CREATE INDEX IF NOT EXISTS  user_mid_roles_index ON user_roles (merchant_id);