Cargo.toml

@@ -18,7 +18,7 @@ tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 tower-http = { version = "0.4.4", features = ["full"] }
 ructe.workspace = true
-tokio-stream = { version = "0.1.14", features = ["sync", "time"] }
+tokio-stream = { version = "0.1.14", features = ["sync"] }
 futures-util = "0.3.28"
 diesel = { version = "2.1.0", features = ["chrono"] }
 diesel-async = { version = "0.4.1", features = ["mysql", "deadpool"] }

src/api.rs

@@ -1,11 +1,8 @@
-use std::time::Duration;
-
 use axum::{
     extract::{Path, State},
-    http::StatusCode,
     response::{
         sse::{Event, KeepAlive},
-        IntoResponse, Response, Sse,
+        IntoResponse, Sse,
     },
     routing::{get, post},
     Form,
@@ -20,10 +17,7 @@ use diesel::{ExpressionMethods, OptionalExtension, QueryDsl, SelectableHelper};
 use diesel_async::{scoped_futures::ScopedFutureExt, AsyncConnection, RunQueryDsl};
 use futures_util::Stream;
 use serde::Deserialize;
-use tokio_stream::{
+use tokio_stream::{wrappers::errors::BroadcastStreamRecvError, StreamExt as _};
-    wrappers::{errors::BroadcastStreamRecvError, BroadcastStream},
-    StreamExt as _,
-};
 
 use crate::AppState;
 
@@ -100,7 +94,7 @@ async fn add_drink(
     State(state): State<AppState>,
     HxRequest(hx): HxRequest,
     Form(form): Form<DrinkForm>,
-) -> Response {
+) -> impl IntoResponse {
     let mut conn = state.connection.get().await.unwrap();
 
     let open_shift: Option<Shift> = {
@@ -114,18 +108,9 @@ async fn add_drink(
             .optional()
             .expect("Query failed: No open shifts found")
     };
+    let open_shift = open_shift.unwrap();
 
-    if open_shift.is_none() {
+    async {
-        return (
-            StatusCode::CONFLICT,
-            "Unable to add a drink when no shift is open",
-        )
-            .into_response();
-    }
-
-    let open_shift = unsafe { open_shift.unwrap_unchecked() };
-
-    {
         use cm_lib::schema::drinks::dsl::*;
 
         diesel::insert_into(drinks)
@@ -133,7 +118,8 @@ async fn add_drink(
             .execute(&mut conn)
             .await
             .unwrap()
-    };
+    }
+    .await;
 
     let mut headers = axum::http::HeaderMap::new();
     headers.insert("HX-Push-Url", "/".parse().unwrap());
@@ -141,7 +127,6 @@ async fn add_drink(
         headers,
         render!(crate::templates::home_html, Some(open_shift), !hx),
     )
-        .into_response()
 }
 
 #[derive(Deserialize, Debug)]
@@ -193,9 +178,9 @@ async fn add_dancer(
 async fn ada_subscribe(
     State(state): State<AppState>,
 ) -> Sse<impl Stream<Item = Result<Event, BroadcastStreamRecvError>>> {
-    let stream = BroadcastStream::new(state.sse_handler.ada_sender.subscribe())
+    let stream =
-        .map(|r| r.map(|s| Event::default().event("ada").data(s)))
+        tokio_stream::wrappers::BroadcastStream::new(state.sse_handler.ada_sender.subscribe())
-        .throttle(Duration::from_secs(1));
+            .map(|r| r.map(|s| Event::default().event("ada").data(s)));
 
     Sse::new(stream).keep_alive(KeepAlive::default())
 }

src/main.rs

@@ -20,7 +20,9 @@ use axum::{
     Router,
 };
 
-use diesel::prelude::*;
+use diesel::{
+    result::OptionalExtension, BelongingToDsl, ExpressionMethods, QueryDsl, SelectableHelper,
+};
 
 use diesel_async::{
     pooled_connection::{deadpool::Pool, AsyncDieselConnectionManager},
@@ -113,10 +115,11 @@ async fn root(State(state): State<AppState>, HxRequest(hx): HxRequest) -> impl I
 
         shifts
             .filter(end.is_null())
+            .select(Shift::as_select())
             .first(&mut conn)
             .await
             .optional()
-            .expect("Query failed")
+            .expect("Query failed: No open shifts found")
     };
 
     tracing::debug!("{open_shift:?}");
@@ -129,10 +132,6 @@ async fn drinks(Path(id): Path<u32>) -> impl IntoResponse {
 }
 
 async fn shift_report(State(state): State<AppState>, Path(id): Path<u32>) -> impl IntoResponse {
-    // TODO: Massive rework of how this works
-    // was experimenting with traits, but most of that
-    // didn't lead to what I was aiming for
-
     let mut conn = state.connection.get().await.unwrap();
     let shift: Shift = shifts::table.find(id).first(&mut conn).await.unwrap();
     let drinks: Vec<Drink> = Drink::belonging_to(&shift).load(&mut conn).await.unwrap();
@@ -144,13 +143,13 @@ async fn shift_report(State(state): State<AppState>, Path(id): Path<u32>) -> imp
 }
 
 async fn shift_reports(State(state): State<AppState>) -> impl IntoResponse {
-    let mut shifts: Vec<Shift> = {
+    let mut conn = state.connection.get().await.unwrap();
-        use cm_lib::schema::shifts::dsl::*;
 
-        let mut conn = state.connection.get().await.unwrap();
+    let mut shifts: Vec<Shift> = shifts::table
-
+        .select(Shift::as_select())
-        shifts.load(&mut conn).await.unwrap()
+        .load(&mut conn)
-    };
+        .await
+        .unwrap();
 
     shifts.sort_by(|a, b| b.start.cmp(&a.start));
 

src/sse_handler.rs

@@ -3,17 +3,6 @@ pub enum SseMessage {
     RefreshAda,
 }
 
-// I think for the ada_sender, a watch would be ideal if I want to always use full renders of the
-// watchlist, however if I want more atomic updates, it may be useful to keep the broadcast's
-// buffer and return an html component that requests a complete rerender if the handler lagged
-/// A container struct for keeping access to the channel senders for both channels used in the sse
-/// pipeline
-///
-/// For something wishing to listen in on the broadcast html stream, you should call `.subscribe()`
-/// on ada_sender
-///
-/// Anything wishing to send to sse should pass an [SseMessage] to the [tokio::sync::mpsc::Sender]
-/// in sse_sender
 #[derive(Debug, Clone)]
 pub struct SseHandler {
     pub ada_sender: tokio::sync::broadcast::Sender<String>,
@@ -21,8 +10,6 @@ pub struct SseHandler {
 }
 
 impl SseHandler {
-    /// This function initializes the channels used for sse, as well as spawning a manager task for
-    /// handling sse messages
     pub fn init() -> Self {
         let (ada_sender, _) = tokio::sync::broadcast::channel(10);
         let (sse_sender, mut sse_receiver) = tokio::sync::mpsc::channel(10);
@@ -50,7 +37,6 @@ impl SseHandler {
 }
 
 async fn get_ada_list() -> String {
-    // TODO: need to scope out if these errors need to be handled
     let mut buf = Vec::new();
 
     crate::templates::components::ada_list_html(&mut buf).unwrap();