Added Sending Messages test

Cargo.toml

@@ -15,6 +15,7 @@ members = [
     "crates/telemetry",
     "crates/token_manager",
     "crates/utils",
+    "crates/discord",
 ]
 
 default-members = ["crates/api"]
@@ -26,6 +27,7 @@ default-members = ["crates/api"]
 #
 api = { path = "crates/api" }
 db = { path = "crates/db" }
+discord = { path = "crates/discord" }
 entities = { path = "crates/entities" }
 error = { path = "crates/error" }
 id-generator = { path = "crates/id-generator" }
@@ -39,31 +41,34 @@ utils = { path = "crates/utils" }
 #
 # External Crates
 #
+regex = "1"
 memchr = "2"
 ring = "0.17"
 prost = "0.13"
 tracing = "0.1"
+futures = "0.3"
 ntex-cors = "2"
 dotenvy = "0.15"
 sailfish = "0.9"
 serde_trim = "1"
 base64-simd = "0.8"
 quick_cache = "0.6"
 prost-build = "0.13"
+twilight-model = "0.15"
 postgres-derive = "0.4"
 deadpool-postgres = "0.14"
 tokio = { version = "1", features = ["full"] }
 chrono = { version = "0.4", features = ["serde"] }
 reqwest = { version = "0.12", features = ["json"] }
 openssl = { version = "0.10", features = ["v110"] }
-regex = { version = "1", features = ["unstable"] }
 tracing-log = { version = "0.2", features = ["ahash"] }
 tokio-stream = { version = "0.1", features = ["sync"] }
 serde = { version = "1", features = ["rc", "unstable"] }
 ntex = { version = "2", features = ["tokio", "openssl"] }
 ahash = { version = "0.8", features = ["compile-time-rng"] }
 dashmap = { version = "6", features = ["inline", "raw-api"] }
 refinery = { version = "0.8", features = ["tokio-postgres"] }
+twilight-http = { version = "0.15", features = ["simd-json"] }
 postgres-types = { version = "0.2", features = ["with-chrono-0_4"] }
 parking_lot = { version = "0.12", features = ["arc_lock", "nightly"] }
 garde = { version = "0.20", features = ["derive", "email", "email-idna"] }

crates/api/Cargo.toml

@@ -14,6 +14,7 @@ name = "api"
 path = "src/main.rs"
 
 [dependencies]
+discord.workspace = true
 db.workspace = true
 token_manager.workspace = true
 error.workspace = true

crates/discord/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "discord"
+version = "0.1.0"
+edition = "2021"
+
+[lints]
+workspace = true
+
+
+[lib]
+path = "src/discord.rs"
+doctest = false
+
+[dependencies]
+tokio.workspace = true
+error.workspace = true
+futures.workspace = true
+dashmap.workspace = true
+twilight-http.workspace = true
+twilight-model.workspace = true

crates/discord/src/discord.rs

@@ -0,0 +1,167 @@
+use std::{ops::Deref, sync::Arc, time::Instant};
+
+use dashmap::DashMap;
+use error::AppResult;
+use futures::future::join_all;
+use tokio::task::JoinHandle;
+use twilight_http::Client;
+use twilight_model::{
+    channel::message::{
+        embed::{EmbedAuthor, EmbedField, EmbedFooter},
+        Embed,
+    },
+    id::{
+        marker::{ChannelMarker, UserMarker},
+        Id,
+    },
+    util::Timestamp,
+};
+
+#[derive(Clone)]
+pub struct DiscordClient {
+    inner: Arc<DiscordClientInner>,
+}
+
+pub struct DiscordClientInner {
+    cache: DashMap<Id<UserMarker>, Id<ChannelMarker>>,
+    client: Client,
+}
+
+#[derive(Clone)]
+pub struct RaceData {
+    pub championship_name: Box<str>,
+    pub circuit: String,
+    pub start_time: String,
+}
+
+impl Deref for DiscordClient {
+    type Target = DiscordClientInner;
+
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+impl DiscordClient {
+    pub fn new(token: String) -> Self {
+        let inner = DiscordClientInner {
+            client: Client::new(token),
+            cache: DashMap::new(),
+        };
+
+        Self {
+            inner: Arc::new(inner),
+        }
+    }
+
+    pub async fn send_race_notification(&self, user_id: i64, race_data: RaceData) -> AppResult<()> {
+        let id = Self::to_user_id(user_id);
+        let channel_id = self.get_or_create_dm_channel(id).await?;
+        let embed = Self::embed_message(race_data);
+
+        self.client
+            .create_message(channel_id)
+            .embeds(&[embed])
+            .unwrap()
+            .await?;
+
+        Ok(())
+    }
+
+    //TODO: Capture all errors
+    pub async fn send_race_notifications(
+        &self,
+        user_ids: &[i64],
+        race_data: RaceData,
+    ) -> AppResult<()> {
+        let tasks: Vec<JoinHandle<AppResult<()>>> = user_ids
+            .iter()
+            .map(|&user_id| {
+                let self_clone = self.clone();
+                let race_data_clone = race_data.clone();
+                tokio::spawn(async move {
+                    self_clone
+                        .send_race_notification(user_id, race_data_clone)
+                        .await
+                })
+            })
+            .collect();
+
+        let results = join_all(tasks).await;
+
+        for result in results {
+            result??;
+        }
+
+        Ok(())
+    }
+
+    #[inline]
+    async fn get_or_create_dm_channel(
+        &self,
+        user_id: Id<UserMarker>,
+    ) -> AppResult<Id<ChannelMarker>> {
+        if let Some(entry) = self.cache.get(&user_id) {
+            return Ok(*entry.value());
+        }
+
+        let channel = self
+            .client
+            .create_private_channel(user_id)
+            .await?
+            .model()
+            .await?;
+
+        self.cache.insert(user_id, channel.id);
+
+        Ok(channel.id)
+    }
+
+    #[inline]
+    fn embed_message(race_data: RaceData) -> Embed {
+        Embed {
+            author: Some(EmbedAuthor {
+                name: race_data.championship_name.into_string(),
+                icon_url: None,
+                proxy_icon_url: None,
+                url: None,
+            }),
+            title: Some("Upcoming Race".to_owned()),
+            description: Some(
+                "The race is about to begin. Get ready for the action on the track!".to_owned(),
+            ),
+            color: Some(0x00ff0),
+            fields: vec![
+                EmbedField {
+                    name: "Circuit".to_owned(),
+                    value: race_data.circuit,
+                    inline: true,
+                },
+                EmbedField {
+                    name: "Start Time".to_owned(),
+                    value: race_data.start_time,
+                    inline: true,
+                },
+            ],
+            footer: Some(EmbedFooter {
+                text: "Good luck and may the best driver win!".to_owned(),
+                icon_url: None,
+                proxy_icon_url: None,
+            }),
+            timestamp: Some(
+                Timestamp::from_secs(Instant::now().elapsed().as_secs() as i64).unwrap(),
+            ),
+            video: None,
+            image: None,
+            provider: None,
+            thumbnail: None,
+            url: None,
+            kind: "".to_owned(),
+        }
+    }
+
+    #[inline]
+    fn to_user_id(id: i64) -> Id<UserMarker> {
+        Id::new(id as u64)
+    }
+}

crates/error/Cargo.toml

@@ -15,4 +15,5 @@ ntex.workspace = true
 tracing.workspace = true
 reqwest.workspace = true
 sailfish.workspace = true
+twilight-http.workspace = true
 deadpool-postgres.workspace = true

crates/error/src/error.rs

@@ -37,6 +37,7 @@ pub enum AppError {
     Common(CommonError),
     F1(F1ServiceError),
     Firewall(FirewallError),
+    Twilight,
     PgError,
     PgPool,
     Reqwest,
@@ -78,6 +79,20 @@ impl From<sailfish::RenderError> for AppError {
     }
 }
 
+impl From<twilight_http::Error> for AppError {
+    fn from(value: twilight_http::Error) -> Self {
+        error!("Twilight error: {:?}", value);
+        AppError::Twilight
+    }
+}
+
+impl From<twilight_http::response::DeserializeBodyError> for AppError {
+    fn from(value: twilight_http::response::DeserializeBodyError) -> Self {
+        error!("Twilight Deserialize error: {:?}", value);
+        AppError::Twilight
+    }
+}
+
 impl AppError {
     const fn error_status(&self) -> StatusCode {
         match self {
@@ -92,6 +107,7 @@ impl AppError {
             AppError::PgPool => StatusCode::INTERNAL_SERVER_ERROR,
             AppError::Reqwest => StatusCode::INTERNAL_SERVER_ERROR,
             AppError::Sailfish => StatusCode::INTERNAL_SERVER_ERROR,
+            AppError::Twilight => StatusCode::INTERNAL_SERVER_ERROR,
         }
     }
 
@@ -108,6 +124,7 @@ impl AppError {
             AppError::PgPool => "Pool error",
             AppError::Reqwest => "Reqwest error",
             AppError::Sailfish => "Email Render Error",
+            AppError::Twilight => "Twilight error",
         }
     }
 }

crates/telemetry/src/manager.rs

@@ -5,7 +5,7 @@ use ntex::{
 };
 use parking_lot::{Mutex, RwLock};
 use prost::Message;
-use std::{ops::Deref, ptr::addr_of, sync::Arc, time::Duration};
+use std::{ops::Deref, sync::Arc, time::Duration};
 use tokio::sync::{
     broadcast::{Receiver, Sender},
     oneshot,
@@ -545,13 +545,12 @@ impl PlayerTelemetry {
         let car_damage = self.car_damage.get_or_insert_with(Default::default);
 
         car_damage.tyres_wear.clear();
-        unsafe {
-            let brakes_temp_ptr = addr_of!(data.tyres_wear);
 
-            car_damage
-                .tyres_wear
-                .extend_from_slice(&brakes_temp_ptr.read_unaligned());
-        }
+        let brakes_temp_ptr = &raw const data.tyres_wear;
+
+        car_damage
+            .tyres_wear
+            .extend_from_slice(unsafe { &brakes_temp_ptr.read_unaligned() });
 
         car_damage.tyres_damage.clear();
         car_damage
@@ -623,14 +622,12 @@ impl PlayerTelemetry {
         telemetry.engine_temperature = Some(data.engine_temperature as u32);
 
         telemetry.brakes_temperature.clear();
-        unsafe {
-            let brakes_temp_ptr = addr_of!(data.brakes_temperature);
-            let brakes_temp = brakes_temp_ptr.read_unaligned();
+        let brakes_temp_ptr = &raw const data.brakes_temperature;
+        let brakes_temp = unsafe { brakes_temp_ptr.read_unaligned() };
 
-            telemetry
-                .brakes_temperature
-                .extend(brakes_temp.iter().map(|&x| x as u32));
-        }
+        telemetry
+            .brakes_temperature
+            .extend(brakes_temp.iter().map(|&x| x as u32));
 
         telemetry.tyres_surface_temperature.clear();
         telemetry
@@ -643,13 +640,12 @@ impl PlayerTelemetry {
             .extend(data.tyres_inner_temperature.iter().map(|&x| x as u32));
 
         telemetry.tyres_pressure.clear();
-        unsafe {
-            let tyres_pressure_ptr = addr_of!(data.tyres_pressure);
 
-            telemetry
-                .tyres_pressure
-                .extend_from_slice(&tyres_pressure_ptr.read_unaligned());
-        }
+        let tyres_pressure_ptr = &raw const data.tyres_pressure;
+
+        telemetry
+            .tyres_pressure
+            .extend_from_slice(unsafe { &tyres_pressure_ptr.read_unaligned() });
     }
 
     /// Computes the difference between two PlayerTelemetry instances