Refactor User

src/api/v1/auth/csh.rs

@@ -1,14 +1,16 @@
-use crate::api::v1::auth::models::UserRealm;
 use crate::api::v1::auth::models::{CSHUserInfo, UserInfo};
 use crate::app::AppState;
+use crate::db::user::{UserData, UserRealm};
 use actix_session::Session;
 use actix_web::http::header;
 use actix_web::{get, Scope};
 use actix_web::{web, HttpResponse, Responder};
+use log::error;
 use oauth2::reqwest::async_http_client;
 use oauth2::{AuthorizationCode, TokenResponse};
 use reqwest::Client;
 use serde::Deserialize;
+use serde_json::json;
 use utoipa::{OpenApi, ToSchema};
 
 use crate::api::v1::auth::common;
@@ -70,15 +72,20 @@ async fn auth(
         .await
         .unwrap();
 
-    sqlx::query!(
-        "INSERT INTO users (id, realm, name, email) VALUES ($1, $2, $3, $4) ON CONFLICT (id) DO UPDATE SET realm = EXCLUDED.realm, name = EXCLUDED.name, email = EXCLUDED.email;",
-        user_info.ldap_id,
-        UserRealm::Csh as _,
+    if let Err(err) = UserData::insert_new(
+        user_info.ldap_id.clone(),
+        UserRealm::Csh,
         format!("{} {}", user_info.given_name, user_info.family_name),
-        user_info.email
+        user_info.email.clone(),
+        &data.db,
     )
-    .execute(&data.db)
-    .await.unwrap();
+    .await
+    {
+        error!("{}", err);
+        return HttpResponse::InternalServerError().json(json!({
+            "error": "Failed to add user to database"
+        }));
+    }
 
     session.insert("login", true).unwrap();
     session

src/api/v1/auth/google.rs

@@ -1,15 +1,17 @@
 use crate::api::v1::auth::common;
-use crate::api::v1::auth::models::UserRealm;
 use crate::api::v1::auth::models::{GoogleUserInfo, UserInfo};
 use crate::app::AppState;
+use crate::db::user::{UserData, UserRealm};
 use actix_session::Session;
 use actix_web::http::header;
 use actix_web::{get, web, Scope};
 use actix_web::{HttpResponse, Responder};
+use log::error;
 use oauth2::reqwest::async_http_client;
 use oauth2::{AuthorizationCode, TokenResponse};
 use reqwest::Client;
 use serde::Deserialize;
+use serde_json::json;
 use utoipa::{OpenApi, ToSchema};
 
 #[derive(OpenApi)]
@@ -75,15 +77,20 @@ async fn auth(
         .await
         .unwrap();
 
-    sqlx::query!(
-        "INSERT INTO users (id, realm, name, email) VALUES ($1, $2, $3, $4) ON CONFLICT (id) DO UPDATE SET realm = EXCLUDED.realm, name = EXCLUDED.name, email = EXCLUDED.email;",
-        user_info.sub,
-        UserRealm::Google as _,
+    if let Err(err) = UserData::insert_new(
+        user_info.sub.clone(),
+        UserRealm::Google,
         format!("{} {}", user_info.given_name, user_info.family_name),
-        user_info.email
+        user_info.email.clone(),
+        &data.db,
     )
-    .execute(&data.db)
-    .await.unwrap();
+    .await
+    {
+        error!("{}", err);
+        return HttpResponse::InternalServerError().json(json!({
+            "error": "Failed to add user to database"
+        }));
+    }
 
     session.insert("login", true).unwrap();
     session

src/api/v1/auth/models.rs

@@ -1,12 +1,4 @@
 use serde::{Deserialize, Serialize};
-use utoipa::ToSchema;
-
-#[derive(Serialize, Deserialize, sqlx::Type)]
-#[sqlx(type_name = "user_realm", rename_all = "lowercase")]
-pub enum UserRealm {
-    Csh,
-    Google,
-}
 
 #[derive(Serialize, Deserialize, Debug)]
 pub struct CSHUserInfo {
@@ -38,15 +30,6 @@ pub struct UserInfo {
     pub groups: Vec<String>,
 }
 
-#[derive(Serialize, Deserialize, sqlx::Type, ToSchema, Clone)]
-#[serde(rename_all = "camelCase")]
-pub struct UserData {
-    pub id: String,
-    pub realm: String,
-    pub name: String,
-    pub email: String,
-}
-
 impl From<CSHUserInfo> for UserInfo {
     fn from(user_info: CSHUserInfo) -> Self {
         let username = user_info.preferred_username;

src/api/v1/event/car/mod.rs

@@ -1,7 +1,7 @@
 use crate::api::v1::auth::models::UserInfo;
 use crate::app::{AppState, MultipleRiderChange, RedisJob};
 use crate::db::car::{Car, CarData};
-use crate::{api::v1::auth::models::UserData, auth::SessionAuth};
+use crate::{auth::SessionAuth, db::user::UserData};
 use actix_session::Session;
 use actix_web::{
     delete, get, post, put,

src/api/v1/event/mod.rs

@@ -17,7 +17,7 @@ use crate::auth::SessionAuth;
 
 use utoipa::OpenApi;
 
-use super::auth::models::UserData;
+use crate::db::user::UserData;
 
 mod car;
 

src/api/v1/user.rs

@@ -7,7 +7,7 @@ use crate::auth::SessionAuth;
 
 use utoipa::OpenApi;
 
-use super::auth::models::UserData;
+use crate::db::user::UserData;
 
 #[derive(OpenApi)]
 #[openapi(paths(user_search), components(schemas(UserData)))]

src/app.rs

@@ -5,7 +5,6 @@ use redis::aio::MultiplexedConnection;
 use redis_work_queue::KeyPrefix;
 use serde::{Deserialize, Serialize};
 use sqlx::PgPool;
-use utoipa::ToSchema;
 
 #[derive(Clone)]
 pub struct AppState {
@@ -18,15 +17,6 @@ pub struct AppState {
     pub csh_userinfo_url: String,
 }
 
-#[derive(Serialize, Deserialize, sqlx::Type, ToSchema, Clone)]
-#[serde(rename_all = "camelCase")]
-pub struct UserData {
-    pub id: String,
-    pub realm: String,
-    pub name: String,
-    pub email: String,
-}
-
 #[derive(Serialize, Deserialize)]
 pub struct SimpleRiderChange {
     pub event_id: i32,

src/db/car.rs

@@ -4,7 +4,7 @@ use serde::{Deserialize, Serialize};
 use sqlx::{query, query_as, Executor, Postgres};
 use utoipa::ToSchema;
 
-use crate::app::UserData;
+use crate::db::user::UserData;
 
 #[derive(Deserialize, ToSchema)]
 #[serde(rename_all = "camelCase")]

src/db/event.rs

@@ -4,7 +4,7 @@ use serde::{Deserialize, Serialize};
 use sqlx::{query_as, Executor, Postgres};
 use utoipa::ToSchema;
 
-use crate::app::UserData;
+use crate::db::user::UserData;
 
 #[derive(Serialize, Deserialize, sqlx::FromRow, ToSchema)]
 #[serde(rename_all = "camelCase")]

src/db/mod.rs

@@ -1,2 +1,3 @@
 pub mod car;
 pub mod event;
+pub mod user;

src/db/user.rs

@@ -0,0 +1,81 @@
+use std::collections::HashMap;
+
+use anyhow::{anyhow, Result};
+use serde::{Deserialize, Serialize};
+use sqlx::{query_as, Executor, Postgres};
+use utoipa::ToSchema;
+
+#[derive(Serialize, Deserialize, sqlx::Type)]
+#[sqlx(type_name = "user_realm", rename_all = "lowercase")]
+pub enum UserRealm {
+    Csh,
+    Google,
+}
+
+#[derive(Serialize, Deserialize, sqlx::Type, ToSchema, Clone)]
+#[serde(rename_all = "camelCase")]
+pub struct UserData {
+    pub id: String,
+    pub realm: String,
+    pub name: String,
+    pub email: String,
+}
+
+impl UserData {
+    pub async fn insert_new<'c, C>(
+        id: String,
+        realm: UserRealm,
+        name: String,
+        email: String,
+        conn: C,
+    ) -> Result<Self>
+    where
+        C: Executor<'c, Database = Postgres>,
+    {
+        query_as!(
+            UserData,
+            r#"INSERT INTO users (id, realm, name, email)
+            VALUES ($1, $2, $3, $4)
+            ON CONFLICT (id) DO UPDATE SET realm = EXCLUDED.realm, name = EXCLUDED.name, email = EXCLUDED.email
+            RETURNING id AS "id!", realm::text AS "realm!", name AS "name!", email AS "email!";"#,
+            id,
+            realm as _,
+            name,
+            email
+        )
+        .fetch_one(conn)
+        .await.map_err(|err| anyhow!("Failed to insert/update user: {}", err))
+    }
+    pub async fn select_map<'c, C>(ids: Vec<String>, conn: C) -> Result<HashMap<String, Self>>
+    where
+        C: Executor<'c, Database = Postgres>,
+    {
+        let data = query_as!(
+            UserData,
+            r#"
+            SELECT id AS "id!", realm::text AS "realm!", name AS "name!", email AS "email!"
+            FROM users WHERE id IN (SELECT UNNEST($1::VARCHAR[]))
+            "#,
+            &ids
+        )
+        .fetch_all(conn)
+        .await
+        .map_err(|err| anyhow!("Failed to get users: {}", err))?;
+        Ok(HashMap::from_iter(
+            data.iter().map(|user| (user.id.clone(), user.clone())),
+        ))
+    }
+    pub async fn select_one<'c, C>(id: String, conn: C) -> Result<Option<Self>>
+    where
+        C: Executor<'c, Database = Postgres>,
+    {
+        query_as!(
+            UserData,
+            r#"
+            SELECT users.id AS "id!", users.realm::text AS "realm!", users.name AS "name!", users.email AS "email!"
+            FROM users where id = $1;
+            "#, id
+        ).fetch_optional(conn).await
+        .map_err(|err| anyhow!("Failed to Get Events: {}", err))
+    }
+}

src/worker.rs

@@ -1,7 +1,4 @@
-use std::{
-    collections::{HashMap, HashSet},
-    env,
-};
+use std::{collections::HashSet, env};
 
 use log::error;
 use redis::{aio::MultiplexedConnection, RedisResult};
@@ -10,7 +7,8 @@ use sqlx::{postgres::PgPoolOptions, query, query_as, Pool, Postgres};
 use std::time::Duration;
 
 use crate::{
-    app::{RedisJob, SimpleRiderChange, UserData},
+    app::{RedisJob, SimpleRiderChange},
+    db::user::UserData,
     pings::PingClient,
 };
 
@@ -65,32 +63,14 @@ async fn get_driver(car_id: i32, db_pool: &Pool<Postgres>) -> UserData {
     ).fetch_one(db_pool).await.unwrap()
 }
 
-async fn get_users_by_id(ids: Vec<String>, db_pool: &Pool<Postgres>) -> HashMap<String, UserData> {
-    let data = query_as!(
-        UserData,
-        r#"
-        SELECT id AS "id!", realm::text AS "realm!", name AS "name!", email AS "email!"
-        FROM users WHERE id IN (SELECT UNNEST($1::VARCHAR[]))
-        "#,
-        &ids
-    )
-    .fetch_all(db_pool)
-    .await
-    .unwrap();
-    HashMap::from_iter(data.iter().map(|user| (user.id.clone(), user.clone())))
-}
-
 async fn get_simple_data(
     data: SimpleRiderChange,
     db_pool: &Pool<Postgres>,
 ) -> (String, UserData, UserData) {
-    let rider = query_as!(
-        UserData,
-        r#"
-        SELECT users.id AS "id!", users.realm::text AS "realm!", users.name AS "name!", users.email AS "email!"
-        FROM users where id = $1;
-        "#, data.rider_id
-    ).fetch_one(db_pool).await.unwrap();
+    let rider = UserData::select_one(data.rider_id, db_pool)
+        .await
+        .unwrap()
+        .unwrap();
     (
         get_event_name(data.event_id, db_pool).await,
         get_driver(data.car_id, db_pool).await,
@@ -137,15 +117,16 @@ async fn work(job: &Item, db_pool: &Pool<Postgres>, pings: &PingClient) -> Resul
             let driver = get_driver(data.car_id, db_pool).await;
             let old_set: HashSet<String> = HashSet::from_iter(data.old_riders);
             let new_set: HashSet<String> = HashSet::from_iter(data.new_riders);
-            let user_map = get_users_by_id(
+            let user_map = UserData::select_map(
                 old_set
                     .difference(&new_set)
                     .chain(new_set.difference(&old_set))
                     .map(|s| s.to_string())
                     .collect(),
                 db_pool,
             )
-            .await;
+            .await
+            .unwrap();
             for removed in old_set.difference(&new_set) {
                 let user = user_map.get(removed).unwrap();
                 if user.realm != "csh" {