fix command_repo transactions, update command_executor

Author: Orciument

backend/src/commands/command_controller.rs

@@ -11,6 +11,7 @@ use num_derive::FromPrimitive;
 use serde::{Deserialize, Serialize};
 use sqlx::Type;
 use std::str::FromStr;
+use crate::commands::command_repo::SaveCommandError;
 
 #[derive(Deserialize, Serialize)]
 pub struct MessagePattern {
@@ -38,12 +39,11 @@ pub struct Command {
 #[derive(Deserialize, Serialize, Type, FromPrimitive)]
 #[repr(u8)]
 pub enum CooldownType {
-    Seconds = 0,
-    Messages = 1,
+    SECONDS = 0,
+    MESSAGES = 1,
 }
 
 //TODO use new return type that only returns enough information to render commands table, request the entire object on edit open
-//TODO update command_executor
 pub async fn get_all_user_commands(
     State(state): State<AxumState>,
     Query(search): Query<String>,
@@ -89,7 +89,7 @@ pub async fn set_enabled(
 ) -> AxResult<impl IntoResponse> {
     //TODO move into query parameter
     let enabled = bool::from_str(body.as_ref()).map_err(|_| StatusCode::BAD_REQUEST)?;
-    command_repo::set_enabled(&state.prod_db, trigger_id.as_ref(), enabled)
+    command_repo::set_enabled(state, trigger_id.as_ref(), enabled)
         .await
         .map_err(|_| {
             // log
@@ -104,8 +104,8 @@ pub async fn set_visible(
     Query(trigger_id): Query<String>,
     body: String
 ) -> AxResult<impl IntoResponse> {
-    let visible = bool::from_str(body.as_ref()).map_err(|_| StatusCode::BAD_REQUEST)?;
-    command_repo::set_visible(&state.prod_db, trigger_id.as_ref(), visible)
+    let visible = bool::from_str(body.as_str()).map_err(|_| StatusCode::BAD_REQUEST)?;
+    command_repo::set_visible(&state.prod_db, trigger_id.as_str(), visible)
         .await
         .map_err(|_| {
             // log
@@ -118,19 +118,26 @@ pub async fn set_visible(
 pub async fn save(
     State(state): State<AxumState>,
     Json(to_save): Json<Command>,
-) -> AxResult<impl IntoResponse> {
-    command_repo::save(&state.prod_db, &to_save).await.map_err(|_| {
-        // log
-        StatusCode::INTERNAL_SERVER_ERROR
-    })?;
-    Ok(())
+) -> impl IntoResponse {
+    match command_repo::save(state, &to_save).await {
+        Ok(()) => StatusCode::OK,
+        Err(SaveCommandError::DbError(db_error)) => {
+            // log
+            StatusCode::INTERNAL_SERVER_ERROR
+        }
+        Err(SaveCommandError::RegexError(r)) => {
+            // log, but actually more return. This error needs to reach the user in the panel
+            //TODO figure out how to return this error to the user
+            StatusCode::BAD_REQUEST
+        }
+    }
 }
 
 pub async fn delete_by_id(
     State(state): State<AxumState>,
     Path(trigger_id): Path<String>,
 ) -> AxResult<impl IntoResponse> {
-    command_repo::delete_by_id(&state.prod_db, trigger_id.as_ref()).await.map_err(|_| {
+    command_repo::delete_by_id(state, trigger_id.as_ref()).await.map_err(|_| {
         // log
         StatusCode::INTERNAL_SERVER_ERROR
     })?;

backend/src/commands/command_executor_service.rs

@@ -2,7 +2,7 @@ use super::cooldown_service::CooldownService;
 use crate::commands::template_service::TemplateService;
 use crate::AppState;
 use num_derive::FromPrimitive;
-use regex::Regex;
+use regex::{Regex, RegexBuilder, RegexSet};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::pin::Pin;
@@ -11,7 +11,8 @@ use std::time::Instant;
 use sqlx::Type;
 use tokio::sync::broadcast::error::RecvError;
 use tokio::sync::broadcast::Receiver;
-
+use crate::commands::command_controller::{Command, CooldownType, MessagePattern};
+use crate::db::ProdDB;
 // ChatMessage
 
 #[allow(dead_code)]
@@ -98,13 +99,56 @@ static TEXT_COMMAND_CALLBACK: TriggerCallback = |app_state, trigger_id, _chat_me
 
 // Service
 
+#[derive(Default)]
 pub struct CommandExecutorService {
     triggers: Vec<CommandTrigger>,
     cooldown_service: CooldownService,
 }
 
-
 impl CommandExecutorService {
+    pub(crate) fn remove_command(&self, command_id: &TriggerId) {
+        // self.triggers.retain(|c| c.id.as_ref() != command_id);
+    }
+
+    pub(crate) fn upsert_command(&self, command: &Command) -> Result<(), regex::Error> {
+        self.remove_command(command.id.as_ref());
+        let global_cooldown = match command.global_cooldown_type {
+            CooldownType::SECONDS => ChatCooldown::SECONDS(command.global_cooldown_amount),
+            CooldownType::MESSAGES => ChatCooldown::MESSAGES(command.global_cooldown_amount)
+        };
+        let user_cooldown = match command.user_cooldown_type {
+            CooldownType::SECONDS => ChatCooldown::SECONDS(command.user_cooldown_amount),
+            CooldownType::MESSAGES => ChatCooldown::MESSAGES(command.user_cooldown_amount)
+        };
+        // self.triggers.push(CommandTrigger {
+        //     id: command.id.clone().into_boxed_str(),
+        //     global_cooldown,
+        //     user_cooldown,
+        //     permission: command.permission,
+        //     patterns: Self::convert_patterns(command.patterns.as_ref())?,
+        //     callback: TEXT_COMMAND_CALLBACK
+        // });
+        Ok(())
+    }
+
+    pub(crate) fn refresh_patterns(&self, prod_db: &ProdDB, trigger_id: &TriggerId) {
+        todo!()
+    }
+
+    fn convert_patterns(patterns: &[MessagePattern]) -> Result<Vec<Regex>,regex::Error> {
+        patterns.iter()
+            .filter(|x| x.is_enabled)
+            .map(|x1| {
+                match x1.is_regex {
+                    true => Regex::new(x1.pattern.as_str()),
+                    false => RegexBuilder::new(format!("^{}(?: |$).*", &x1.pattern).as_str())
+                        .case_insensitive(true)
+                        .build(),
+                }
+            })
+            .collect()
+    }
+
     #[allow(dead_code)]
     pub async fn receive_commands(&self, app_state: Arc<AppState>, mut receiver: Receiver<ChatMessage>){
         loop {

backend/src/commands/command_repo.rs

@@ -1,12 +1,13 @@
+use crate::axum::AxumState;
 use crate::commands::command_controller::{Command, CooldownType, MessagePattern};
 use crate::commands::command_executor_service::{TriggerId, TwitchUserPermission};
 use crate::commands::template_service::StringTemplate;
 use crate::db::ProdDB;
-use anyhow::{Context, Error};
+use anyhow::Context;
 use num_traits::FromPrimitive;
-use sqlx::{query, query_as, AnyExecutor, Execute, MySql, MySqlConnection, MySqlExecutor, Transaction};
-use std::ops::{Deref, DerefMut};
 use sqlx::mysql::MySqlQueryResult;
+use sqlx::{query, query_as, MySql, MySqlExecutor, Transaction};
+use std::ops::{Deref, DerefMut};
 
 pub struct CommandTable {
     id: String,
@@ -195,15 +196,18 @@ async fn get_patterns(prod_db: &ProdDB, d: Vec<CommandTable>) -> anyhow::Result<
     Ok(commands)
 }
 
-pub(crate) async fn set_enabled(prod_db: &ProdDB, trigger_id: &TriggerId, enabled: bool) -> anyhow::Result<Option<()>> {
+pub(crate) async fn set_enabled(state: AxumState, trigger_id: &TriggerId, enabled: bool) -> anyhow::Result<Option<()>> {
+    let mut transaction = state.prod_db.begin().await?;
     let affected = query!("UPDATE `sys-chat_trigger-patterns` SET is_enabled = ? WHERE parent_trigger_id = ?", enabled, trigger_id)
-        .execute(prod_db.deref())
+        .execute(transaction.deref_mut())
         .await
         .context("failed to set enabled on command patterns")?
         .rows_affected();
     if affected == 0 {
         return Ok(None)
     }
+    state.command_executor_service.refresh_patterns(&state.prod_db, trigger_id);
+    transaction.commit().await?;
     Ok(Some(()))
 }
 
@@ -219,8 +223,25 @@ pub(crate) async fn set_visible(prod_db: &ProdDB, trigger_id: &TriggerId, visibl
     Ok(Some(()))
 }
 
-pub(crate) async fn save(prod_db: &ProdDB, command: &Command) -> anyhow::Result<()> {
-    let mut transaction = prod_db.begin().await?;
+pub(crate) enum SaveCommandError {
+    DbError(anyhow::Error),
+    RegexError(regex::Error),
+}
+
+pub(crate) async fn save(state: AxumState, command: &Command) -> Result<(), SaveCommandError> {
+    let mut transaction = state.prod_db.begin().await.context("failed to start save command transaction")
+        .map_err(|e| SaveCommandError::DbError(e))?;
+    save_command(&command, &mut transaction).await
+        .map_err(|e| SaveCommandError::DbError(e))?;
+
+    state.command_executor_service.upsert_command(&command)
+        .map_err(|e| SaveCommandError::RegexError(e))?;
+    transaction.commit().await.context("failed to commit save command transaction")
+        .map_err(|e| SaveCommandError::DbError(e))?;
+    Ok(())
+}
+
+async fn save_command<'a>(command: &Command, transaction: &mut Transaction<'a, MySql>) -> anyhow::Result<()> {
     if let Some(template) = &command.template {
         save_template(transaction.deref_mut(), template).await?;
     }
@@ -233,14 +254,13 @@ pub(crate) async fn save(prod_db: &ProdDB, command: &Command) -> anyhow::Result<
             id = ?, is_auto_generated = ?, description = ?, global_cooldown_amount = ?, global_cooldown_type = ?, permission = ?, user_cooldown_amount = ?, user_cooldown_type = ?, template_id = ?
             "#, command.id, command.is_auto_generated, command.description, command.global_cooldown_amount, command.global_cooldown_type, command.permission, command.user_cooldown_amount, command.user_cooldown_type, template_id,
         command.id, command.is_auto_generated, command.description, command.global_cooldown_amount, command.global_cooldown_type, command.permission, command.user_cooldown_amount, command.user_cooldown_type, template_id,
-    )
-        .execute(prod_db.deref())
+    ).execute(transaction.deref_mut())
         .await
         .context("failed to set visible on command patterns")?;
     Ok(())
 }
 
-async fn save_pattern<'a, E: MySqlExecutor<'a>>(prod_db: E, pattern: &MessagePattern, command_id: &str) -> Result<MySqlQueryResult, Error> {
+async fn save_pattern<'a, E: MySqlExecutor<'a>>(prod_db: E, pattern: &MessagePattern, command_id: &str) -> Result<MySqlQueryResult, anyhow::Error> {
     query!(r#"INSERT `sys-chat_trigger-patterns` (pattern, is_enabled, is_regex, is_visible, parent_trigger_id)
                 VALUE (?, ?, ?, ? , ?) ON DUPLICATE KEY UPDATE
                 pattern = ?, is_enabled = ?, is_regex = ?, is_visible = ?, parent_trigger_id = ?
@@ -251,23 +271,25 @@ async fn save_pattern<'a, E: MySqlExecutor<'a>>(prod_db: E, pattern: &MessagePat
         .context("failed to save command template")
 }
 
-async fn save_template<'a, E: MySqlExecutor<'a>>(prod_db: E, template: &StringTemplate) -> Result<MySqlQueryResult, Error> {
+async fn save_template<'a, E: MySqlExecutor<'a>>(prod_db: E, template: &StringTemplate) -> Result<MySqlQueryResult, anyhow::Error> {
     query!("INSERT `sys-string_templates` (id, message_color, template) VALUE (?, ?, ?) ON DUPLICATE KEY UPDATE template = ?, message_color = ?",
             template.id, template.template, template.message_color, template.template, template.message_color)
         .execute(prod_db)
         .await
         .context("failed to save command template")
 }
 
-pub(crate) async fn delete_by_id(prod_db: &ProdDB, trigger_id: &TriggerId) -> anyhow::Result<()> {
-     query!("DELETE FROM `sys-string_templates` WHERE id = (SELECT template_id FROM `sys-chat_trigger-trigger` WHERE id = ?)", trigger_id)
-        .execute(prod_db.deref())
+pub(crate) async fn delete_by_id(state: AxumState, trigger_id: &TriggerId) -> anyhow::Result<()> {
+    let pool = state.prod_db.deref();
+    query!("DELETE FROM `sys-string_templates` WHERE id = (SELECT template_id FROM `sys-chat_trigger-trigger` WHERE id = ?)", trigger_id)
+        .execute(pool)
         .await
         .context("failed to delete string template")?;
      query!("DELETE FROM `sys-chat_trigger-trigger` WHERE id = ?", trigger_id)
-        .execute(prod_db.deref())
+        .execute(pool)
         .await
         .context("failed to delete command trigger")?;
+    state.command_executor_service.remove_command(&trigger_id);
     Ok(())
 }
 

backend/src/lib.rs

@@ -11,6 +11,7 @@ use std::str::FromStr;
 use std::sync::{Arc, RwLock};
 use tokio::runtime::Handle;
 use url::Url;
+use crate::commands::command_executor_service::CommandExecutorService;
 
 mod oauth_service;
 mod webserver_authentication;
@@ -40,7 +41,8 @@ pub async fn start() {
         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(),
-        })
+        }),
+        command_executor_service: CommandExecutorService::default(),
     });
     let a2 = app_state.clone();
     Handle::current().spawn(async { axum::axum(5000, a2).await });
@@ -73,6 +75,7 @@ struct AppState {
     pub oauth_service: OAuthService,
     pub webserver_config: RwLock<WebserverConfig>,
     pub twitch_service: TwitchService,
+    pub command_executor_service: CommandExecutorService
 }
 
 #[allow(dead_code)]