remove static panel_base_url, implement cooldowns

Author: Orciument

rust_poc/src/commands.rs

@@ -1,11 +1,11 @@
+use crate::cooldown_service::CooldownService;
 use crate::AppState;
 use regex::Regex;
 use std::ops::Deref;
 use std::sync::Arc;
 use std::time::Instant;
 use tokio::sync::broadcast::error::RecvError;
 use tokio::sync::broadcast::Receiver;
-
 // ChatMessage
 
 #[derive(Clone, Ord, PartialOrd, PartialEq, Eq)]
@@ -23,20 +23,23 @@ pub enum TwitchUserPermission {
     System,
 }
 
+pub type TwitchUserId = str;
+
 #[derive(Clone)]
 pub struct TwitchUser {
-    pub id: String,
+    pub id: Box<TwitchUserId>,
     pub name: String,
     pub permission: TwitchUserPermission,
     pub subscriber_months: u16,
     pub subscription_tier: u16,
 }
 
+// official id, uuid
+pub type TwitchMessageId = String;
+
 #[derive(Clone)]
 pub struct ChatMessage {
-    pub message_id: String,
-    pub user_message_index: u32,
-    pub global_message_index: u32,
+    pub message_id: TwitchMessageId,
     pub message: String,
     pub user: TwitchUser,
     pub is_highlighted_message: bool,
@@ -51,26 +54,21 @@ pub struct ChatMessage {
 
 // Triggers
 
-type TriggerId<'a> = &'a str;
-type TriggerCallback = fn(&AppState, TriggerId, &ChatMessage) -> ();
-
-pub enum CooldownType {
-    SECONDS,
-    MESSAGES
-}
+pub type TriggerId = str;
+pub type TriggerCallback = fn(&AppState, &TriggerId, &ChatMessage) -> ();
 
-pub struct ChatCooldown {
-    cooldown_type: CooldownType,
-    amount: u16,
+pub enum ChatCooldown {
+    SECONDS(u32),
+    MESSAGES(u16)
 }
 
 pub struct CommandTrigger {
-    id: String,
-    patterns: Vec<Regex>,
-    permission: TwitchUserPermission,
-    user_cooldown: ChatCooldown,
-    global_cooldown: ChatCooldown,
-    callback: Box<TriggerCallback>,
+    pub id: Box<TriggerId>,
+    pub patterns: Vec<Regex>,
+    pub permission: TwitchUserPermission,
+    pub user_cooldown: ChatCooldown,
+    pub global_cooldown: ChatCooldown,
+    pub callback: Box<TriggerCallback>,
 }
 
 static TEXT_COMMAND_CALLBACK: TriggerCallback = |_app_state , _trigger_id, _chat_message| {
@@ -88,6 +86,7 @@ static TEXT_COMMAND_CALLBACK: TriggerCallback = |_app_state , _trigger_id, _chat
 
 pub struct CommandService {
     triggers: Vec<CommandTrigger>,
+    cooldown_service: CooldownService,
 }
 
 struct ReceiverClosed;
@@ -104,13 +103,13 @@ impl CommandService {
                 }
             };
             for trigger in self.triggers.iter() {
-                CommandService::execute_trigger_if_matching(app_state.deref(), trigger, &message)
+                self.execute_trigger_if_matching(app_state.deref(), trigger, &message)
             }
         }
         Ok(())
     }
 
-    fn execute_trigger_if_matching(app_state: &AppState, trigger: &CommandTrigger, chat_message: &ChatMessage) {
+    fn execute_trigger_if_matching(&self, app_state: &AppState, trigger: &CommandTrigger, chat_message: &ChatMessage) {
         if chat_message.user.permission < trigger.permission {
             // logger.debug("User {} with {}, missing {} permission for command {}", message.user().name(), message.user().permission(), trigger.permission(), trigger.id());
             return;
@@ -120,17 +119,14 @@ impl CommandService {
             return;
         }
 
-        // if (inGlobalCooldown(message, trigger.id(), trigger.globalCooldown())) {
-        //     logger.debug("Call to command {} from {} rejected because of global cooldowns", trigger.id(), message.user().name());
-        //     return;
-        // }
-        // if (inUserCooldown(message, trigger.id(), trigger.userCooldown())) {
-        //     logger.debug("Call to command {} from {} rejected because of user cooldowns", trigger.id(), message.user().name());
-        //     return;
-        // }
-        // updateCooldownState(message, trigger.id(), trigger.globalCooldown(), trigger.userCooldown());
-
-        (trigger.callback)(app_state, trigger.id.as_str(), chat_message);
+        let cooldown_res = self.cooldown_service.check_update_cooldown(chat_message, trigger.id.as_ref(), &trigger.user_cooldown, &trigger.global_cooldown);
+        if cooldown_res.is_some() {
+            // logger.debug("Call to command {} from {} rejected because of global cooldowns", trigger.id(), message.user().name());
+            // logger.debug("Call to command {} from {} rejected because of user cooldowns", trigger.id(), message.user().name());
+            return;
+        }
+
+        (trigger.callback)(app_state, trigger.id.as_ref(), chat_message);
     }
 }
 

rust_poc/src/cooldown_service.rs

@@ -0,0 +1,120 @@
+use crate::commands::{ChatCooldown, ChatMessage, TriggerId, TwitchUserId};
+use std::collections::BTreeMap;
+use std::sync::atomic::AtomicU64;
+use std::sync::atomic::Ordering::Relaxed;
+use std::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
+use std::time::Instant;
+
+#[derive(Default)]
+pub struct CooldownService {
+    global_message_counter: AtomicU64,
+    state: RwLock<InnerState>
+}
+
+type SequentialMessageId = u64;
+
+//TODO test if we can replace this lock and inner solution with concurrent: dashmap
+
+#[derive(Default)]
+struct InnerState {
+    global_message: BTreeMap<Box<TriggerId>, SequentialMessageId>,
+    global_seconds: BTreeMap<Box<TriggerId>, Instant>,
+    user_message: BTreeMap<(Box<TriggerId>, Box<TwitchUserId>), SequentialMessageId>,
+    user_seconds: BTreeMap<(Box<TriggerId>, Box<TwitchUserId>), Instant>,
+}
+
+pub enum RejectionReason {
+    GlobalCooldown,
+    UserCooldown
+}
+
+impl CooldownService {
+    /// Checks and updates the cooldowns in one operation to reduce blocking of the locks
+    /// ## **Only call this function once for each message**
+    /// No deduplication is performed, and the same message will be interpreted as two new messages.
+    /// This will lead to errors in the behavior of the cooldowns
+    pub fn check_update_cooldown(
+        &self,
+        chat_message: &ChatMessage,
+        trigger_id: &TriggerId,
+        user_cooldown: &ChatCooldown,
+        global_cooldown: &ChatCooldown,
+    ) -> Option<RejectionReason> {
+        let mut wg = self.write_guard();
+        let sequential_message_id = self.global_message_counter.fetch_add(1, Relaxed) + 1;
+        match global_cooldown {
+            ChatCooldown::MESSAGES(amount) => {
+                let last_sequential_message_id = wg.global_message.get(trigger_id);
+                if last_sequential_message_id.is_some() {
+                    let delta = sequential_message_id - last_sequential_message_id.unwrap();
+                    if delta <= *amount as u64 {
+                        return Some(RejectionReason::GlobalCooldown);
+                    }
+                }
+                wg.global_message.insert(Box::from(trigger_id), sequential_message_id);
+            }
+            ChatCooldown::SECONDS(seconds) => {
+                let last_instant = wg.global_seconds.get(trigger_id);
+                if last_instant.is_some() {
+                    let seconds_between = last_instant.unwrap().duration_since(chat_message.send_at).as_secs();
+                    if seconds_between <= *seconds as u64 {
+                        return Some(RejectionReason::GlobalCooldown);
+                    }
+                }
+                wg.global_seconds.insert(Box::from(trigger_id), chat_message.send_at);
+            }
+        };
+        match user_cooldown {
+            ChatCooldown::MESSAGES(amount) => {
+                let key = (Box::from(trigger_id), chat_message.user.id.clone());
+                let last_sequential_message_id = wg.user_message.get(&key);
+                if last_sequential_message_id.is_some() {
+                    let delta = sequential_message_id - last_sequential_message_id.unwrap();
+                    if delta <= *amount as u64 {
+                        return Some(RejectionReason::UserCooldown);
+                    }
+                }
+                wg.user_message.insert(key, sequential_message_id);
+            }
+            ChatCooldown::SECONDS(seconds) => {
+                let key = (Box::from(trigger_id), chat_message.user.id.clone());
+                let last_instant = wg.user_seconds.get(&key);
+                if last_instant.is_some() {
+                    let seconds_between = last_instant.unwrap().duration_since(chat_message.send_at).as_secs();
+                    if seconds_between <= *seconds as u64 {
+                        return Some(RejectionReason::UserCooldown);
+                    }
+                }
+                wg.user_seconds.insert(key, chat_message.send_at);
+            }
+        };
+        None
+    }
+
+    fn read_guard(&self) -> RwLockReadGuard<InnerState> {
+        match self.state.read() {
+            Ok(guard) => guard,
+            Err(poisoned) => {
+                self.state.clear_poison();
+                drop(poisoned);
+                let mut wg = self.state.write().unwrap();
+                *wg = InnerState::default();
+                self.state.read().unwrap()
+            },
+        }
+    }
+
+    fn write_guard(&self) -> RwLockWriteGuard<InnerState> {
+        match self.state.write() {
+            Ok(guard) => guard,
+            Err(poisoned) => {
+                self.state.clear_poison();
+                drop(poisoned);
+                let mut wg = self.state.write().unwrap();
+                *wg = InnerState::default();
+                wg
+            },
+        }
+    }
+
+}

rust_poc/src/lib.rs

@@ -3,10 +3,13 @@ use crate::oauth_service::OAuthService;
 use crate::session_service::SessionService;
 use rand::distr::Alphanumeric;
 use rand::Rng;
+use serde::{Deserialize, Serialize};
 use sqlx::mysql::MySqlConnectOptions;
 use sqlx::MySqlPool;
-use std::sync::Arc;
+use std::str::FromStr;
+use std::sync::{Arc, RwLock};
 use tokio::runtime::Handle;
+use url::Url;
 
 mod oauth_service;
 mod webserver_authentication;
@@ -17,8 +20,7 @@ mod session_service;
 mod panel_user;
 mod panel_user_service;
 mod commands;
-
-pub(crate) static PANEL_BASE_URL: &'static str = "http://localhost:5173";
+mod cooldown_service;
 
 pub async fn start() {
     //check if all mandatory configuration values are set
@@ -34,6 +36,10 @@ pub async fn start() {
         session_service: SessionService::default(),
         oauth_service: OAuthService::default(),
         prod_db,
+        webserver_config: RwLock::new(WebserverConfig {
+            panel_base_url: Url::from_str("http://localhost:5173").unwrap(),
+            server_base_url: Url::from_str("http://localhost:4771").unwrap(),
+        })
     });
     let a2 = app_state.clone();
     Handle::current().spawn(async { axum::axum(5000, a2).await });
@@ -44,10 +50,27 @@ pub async fn start() {
     println!("oauth: {:?}", oauth);
 }
 
+#[derive(Clone, Deserialize, Serialize)]
+struct DbConfig {
+    db_host: String,
+    db_port: u16,
+    db_username: String,
+    db_password: String,
+    db_database: String,
+}
+
+/// Very basic, but can already be saved in the Database
+struct WebserverConfig {
+    panel_base_url: Url,
+    server_base_url: Url
+    // maybe we need to add stuff like cors and disable auth here
+}
+
 struct AppState {
     pub prod_db: ProdDB,
     pub session_service: SessionService,
     pub oauth_service: OAuthService,
+    pub webserver_config: RwLock<WebserverConfig>
 }
 
 #[allow(dead_code)]
@@ -62,7 +85,7 @@ mod _services {
     // AuthenticationService;
     struct PanelWebServer;
     struct WatchtimeService;
-    struct CommandsService;
+    // CommandsService;
     struct TimerService;
     struct GiveawayService;
     struct StreamInfoEditorService;

rust_poc/src/oauth_endpoint.rs

@@ -1,11 +1,11 @@
 use crate::axum::{url_encode, AxumState};
 use crate::oauth_service::OauthReturnError;
 use crate::webserver_authentication::Moderator;
-use crate::PANEL_BASE_URL;
 use axum::body::Body;
 use axum::extract::{Path, Query, State};
 use axum::response::IntoResponse;
 use axum::Json;
+use reqwest::StatusCode;
 use serde::Deserialize;
 
 /// Get redirect url for a particular service. The url is fully formed with the host accessible from the outside.
@@ -34,24 +34,30 @@ pub async fn receive_oauth(
     Path(service): Path<String>,
     Query(query): Query<ReceiveOAuthQuery>,
 ) -> impl IntoResponse {
+    let panel_base_url = if let Ok(v) = state.webserver_config.read() {
+        v.panel_base_url.as_str().to_string()
+    } else {
+        // log lock poisoned
+        return StatusCode::INTERNAL_SERVER_ERROR.into_response();
+    };
     // Body:new( should be Redirect:to(& but for testing with postman, this is deactivated
     if query.error.is_some() || query.error_description.is_some() {
-        return Body::new(format!("{}?success=false&error={}", PANEL_BASE_URL, url_encode(&query.error_description.unwrap_or(query.error.unwrap()))));
+        return Body::new(format!("{}?success=false&error={}", panel_base_url, url_encode(&query.error_description.unwrap_or(query.error.unwrap())))).into_response();
     }
     if query.scope.is_none() {
-        return Body::new(format!("{}?success=false&error={}", PANEL_BASE_URL, url_encode("Query param scope is required for non error Oauth response")))
+        return Body::new(format!("{}?success=false&error={}", panel_base_url, url_encode("Query param scope is required for non error Oauth response"))).into_response()
     }
     if query.code.is_none() {
-        return Body::new(format!("{}?success=false&error={}", PANEL_BASE_URL, url_encode("Query param code is required for non error Oauth response")))
+        return Body::new(format!("{}?success=false&error={}", panel_base_url, url_encode("Query param code is required for non error Oauth response"))).into_response()
     }
-    Body::new(format!("{PANEL_BASE_URL}{}", match state.oauth_service.return_oauth(service, query.state, query.scope.unwrap(), query.code.unwrap()) {
-        Ok(()) => format!("{}?success=true", PANEL_BASE_URL),
-        Err(OauthReturnError::NotRequested) => format!("{}?success=false&error={}", PANEL_BASE_URL, url_encode("This oauth was never requested from the bot")),
+    Body::new(format!("{}{}", panel_base_url, match state.oauth_service.return_oauth(service, query.state, query.scope.unwrap(), query.code.unwrap()) {
+        Ok(()) => format!("{}?success=true", panel_base_url),
+        Err(OauthReturnError::NotRequested) => format!("{}?success=false&error={}", panel_base_url, url_encode("This oauth was never requested from the bot")),
         Err(OauthReturnError::ReturnChannelClosed) => {
             eprintln!("Failed to process auth code, return channel was closed");
-            format!("{}?success=false&error={}", PANEL_BASE_URL, url_encode("Could not process auth code. Internal server error"))
+            format!("{}?success=false&error={}", panel_base_url, url_encode("Could not process auth code. Internal server error"))
         }
-    }))
+    })).into_response()
 }
 
 pub async fn list_oauth(_: Moderator, State(state): State<AxumState>) -> impl IntoResponse {