WIP: Boosted ride stat events over websocket

Author: dustinrouillard

src/connectivity/rabbit.rs

@@ -5,9 +5,12 @@ use lapin::{
   BasicProperties, Channel, Connection, ConnectionProperties, Queue,
 };
 use serde::{Deserialize, Serialize};
-use serde_json::json;
+use serde_json::{json, Value};
 
-use crate::{config::Config, structs::spotify::CurrentPlaying};
+use crate::{
+  config::Config,
+  structs::{boosted::BoostedRideUpdate, spotify::CurrentPlaying},
+};
 
 #[derive(Clone)]
 pub struct RabbitManager {
@@ -16,9 +19,9 @@ pub struct RabbitManager {
 }
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
-struct SpotifySocketData {
+struct RabbitEventsData<Data = Value> {
   t: u8,
-  d: CurrentPlaying,
+  d: Data,
 }
 
 impl RabbitManager {
@@ -50,8 +53,28 @@ impl RabbitManager {
     Self { channel, queue }
   }
 
+  pub async fn publish_ride_state(&mut self, data: &BoostedRideUpdate) {
+    let message = RabbitEventsData {
+      t: 0,
+      d: data.to_owned(),
+    };
+    let json = serde_json::to_string(&message).unwrap();
+
+    self
+      .channel
+      .basic_publish(
+        "",
+        "dstn-gateway-ingest",
+        BasicPublishOptions::default(),
+        json.as_bytes(),
+        BasicProperties::default(),
+      )
+      .await
+      .unwrap();
+  }
+
   pub async fn publish_spotify_current(&mut self, data: &CurrentPlaying) {
-    let message = SpotifySocketData {
+    let message = RabbitEventsData {
       t: 0,
       d: data.to_owned(),
     };

src/helpers/boosted.rs

@@ -0,0 +1,41 @@
+use envconfig::Envconfig;
+use redis::aio::ConnectionManager;
+
+use crate::{
+  config::Config,
+  connectivity::{rabbit::RabbitManager, valkey::ValkeyManager},
+  services::boosted::structs::BoostedStats,
+  structs::boosted::BoostedRideUpdate,
+};
+
+pub async fn send_boosted_event(
+  valkey: &mut ValkeyManager,
+  rabbit: &mut RabbitManager,
+) {
+  let config = Config::init_from_env().unwrap();
+
+  let in_ride = redis::cmd("GET")
+    .arg("boosted/in-ride")
+    .query_async::<ConnectionManager, String>(&mut valkey.cm)
+    .await
+    .unwrap_or(String::from("false"))
+    == "true";
+
+  let client = reqwest::Client::new();
+  let res = client
+    .get(format!("{}/v1/users/stats", config.boosted_api_endpoint))
+    .header("Authorization", config.boosted_api_token)
+    .send()
+    .await
+    .unwrap();
+
+  let json = res.json::<BoostedStats>().await.unwrap();
+
+  let state = BoostedRideUpdate {
+    riding: in_ride,
+    latest_ride: json.latest_ride,
+    stats: json.stats,
+  };
+
+  rabbit.publish_ride_state(&state).await;
+}

src/helpers/mod.rs

@@ -0,0 +1 @@
+pub mod boosted;

src/main.rs

@@ -1,5 +1,6 @@
 pub mod config;
 pub mod connectivity;
+pub mod helpers;
 pub mod modules;
 pub mod services;
 pub mod structs;

src/services/boosted/structs.rs

@@ -1,33 +1,33 @@
 use serde::{Deserialize, Serialize};
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Debug)]
 pub struct BoostedStats {
   pub latest_ride: RideStats,
   pub stats: Stats,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Debug)]
 pub struct RideStats {
   pub started_at: String,
   pub ended_at: String,
   pub duration: f64,
   pub distance: f64,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Debug)]
 pub struct Stats {
   pub boards: Boards,
   pub rides: ValueEntry,
   pub duration: ValueEntry,
   pub distance: ValueEntry,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Debug)]
 pub struct Boards {
   pub distance: f64,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Debug)]
 pub struct ValueEntry {
   pub day: f64,
   pub week: f64,

src/services/hooks/boosted.rs

@@ -4,6 +4,7 @@ use redis::aio::ConnectionManager;
 
 use crate::{
   config::Config,
+  helpers::boosted::send_boosted_event,
   services::hooks::structs::{BoostedHookPayload, BoostedHookType},
   ServerState,
 };
@@ -28,6 +29,7 @@ async fn execute(
   }
 
   let valkey = &mut state.valkey.clone();
+  let rabbit = &mut state.rabbit.clone();
 
   match payload.hook_type {
     BoostedHookType::RideStarted => {
@@ -36,18 +38,24 @@ async fn execute(
         .arg("true")
         .query_async::<ConnectionManager, String>(&mut valkey.cm)
         .await;
+
+      send_boosted_event(valkey, rabbit).await;
     }
     BoostedHookType::RideEnded => {
       let _ = redis::cmd("DEL")
         .arg("boosted/in-ride")
         .query_async::<ConnectionManager, String>(&mut valkey.cm)
         .await;
+
+      send_boosted_event(valkey, rabbit).await;
     }
     BoostedHookType::RideDiscarded => {
       let _ = redis::cmd("DEL")
         .arg("boosted/in-ride")
         .query_async::<ConnectionManager, String>(&mut valkey.cm)
         .await;
+
+      send_boosted_event(valkey, rabbit).await;
     }
   }
 

src/structs/boosted.rs

@@ -0,0 +1,12 @@
+use serde::{Deserialize, Serialize};
+use serde_with::skip_serializing_none;
+
+use crate::services::boosted::structs::{RideStats, Stats};
+
+#[skip_serializing_none]
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BoostedRideUpdate {
+  pub riding: bool,
+  pub latest_ride: RideStats,
+  pub stats: Stats,
+}

src/structs/mod.rs

@@ -1,4 +1,5 @@
 pub mod blog;
+pub mod boosted;
 pub mod github;
 pub mod spotify;
 pub mod uploads;