Add update subscription endpoint

server/src/api/mod.rs

@@ -4,25 +4,48 @@ mod groups;
 mod import;
 mod ticks;
 mod tracks;
+pub(crate) mod update;
 
-use std::default::default;
-use std::net::{IpAddr, Ipv4Addr};
+use std::{
+    default::default,
+    net::{IpAddr, Ipv4Addr},
+    sync::Arc,
+};
 
 use crate::error::Error;
-use crate::rocket::{Build, Rocket};
-use rocket::fs::{FileServer, NamedFile};
-use rocket::{routes, Config};
+use rocket::{
+    fs::{FileServer, NamedFile},
+    response::stream::{Event, EventStream},
+    routes, Build, Config, Rocket, State,
+};
 use sea_orm::DatabaseConnection;
 
 pub(crate) use error::ErrorResponder;
+use tokio::sync::{
+    broadcast::{self, error::RecvError, Receiver},
+    RwLock,
+};
 
-use self::error::ApiResult;
+use self::{error::ApiResult, update::Update};
 
 #[get("/status")]
 fn status() -> &'static str {
     "Ok"
 }
 
+#[get("/updates")]
+async fn stream_updates(rx: &State<Arc<RwLock<Receiver<Update>>>>) -> EventStream![Event + '_] {
+    let rx: Arc<RwLock<Receiver<Update>>> = (rx as &Arc<RwLock<Receiver<Update>>>).clone();
+    EventStream![loop {
+        let mut rx = rx.write().await;
+        match rx.recv().await {
+            Ok(update) => yield update.to_event(),
+            Err(RecvError::Closed) => break,
+            Err(RecvError::Lagged(count)) => yield Update::lagged(count).to_event(),
+        }
+    }]
+}
+
 #[catch(404)]
 async fn spa_index_redirect() -> ApiResult<NamedFile> {
     Ok(NamedFile::open("/src/public/index.html")
@@ -34,6 +57,7 @@ pub(crate) fn start_server(db: DatabaseConnection) -> Rocket<Build> {
     use groups::*;
     use ticks::*;
     use tracks::*;
+    let (tx, rx) = broadcast::channel::<Update>(8);
     let it = rocket::build()
         .configure(Config {
             address: IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
@@ -41,7 +65,9 @@ pub(crate) fn start_server(db: DatabaseConnection) -> Rocket<Build> {
         })
         .register("/", catchers![spa_index_redirect])
         .manage(db)
-        .mount("/api/v1", routes![status])
+        .manage(tx)
+        .manage(rx)
+        .mount("/api/v1", routes![status, stream_updates])
         .mount(
             "/api/v1/tracks",
             routes![
@@ -50,7 +76,8 @@ pub(crate) fn start_server(db: DatabaseConnection) -> Rocket<Build> {
                 ticks_for_track,
                 insert_track,
                 update_track,
-                delete_track
+                delete_track,
+                ticked,
             ],
         )
         .mount(

server/src/api/ticks.rs

@@ -1,13 +1,14 @@
 use either::{Either, Left, Right};
 use rocket::{http::Status, serde::json::Json, State};
 use sea_orm::{prelude::*, DatabaseConnection};
+use tokio::sync::broadcast::Sender;
 
 use crate::{
     entities::{prelude::*, *},
     error::Error,
 };
 
-use super::error::ApiResult;
+use super::{error::ApiResult, update::Update};
 
 #[get("/")]
 pub(super) async fn all_ticks(
@@ -59,10 +60,18 @@ pub(super) async fn update_tick(
 }
 
 #[delete("/<id>")]
-pub(super) async fn delete_tick(db: &State<DatabaseConnection>, id: i32) -> ApiResult<Status> {
-    Ticks::delete_by_id(id)
-        .exec(db as &DatabaseConnection)
-        .await
-        .map_err(Error::from)?;
-    Ok(Status::Ok)
+pub(super) async fn delete_tick(
+    db: &State<DatabaseConnection>,
+    tx: &State<Sender<Update>>,
+    id: i32,
+) -> ApiResult<Status> {
+    let db = db as &DatabaseConnection;
+    let tick = Ticks::find_by_id(id).one(db).await.map_err(Error::from)?;
+    if let Some(tick) = tick {
+        tick.clone().delete(db).await.map_err(Error::from)?;
+        tx.send(Update::tick_cancelled(tick)).map_err(Error::from)?;
+        Ok(Status::Ok)
+    } else {
+        Ok(Status::NotFound)
+    }
 }

server/src/api/tracks.rs

@@ -6,6 +6,9 @@ use rocket::http::Status;
 use rocket::{serde::json::Json, State};
 use sea_orm::{prelude::*, DatabaseConnection};
 use std::default::default;
+use tokio::sync::broadcast::Sender;
+
+use super::update::Update;
 
 #[get("/")]
 pub(super) async fn all_tracks(
@@ -90,13 +93,16 @@ pub(super) async fn delete_track(db: &State<DatabaseConnection>, id: i32) -> Api
 #[patch("/<id>/ticked")]
 pub(super) async fn ticked(
     db: &State<DatabaseConnection>,
+    tx: &State<Sender<Update>>,
     id: i32,
 ) -> ApiResult<Json<ticks::Model>> {
     let tick = ticks::ActiveModel::now(id);
-    Ok(Json(
-        tick.insert(db as &DatabaseConnection)
-            .await
-            .map_err(Error::from)?
-            .to_owned(),
-    ))
+    let tick = tick
+        .insert(db as &DatabaseConnection)
+        .await
+        .map_err(Error::from)?
+        .to_owned();
+    tx.send(Update::tick_added(tick.clone()))
+        .map_err(Error::from)?;
+    Ok(Json(tick))
 }

server/src/api/update.rs

@@ -0,0 +1,56 @@
+use rocket::response::stream::Event;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+
+use crate::entities::ticks;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) enum Update {
+    TickChanged {
+        kind: UpdateType,
+        tick: ticks::Model,
+    },
+    Lagged {
+        kind: UpdateType,
+        count: u64,
+    },
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub(crate) enum UpdateType {
+    TickAdded,
+    TickDropped,
+    Error,
+}
+
+impl Update {
+    pub(crate) fn lagged(count: u64) -> Update {
+        Update::Lagged {
+            kind: UpdateType::Error,
+            count,
+        }
+    }
+    pub(crate) fn tick_added(tick: ticks::Model) -> Self {
+        Self::TickChanged {
+            kind: UpdateType::TickAdded,
+            tick,
+        }
+    }
+    pub(crate) fn tick_cancelled(tick: ticks::Model) -> Self {
+        Self::TickChanged {
+            kind: UpdateType::TickDropped,
+            tick,
+        }
+    }
+    pub(crate) fn to_event(&self) -> Event {
+        use Update::*;
+        match self {
+            TickChanged { kind, tick } => Event::json(tick).event(format!("{kind:?}")),
+            Lagged { kind, count } => {
+                Event::json(&json! {{"message": "error: lagged", "count": count}})
+                    .event(format!("{kind:?}"))
+            }
+        }
+    }
+}

server/src/error.rs

@@ -3,7 +3,7 @@ use std::string;
 use derive_builder::UninitializedFieldError;
 
 #[derive(Debug, thiserror::Error)]
-pub enum Error {
+pub(crate) enum Error {
     #[error(transparent)]
     Builder(#[from] UninitializedFieldError),
     #[error(transparent)]
@@ -16,6 +16,8 @@ pub enum Error {
     Unreachable,
     #[error(transparent)]
     Utf8(#[from] string::FromUtf8Error),
+    #[error(transparent)]
+    ChannelSendError(#[from] tokio::sync::broadcast::error::SendError<crate::api::update::Update>),
 }
 
-pub type Result<T> = std::result::Result<T, Error>;
+pub(crate) type Result<T> = std::result::Result<T, Error>;