What did I do

Author: jabbate19

Cargo.lock

@@ -9,16 +9,20 @@ actix-web = { version = "4.9.0", features = ["cookies"] }
 anyhow = "1.0.88"
 base64 = "0.22.1"
 chrono = { version = "0.4.38", features = ["serde"] }
+clap = { version = "4.5.20", features = ["derive"] }
 dotenv = "0.15.0"
 env_logger = "0.11.5"
 futures-util = "0.3.30"
 include_dir = "0.7.4"
 log = "0.4.22"
 mime_guess = "2.0.5"
 oauth2 = "4.4.2"
+redis = { version = "0.26.1", features = ["aio", "tokio-comp"] }
+redis-work-queue = "0.3.0"
 reqwest = { version = "0.12.7", features = ["json"] }
 serde = { version = "1.0.210", features = ["derive"] }
 serde_json = "1.0.128"
 sqlx = { version = "0.8.2", features = ["chrono", "postgres", "runtime-tokio"] }
+tokio = { version = "1.40.0", features = ["full"] }
 utoipa = { version = "5.0.0-beta.0", features = ["actix_extras", "chrono"] }
 utoipa-swagger-ui = { version = "7.1.1-beta.0", features = ["actix-web"] }

Cargo.toml

@@ -51,4 +51,4 @@ WORKDIR /app
 
 COPY --from=builder /app/target/release/rideboard-v2 .
 
-CMD ["./rideboard-v2"]
+CMD ["./rideboard-v2", "server"]

Dockerfile

@@ -1,6 +1,6 @@
 use crate::api::v1::auth::models::UserRealm;
 use crate::api::v1::auth::models::{CSHUserInfo, UserInfo};
-use crate::AppState;
+use crate::app::AppState;
 use actix_session::Session;
 use actix_web::http::header;
 use actix_web::{get, Scope};

src/api/v1/auth/csh.rs

@@ -1,7 +1,7 @@
 use crate::api::v1::auth::common;
 use crate::api::v1::auth::models::UserRealm;
 use crate::api::v1::auth::models::{GoogleUserInfo, UserInfo};
-use crate::AppState;
+use crate::app::AppState;
 use actix_session::Session;
 use actix_web::http::header;
 use actix_web::{get, web, Scope};

src/api/v1/auth/google.rs

@@ -1,5 +1,5 @@
 use crate::api::v1::auth::models::UserInfo;
-use crate::AppState;
+use crate::app::AppState;
 use crate::{api::v1::auth::models::UserData, auth::SessionAuth};
 use actix_session::Session;
 use actix_web::{
@@ -315,9 +315,13 @@ async fn update_car(
     match updated {
         Ok(Some(_)) => {}
         Ok(None) => {
-            return HttpResponse::NotFound().body("Car not found or you are not the driver.")
+            tx.rollback().await.unwrap();
+            return HttpResponse::NotFound().body("Car not found or you are not the driver.");
+        }
+        Err(_) => {
+            tx.rollback().await.unwrap();
+            return HttpResponse::InternalServerError().body("Failed to update car");
         }
-        Err(_) => return HttpResponse::InternalServerError().body("Failed to update car"),
     }
 
     // Used for sending pings

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

@@ -1,13 +1,16 @@
-use crate::api::v1::event::UserInfo;
+use std::borrow::BorrowMut;
+
+use crate::app::{AppState, SimpleRiderChange};
 use crate::auth::SessionAuth;
-use crate::AppState;
+use crate::{api::v1::event::UserInfo, app::RedisJob};
 use actix_session::Session;
 use actix_web::{
     delete, post,
     web::{self},
     HttpResponse, Responder, Scope,
 };
 use log::error;
+use redis_work_queue::{Item, WorkQueue};
 use sqlx::query;
 use utoipa::OpenApi;
 
@@ -67,7 +70,21 @@ async fn create_rider(
     .await;
 
     match result {
-        Ok(_) => HttpResponse::Ok().body("Joined Car"),
+        Ok(_) => {
+            let work_queue = WorkQueue::new(data.work_queue_key.clone());
+            let item = Item::from_json_data(&RedisJob::Join(SimpleRiderChange {
+                event_id,
+                car_id,
+                rider_id: user_id,
+            }))
+            .unwrap();
+            let mut redis = data.redis.lock().unwrap().clone();
+            work_queue
+                .add_item(&mut redis, &item)
+                .await
+                .expect("failed to add item to work queue");
+            HttpResponse::Ok().body("Joined Car")
+        }
         Err(e) => {
             error!("Failed to Add Rider: {}", e);
             HttpResponse::InternalServerError().body("Failed to create car")
@@ -90,18 +107,33 @@ async fn delete_rider(
     session: Session,
     path: web::Path<(i32, i32)>,
 ) -> impl Responder {
-    let (_event_id, car_id) = path.into_inner();
+    let (event_id, car_id) = path.into_inner();
+    let user_id = session.get::<UserInfo>("userinfo").unwrap().unwrap().id;
 
     let deleted = sqlx::query!(
         "DELETE FROM rider WHERE car_id = $1 AND rider = $2",
         car_id,
-        session.get::<UserInfo>("userinfo").unwrap().unwrap().id
+        user_id
     )
     .execute(&data.db)
     .await;
 
     match deleted {
-        Ok(_) => HttpResponse::Ok().body("Rider deleted"),
+        Ok(_) => {
+            let work_queue = WorkQueue::new(data.work_queue_key.clone());
+            let item = Item::from_json_data(&RedisJob::Leave(SimpleRiderChange {
+                event_id,
+                car_id,
+                rider_id: user_id,
+            }))
+            .unwrap();
+            let mut redis = data.redis.lock().unwrap().clone();
+            work_queue
+                .add_item(&mut redis, &item)
+                .await
+                .expect("failed to add item to work queue");
+            HttpResponse::Ok().body("Rider deleted")
+        }
         Err(_) => HttpResponse::InternalServerError().body("Failed to delete rider"),
     }
 }

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

@@ -11,8 +11,8 @@ use serde::{Deserialize, Serialize};
 use serde_json::json;
 use sqlx::query_as;
 
+use crate::app::AppState;
 use crate::auth::SessionAuth;
-use crate::AppState;
 
 use utoipa::{OpenApi, ToSchema};
 

src/api/v1/event/mod.rs

@@ -2,8 +2,8 @@ use actix_web::{get, web, HttpResponse, Responder, Scope};
 use serde::Deserialize;
 use sqlx::query_as;
 
+use crate::app::AppState;
 use crate::auth::SessionAuth;
-use crate::AppState;
 
 use utoipa::OpenApi;
 

src/api/v1/user.rs

@@ -0,0 +1,51 @@
+use std::sync::{Arc, Mutex};
+
+use oauth2::basic::BasicClient;
+use redis::aio::MultiplexedConnection;
+use redis_work_queue::KeyPrefix;
+use serde::{Deserialize, Serialize};
+use sqlx::PgPool;
+use utoipa::ToSchema;
+
+#[derive(Clone)]
+pub struct AppState {
+    pub db: PgPool,
+    pub redis: Arc<Mutex<MultiplexedConnection>>,
+    pub work_queue_key: KeyPrefix,
+    pub google_oauth: BasicClient,
+    pub google_userinfo_url: String,
+    pub csh_oauth: BasicClient,
+    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,
+    pub car_id: i32,
+    pub rider_id: String,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct MultipleRiderChange {
+    pub event_id: i32,
+    pub car_id: i32,
+    pub old_riders: Vec<String>,
+    pub new_riders: Vec<String>,
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(tag = "type")]
+pub enum RedisJob {
+    Join(SimpleRiderChange),
+    Leave(SimpleRiderChange),
+    RiderUpdate(MultipleRiderChange),
+}

src/app.rs

@@ -1,107 +1,34 @@
-use actix_session::storage::CookieSessionStore;
-use actix_session::SessionMiddleware;
-use actix_web::cookie::Key;
-use actix_web::{middleware::Logger, web, App, HttpResponse, HttpServer, Responder};
-use anyhow::anyhow;
-use base64::prelude::*;
-use include_dir::{include_dir, Dir};
-use log::info;
-use oauth2::basic::BasicClient;
-use sqlx::{postgres::PgPoolOptions, PgPool};
-use std::env;
-
-mod api;
+pub mod api;
+pub mod app;
 mod auth;
-//mod pings; // Undo this when developing it
-
-#[derive(Clone)]
-struct AppState {
-    db: PgPool,
-    google_oauth: BasicClient,
-    google_userinfo_url: String,
-    csh_oauth: BasicClient,
-    csh_userinfo_url: String,
-}
-
-// Embed the 'static' directory into the binary
-static STATIC_DIR: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/src/frontend/dist");
-
-async fn serve_file(path: web::Path<String>) -> impl Responder {
-    let file_path = path.into_inner();
-    if let Some(file) = STATIC_DIR.get_file(&file_path) {
-        let content = file.contents();
-        let mime = mime_guess::from_path(&file_path).first_or_octet_stream();
-        HttpResponse::Ok().content_type(mime.as_ref()).body(content)
-    } else {
-        HttpResponse::NotFound().body("File not found")
-    }
+pub mod pings;
+mod server;
+mod worker;
+
+use clap::{Parser, Subcommand};
+
+#[derive(Parser)]
+#[command(name = "App")]
+#[command(about = "An application with async server and worker subcommands", long_about = None)]
+struct Cli {
+    #[command(subcommand)]
+    command: Commands,
 }
 
-async fn serve_index() -> impl Responder {
-    if let Some(file) = STATIC_DIR.get_file("index.html") {
-        let content = file.contents();
-        let mime = mime_guess::from_path("index.html").first_or_octet_stream();
-        HttpResponse::Ok().content_type(mime.as_ref()).body(content)
-    } else {
-        HttpResponse::NotFound().body("File not found")
-    }
+#[derive(Subcommand)]
+enum Commands {
+    /// Start the async server
+    Server,
+    /// Start the async worker
+    Worker,
 }
 
-#[actix_web::main]
+#[tokio::main]
 async fn main() -> std::io::Result<()> {
-    env_logger::init();
-    dotenv::dotenv().ok();
-
-    let host = env::var("HOST").unwrap_or("127.0.0.1".to_string());
-    let host_inner = host.clone();
-    let port: i32 = match &env::var("PORT").map(|port| port.parse()) {
-        Ok(Ok(p)) => *p,
-        Ok(Err(_)) => 8080,
-        Err(_) => 8080,
-    };
-
-    let db_pool = PgPoolOptions::new()
-        .max_connections(5)
-        .connect(&env::var("DATABASE_URL").expect("DATABASE_URL must be set"))
-        .await
-        .expect("Failed to create pool");
+    let cli = Cli::parse();
 
-    let session_key = env::var("SESSION_KEY")
-        .map_err(|e| anyhow!("Failed to get Env Var: {}", e))
-        .and_then(|key64| {
-            BASE64_STANDARD
-                .decode(key64)
-                .map_err(|e| anyhow!("Failed to decode session key: {}", e))
-        })
-        .map(|key| Key::from(&key))
-        .unwrap_or(Key::generate());
-
-    info!("Starting server at http://{host}:{port}");
-    HttpServer::new(move || {
-        let (google_client, csh_client) = auth::get_clients(&host_inner, port);
-
-        App::new()
-            .app_data(web::Data::new(AppState {
-                db: db_pool.clone(),
-                google_oauth: google_client,
-                google_userinfo_url: "https://openidconnect.googleapis.com/v1/userinfo".to_string(),
-                csh_oauth: csh_client,
-                csh_userinfo_url: env::var("CSH_USERINFO_URL")
-                    .expect("Missing Userinfo URL for CSH Auth"),
-            }))
-            .wrap(
-                SessionMiddleware::builder(CookieSessionStore::default(), session_key.clone())
-                    .cookie_secure(env::var("DEVELOPMENT").is_err())
-                    .build(),
-            )
-            .wrap(Logger::default())
-            .service(api::scope())
-            .route("/", web::get().to(serve_index))
-            .route("/history", web::get().to(serve_index))
-            .route("/login", web::get().to(serve_index))
-            .route("/{filename:.*}", web::get().to(serve_file))
-    })
-    .bind(format!("{host}:{port}"))?
-    .run()
-    .await
+    match &cli.command {
+        Commands::Server => server::main().await,
+        Commands::Worker => worker::main().await,
+    }
 }