clubmanager/src/api.rs

use axum::{
    extract::{Path, State},
    response::{
        sse::{Event, KeepAlive},
        IntoResponse, Sse,
    },
    routing::{get, post},
    Form,
};
use cm_lib::{
    hx_request::HxRequest,
    models::{NewDrink, Shift},
    schema::shifts,
};
use diesel::{ExpressionMethods, OptionalExtension, QueryDsl, SelectableHelper};
use diesel_async::{scoped_futures::ScopedFutureExt, AsyncConnection, RunQueryDsl};
use futures_util::Stream;
use serde::Deserialize;
use tokio_stream::{wrappers::errors::BroadcastStreamRecvError, StreamExt as _};

use crate::AppState;
use cm_lib::render;

pub(crate) fn router() -> axum::Router<AppState> {
    axum::Router::new()
        .route("/drinks", post(add_drink))
        .route("/shifts/open", post(open_shift))
        .route("/shifts/:id/close", post(close_shift))
        .route("/ada/updates", get(ada_subscribe))
}

async fn open_shift(State(state): State<AppState>, HxRequest(hx): HxRequest) -> impl IntoResponse {
    let shift = {
        let mut conn = state.connection.get().await.unwrap();
        conn.transaction(|conn| {
            use cm_lib::schema::shifts::dsl::*;

            async move {
                diesel::insert_into(shifts)
                    .default_values()
                    .execute(conn)
                    .await?;

                shifts
                    .order(id.desc())
                    .select(Shift::as_select())
                    .first(conn)
                    .await
            }
            .scope_boxed()
        })
        .await
        .optional()
        .unwrap()
    };

    render!(crate::templates::home_html, shift, !hx)
}

async fn close_shift(
    State(state): State<AppState>,
    Path(id): Path<u32>,
    HxRequest(hx): HxRequest,
) -> impl IntoResponse {
    let mut conn = state.connection.get().await.unwrap();
    diesel::update(shifts::table.filter(shifts::id.eq(id)))
        .set(shifts::end.eq(Some(chrono::Utc::now().naive_local())))
        .execute(&mut conn)
        .await
        .unwrap();

    render!(crate::templates::home_html, None, !hx)
}

#[derive(Deserialize, Debug)]
struct DrinkForm {
    shift_id: u32,
    price: u32,
    quantity: u32,
}

impl From<DrinkForm> for NewDrink {
    fn from(value: DrinkForm) -> Self {
        NewDrink {
            price: value.price,
            quantity: value.quantity,
            shift: value.shift_id,
        }
    }
}

async fn add_drink(
    State(state): State<AppState>,
    HxRequest(hx): HxRequest,
    Form(form): Form<DrinkForm>,
) -> impl IntoResponse {
    let mut conn = state.connection.get().await.unwrap();

    let open_shift: Option<Shift> = {
        use cm_lib::schema::shifts::dsl::*;

        shifts
            .filter(end.is_null())
            .select(Shift::as_select())
            .first(&mut conn)
            .await
            .optional()
            .expect("Query failed: No open shifts found")
    };
    let open_shift = open_shift.unwrap();

    async {
        use cm_lib::schema::drinks::dsl::*;

        diesel::insert_into(drinks)
            .values::<NewDrink>(form.into())
            .execute(&mut conn)
            .await
            .unwrap()
    }
    .await;

    let mut headers = axum::http::HeaderMap::new();
    headers.insert("HX-Push-Url", "/".parse().unwrap());
    (
        headers,
        render!(crate::templates::home_html, Some(open_shift), !hx),
    )
}

async fn ada_subscribe(
    State(state): State<AppState>,
) -> Sse<impl Stream<Item = Result<Event, BroadcastStreamRecvError>>> {
    let stream =
        tokio_stream::wrappers::BroadcastStream::new(state.sse_handler.ada_sender.subscribe())
            .map(|r| r.map(|s| Event::default().event("ada").data(s)));

    Sse::new(stream).keep_alive(KeepAlive::default())
}