feat(backend): endpoint for updating single board (#1623)

* feat(backend): endpoint for updating single board

* chore(format): remove redundant field names

* chore(): remove comment

* refactor(update): change input on error

* refactor(update): simplify trait

backend/Cargo.toml

@@ -10,7 +10,7 @@ anyhow = "1.0.86"
 axum = {version = "0.7.4", features = ["ws", "macros"]}
 axum-auth = { version = "0.7.0", default-features= false, features = ["auth-bearer"] }
 futures-util = "0.3.30"
-redis = {version = "0.25.3", features = ["tokio-comp", "aio"]}
+redis = {version = "0.26.1", features = ["tokio-comp", "aio", "json"]}
 serde = {version = "1.0.199", features = ["derive"]}
 serde_json = "1.0.116"
 tokio = { version = "1.36.0", features = ["macros", "rt","rt-multi-thread", "signal"] }

backend/src/main.rs

@@ -9,7 +9,7 @@ use axum::{
 };
 
 use axum_auth::AuthBearer;
-use serde_json::{from_str, Value};
+use serde_json::{to_string, Value};
 use tokio::{net::TcpListener, time};
 
 use redis::{AsyncCommands, ConnectionLike};
@@ -21,7 +21,7 @@ mod types;
 
 mod utils;
 use tower_http::cors::CorsLayer;
-use types::{AppError, AppState, Message};
+use types::{AppError, AppState, BoardAction, Message};
 use utils::{graceful_shutdown, setup_redis};
 
 use crate::types::Guard;
@@ -61,6 +61,7 @@ async fn main() {
         .route("/subscribe/:bid", get(subscribe))
         .route("/refresh/:bid", post(trigger))
         .route("/update", post(update))
+        .route("/update/:bid", post(update_board))
         .route("/alive", get(check_health))
         .route("/reset", post(reset_active))
         .with_state(redis_clients)
@@ -115,7 +116,10 @@ async fn trigger(
     if token != state.key {
         return Ok(StatusCode::UNAUTHORIZED);
     }
-    state.master.publish(bid, payload.to_string()).await?;
+    state
+        .master
+        .publish(bid, BoardAction::Refresh { payload })
+        .await?;
     Ok(StatusCode::OK)
 }
 
@@ -130,6 +134,21 @@ async fn update(
     Ok(StatusCode::OK)
 }
 
+async fn update_board(
+    AuthBearer(token): AuthBearer,
+    State(mut state): State<AppState>,
+    Path(bid): Path<String>,
+) -> Result<StatusCode, AppError> {
+    if token != state.key {
+        return Ok(StatusCode::UNAUTHORIZED);
+    }
+    state
+        .master
+        .publish(bid, to_string(&BoardAction::Update)?)
+        .await?;
+    Ok(StatusCode::OK)
+}
+
 async fn subscribe(
     Path(bid): Path<String>,
     State(state): State<AppState>,
@@ -147,11 +166,14 @@ async fn subscribe(
                 if channel == "update" {
                     Message::Update
                 } else {
-                let payload = msg.get_payload::<String>()?;
-                Message::Refresh {
-                    payload: from_str::<Value>(payload.as_str())?,
-                }}
+                let payload = msg.get_payload::<BoardAction>()?;
+
+                match payload {
+                    BoardAction::Refresh { payload } => Message::Refresh { payload },
+                    BoardAction::Update => Message::Update,
+                }
             }
+        }
         () = time::sleep(Duration::from_secs(55)) => {
             Message::Timeout
         }

backend/src/types.rs

@@ -3,8 +3,11 @@ use axum::{
     http::{Response, StatusCode},
     response::IntoResponse,
 };
-use redis::{aio::MultiplexedConnection, AsyncCommands, Client};
-use serde::Serialize;
+use redis::{
+    aio::MultiplexedConnection, from_redis_value, AsyncCommands, Client, FromRedisValue,
+    ToRedisArgs,
+};
+use serde::{Deserialize, Serialize};
 use serde_json::{to_string, Value};
 
 #[derive(Clone)]
@@ -22,6 +25,31 @@ pub enum Message {
     Timeout,
 }
 
+#[derive(Serialize, Deserialize)]
+pub enum BoardAction {
+    Refresh { payload: Value },
+    Update,
+}
+
+impl FromRedisValue for BoardAction {
+    fn from_redis_value(v: &redis::Value) -> redis::RedisResult<Self> {
+        let s: String = from_redis_value(v)?;
+        Ok(serde_json::from_str::<BoardAction>(&s)?)
+    }
+}
+
+impl ToRedisArgs for BoardAction {
+    fn write_redis_args<W>(&self, out: &mut W)
+    where
+        W: ?Sized + redis::RedisWrite,
+    {
+        match to_string(self) {
+            Ok(s) => out.write_arg_fmt(s),
+            Err(_) => out.write_arg_fmt(0_u8),
+        }
+    }
+}
+
 pub struct Guard {
     pub master: MultiplexedConnection,
 }
@@ -65,7 +93,7 @@ pub struct AppError {
 
 impl IntoResponse for AppError {
     fn into_response(self) -> axum::response::Response {
-        (StatusCode::INTERNAL_SERVER_ERROR).into_response()
+        StatusCode::INTERNAL_SERVER_ERROR.into_response()
     }
 }
 

backend/src/utils.rs

@@ -34,6 +34,7 @@ pub async fn setup_redis() -> (MultiplexedConnection, Client) {
         db: 0,
         username: None,
         password: Some(redis_pw),
+        protocol: redis::ProtocolVersion::RESP3,
     };
 
     let master_host = std::env::var("REDIS_MASTER_SERVICE_HOST")