split state into phases, impl command and message websocket listener tasks, impl processing each message in new task

Orciument · Orciument · commit 1f4b40c1974c · 2025-10-30T14:26:05.000+01:00
diff --git a/backend/src/axum.rs b/backend/src/axum.rs
@@ -2,7 +2,7 @@ use crate::commands::command_controller::{delete_by_id, get_all_commands, get_al
 use crate::panel_frontend::embedded_panel_server::{embedded_panel_service, to_panel_redirect, SERVER_PANEL_PATH};
 use crate::service_oauth::oauth_endpoint::{list_oauth, receive_oauth};
 use crate::webserver_authentication::auth_mod;
-use crate::AppState;
+use crate::{WebserverState};
 use axum::middleware::from_fn_with_state;
 use axum::routing::{any, delete, get, post};
 use axum::Router;
@@ -12,7 +12,7 @@ use std::sync::Arc;
 use tower_http::cors::CorsLayer;
 use url::{form_urlencoded, Url};
 
-pub type AxumState = Arc<AppState>;
+pub type AxumState = Arc<WebserverState>;
 
 pub async fn axum(on_port: u16, state: AxumState) {
     let webserver_config = {
diff --git a/backend/src/commands/command_executor_service.rs b/backend/src/commands/command_executor_service.rs
@@ -2,7 +2,7 @@ use super::cooldown_service::CooldownService;
 use crate::commands::command_controller::{Command, CooldownType, MessagePattern};
 use crate::commands::template_service::TemplateService;
 use crate::db::ProdDB;
-use crate::AppState;
+use crate::FullState;
 use num_derive::FromPrimitive;
 use regex::{Regex, RegexBuilder};
 use serde::{Deserialize, Serialize};
@@ -11,8 +11,6 @@ use std::collections::HashMap;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Instant;
-use tokio::sync::broadcast::error::RecvError;
-use tokio::sync::broadcast::Receiver;
 // ChatMessage
 
 #[allow(dead_code)]
@@ -66,7 +64,7 @@ pub struct ChatMessage {
 
 pub type TriggerId = str;
 // pub type TriggerCallback = fn(&AppState, &TriggerId, &ChatMessage) ;
-pub type TriggerCallback = fn(Arc<AppState>, Box<TriggerId>, ChatMessage) -> Pin<Box<dyn Future<Output=()>>>;
+pub type TriggerCallback = fn(Arc<FullState>, Box<TriggerId>, ChatMessage) -> Pin<Box<dyn Future<Output=()>>>;
 
 #[derive(Deserialize, Serialize)]
 pub enum ChatCooldown {
@@ -84,7 +82,7 @@ pub struct CommandTrigger {
 }
 
 static TEXT_COMMAND_CALLBACK: TriggerCallback = |app_state, trigger_id, _chat_message| Box::pin(async move {
-    match TemplateService::get_template_by_trigger_id(&app_state.prod_db, trigger_id.as_ref()).await {
+    match TemplateService::get_template_by_trigger_id(&app_state.l1.prod_db, trigger_id.as_ref()).await {
         Err(_e) => {
             // log
             // logger.debug("Executing text command {}", commandId);
@@ -93,7 +91,7 @@ static TEXT_COMMAND_CALLBACK: TriggerCallback = |app_state, trigger_id, _chat_me
             // log
             //     logger.error("Could not find template id for command id {}", commandId);
         }
-        Ok(Some(t)) => app_state.twitch_service.send_raw_template(t.template.as_ref(), HashMap::new())
+        Ok(Some(t)) => app_state.l2.twitch_service.send_raw_template(t.template.as_ref(), HashMap::new())
     }
 });
 
@@ -151,24 +149,13 @@ impl CommandExecutorService {
             .collect()
     }
 
-    #[allow(dead_code)]
-    pub async fn receive_commands(&self, app_state: Arc<AppState>, mut receiver: Receiver<ChatMessage>){
-        loop {
-            let message = match receiver.recv().await {
-                Ok(m) => m,
-                Err(RecvError::Closed) => return,
-                Err(RecvError::Lagged(_s)) => {
-                    // log skip
-                    continue;
-                }
-            };
-            for trigger in self.triggers.iter() {
-                self.execute_trigger_if_matching(app_state.clone(), trigger, message.clone()).await
-            }
+    pub async fn process_chat_message(&self, app_state: Arc<FullState>, message: ChatMessage) {
+        for trigger in self.triggers.iter() {
+            self.execute_trigger_if_matching(app_state.clone(), trigger, message.clone()).await
         }
     }
 
-    async fn execute_trigger_if_matching(&self, app_state: Arc<AppState>, trigger: &CommandTrigger, chat_message: ChatMessage) {
+    async fn execute_trigger_if_matching(&self, app_state: Arc<FullState>, 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;
diff --git a/backend/src/lib.rs b/backend/src/lib.rs
@@ -1,15 +1,14 @@
-use crate::commands::command_executor_service::CommandExecutorService;
+use crate::commands::command_executor_service::{ChatMessage, CommandExecutorService};
 use crate::commands::twitch_service::TwitchService;
 use crate::db::ProdDB;
-use service_oauth::oauth_service::OAuthService;
 use crate::session_service::SessionService;
-use rand::distr::Alphanumeric;
-use rand::Rng;
 use serde::{Deserialize, Serialize};
+use service_oauth::oauth_service::OAuthService;
 use sqlx::MySqlPool;
 use std::str::FromStr;
-use std::sync::{Arc, RwLock};
+use std::sync::{Arc, OnceLock, RwLock};
 use tokio::runtime::Handle;
+use tokio::sync::broadcast::error::RecvError;
 use url::Url;
 
 mod webserver_authentication;
@@ -26,27 +25,71 @@ pub async fn start() {
     //check if all mandatory configuration values are set
     //start mini webserver
 
+    // L0, mini-webserver stage
     let db_connection = MySqlPool::connect(std::env::var("DATABASE_URL").unwrap().as_str()).await.unwrap();
     let prod_db = ProdDB::new(db_connection);
     println!("established db connection");
 
-    let app_state = Arc::new(AppState {
-        session_service: SessionService::default(),
-        oauth_service: OAuthService::default(),
-        twitch_service: TwitchService::new(),
-        prod_db,
+    // 1st Stage
+    let l1 = Arc::new(L1State {
         webserver_config: RwLock::new(WebserverConfig {
             panel_base_url: Url::from_str("http://localhost:4771/panel").unwrap(),
             server_base_url: Url::from_str("http://localhost:4771").unwrap(),
             panel_auth_twitch_client_id: "zmxjjn3xmncg8ewew6tjk08tub26bb".to_string()
         }),
-        command_executor_service: CommandExecutorService::default(),
+        oauth_service: Default::default(),
+        session_service: Default::default(),
+        prod_db,
     });
-    let a2 = app_state.clone();
+
+    // Start Prod Webserver
+    let webserver_state = Arc::new(WebserverState {
+        l1: l1.clone(),
+        l2: OnceLock::new(),
+    });
+    let a2 = webserver_state.clone();
     Handle::current().spawn(async { axum::axum(4771, a2).await });
 
-    let oauth = app_state.oauth_service.new_oauth_request("twitch".to_string(), "account".to_string(), "".to_string(), OAuthService::random_state());
+    // 2nc Stage
+    let service = TwitchService::new(l1.clone(), ()).await.unwrap();
+    let l2 = Arc::new(L2State {
+        twitch_service: service,
+        command_executor_service: Default::default(),
+    });
+
+    // 3rd. (Full) Stage
+    let _ = webserver_state.l2.set(l2.clone());
+    let full = Arc::new(FullState {
+        l1,
+        l2,
+    });
+
+    let oauth = full.l1.oauth_service.new_oauth_request("twitch".to_string(), "account".to_string(), "".to_string(), OAuthService::random_state());
     println!("oauth: {:?}", oauth);
+
+    let (command_channel, _) = tokio::sync::broadcast::channel::<Box<ChatMessage>>(20);
+    // fanout of messages
+    let full2 = full.clone();
+    let sender2 = command_channel.clone();
+    Handle::current().spawn(async { TwitchService::start_websocket(full2, sender2) });
+    let mut receiver = command_channel.subscribe();
+    let full2 = full.clone();
+    Handle::current().spawn(async {
+        loop {
+            let message = match receiver.recv().await {
+                Ok(m) => *m,
+                Err(RecvError::Closed) => return,
+                Err(RecvError::Lagged(_s)) => {
+                    // log skip
+                    continue;
+                }
+            };
+            let full = full2.clone();
+            Handle::current().spawn(async move {
+                CommandExecutorService::process_chat_message(&full.l2.command_executor_service, full, message).await;
+            });
+        }
+    });
 }
 
 #[derive(Clone, Deserialize, Serialize)]
@@ -68,15 +111,28 @@ struct WebserverConfig {
     // maybe we need to add stuff like cors and disable auth here
 }
 
-struct AppState {
+struct L1State {
     pub prod_db: ProdDB,
     pub session_service: SessionService,
     pub oauth_service: OAuthService,
     pub webserver_config: RwLock<WebserverConfig>,
+}
+
+struct L2State {
     pub twitch_service: TwitchService,
     pub command_executor_service: CommandExecutorService
 }
 
+struct WebserverState {
+    pub l1: Arc<L1State>,
+    pub l2: OnceLock<Arc<L2State>>,
+}
+
+struct FullState {
+    pub l1: Arc<L1State>,
+    pub l2: Arc<L2State>,
+}
+
 #[allow(dead_code)]
 mod _services {
     struct SetupWebserver;
diff --git a/backend/src/service_oauth/oauth_endpoint.rs b/backend/src/service_oauth/oauth_endpoint.rs
@@ -34,7 +34,7 @@ pub async fn receive_oauth(
     Path(service): Path<String>,
     Query(query): Query<ReceiveOAuthQuery>,
 ) -> AxResult<impl IntoResponse> {
-    let panel_base_url = if let Ok(v) = state.webserver_config.read() {
+    let panel_base_url = if let Ok(v) = state.l1.webserver_config.read() {
         v.panel_base_url.as_str().to_string()
     } else {
         // log lock poisoned
@@ -50,7 +50,7 @@ pub async fn receive_oauth(
     if query.code.is_none() {
         return Ok(Body::new(format!("{}?success=false&error={}", panel_base_url, url_encode("Query param code is required for non error Oauth response"))))
     }
-    Ok(Body::new(format!("{}{}", panel_base_url, match state.oauth_service.return_oauth(service, query.state, query.scope.unwrap(), query.code.unwrap()) {
+    Ok(Body::new(format!("{}{}", panel_base_url, match state.l1.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) => {
@@ -61,5 +61,5 @@ pub async fn receive_oauth(
 }
 
 pub async fn list_oauth(State(state): State<AxumState>) -> impl IntoResponse {
-    Json(state.oauth_service.get_active_requests())
+    Json(state.l1.oauth_service.get_active_requests())
 }
