0.1.1

kb_app/src/lib.rs

@@ -15,6 +15,27 @@ pub use crate::splash::SplashOrder;
 use tauri::Emitter;
 use tauri::Manager;
 
+/// Runtime state for started WebSocket clients.
+struct KbWsRuntimeState {
+    clients: std::vec::Vec<kb_lib::WsClient>,
+    relay_tasks: std::vec::Vec<tauri::async_runtime::JoinHandle<()>>,
+}
+
+impl KbWsRuntimeState {
+    fn new() -> Self {
+        Self {
+            clients: std::vec::Vec::new(),
+            relay_tasks: std::vec::Vec::new(),
+        }
+    }
+}
+
+/// Shared application state stored inside Tauri.
+struct KbAppState {
+    config: kb_lib::KbConfig,
+    ws_runtime: tokio::sync::Mutex<KbWsRuntimeState>,
+}
+
 /// Runs the desktop application.
 #[cfg_attr(mobile, tauri::mobile_entry_point)]
 pub fn run() {
@@ -49,8 +70,15 @@ pub fn run() {
         environment = %config.app.environment,
         "starting desktop application"
     );
+    let app_state = KbAppState {
+        config: config.clone(),
+        ws_runtime: tokio::sync::Mutex::new(KbWsRuntimeState::new()),
+    };
     let tracing_builder = tauri_plugin_tracing::Builder::new();
     let mut tauri_builder = tauri::Builder::default();
+    tauri_builder = tauri_builder.manage(app_state);
+    tauri_builder =
+        tauri_builder.invoke_handler(tauri::generate_handler![start_ws_clients, stop_ws_clients]);
     tauri_builder = tauri_builder.plugin(tracing_builder.build::<tauri::Wry>());
     tauri_builder = tauri_builder.setup(|app| {
         let app_handle = app.handle().clone();
@@ -143,3 +171,234 @@ fn emit_splash_order(
         tracing::error!("error emitting splash event '{order}': {error:?}");
     }
 }
+
+#[tauri::command]
+async fn start_ws_clients(
+    app_handle: tauri::AppHandle,
+    state: tauri::State<'_, KbAppState>,
+) -> Result<usize, std::string::String> {
+    {
+        let runtime_guard = state.ws_runtime.lock().await;
+        if !runtime_guard.clients.is_empty() {
+            return Err("websocket clients are already running".to_string());
+        }
+    }
+    let enabled_endpoints: std::vec::Vec<kb_lib::KbWsEndpointConfig> = state
+        .config
+        .solana
+        .ws_endpoints
+        .iter()
+        .filter(|endpoint| endpoint.enabled)
+        .cloned()
+        .collect();
+    if enabled_endpoints.is_empty() {
+        return Err("no enabled websocket endpoint found in config.json".to_string());
+    }
+    kb_emit_app_log(
+        &app_handle,
+        &format!(
+            "[app] starting {} websocket client(s)",
+            enabled_endpoints.len()
+        ),
+    );
+    let mut started_clients: std::vec::Vec<kb_lib::WsClient> = std::vec::Vec::new();
+    let mut relay_tasks: std::vec::Vec<tauri::async_runtime::JoinHandle<()>> = std::vec::Vec::new();
+    for endpoint in enabled_endpoints {
+        kb_emit_app_log(
+            &app_handle,
+            &format!(
+                "[app] preparing websocket endpoint '{}' ({})",
+                endpoint.name, endpoint.url
+            ),
+        );
+        let client_result = kb_lib::WsClient::new(endpoint.clone());
+        let client = match client_result {
+            Ok(client) => client,
+            Err(error) => {
+                kb_shutdown_started_clients(&started_clients, &mut relay_tasks).await;
+                return Err(format!(
+                    "cannot create websocket client for endpoint '{}': {}",
+                    endpoint.name, error
+                ));
+            }
+        };
+        let mut event_receiver = client.subscribe_events();
+        let relay_app_handle = app_handle.clone();
+        let relay_task = tauri::async_runtime::spawn(async move {
+            loop {
+                let recv_result = event_receiver.recv().await;
+                match recv_result {
+                    Ok(event) => {
+                        let line = kb_format_ws_event(&event);
+                        kb_emit_app_log(&relay_app_handle, &line);
+                    }
+                    Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
+                        kb_emit_app_log(
+                            &relay_app_handle,
+                            &format!(
+                                "[ws] event receiver lagged and skipped {} message(s)",
+                                skipped
+                            ),
+                        );
+                    }
+                    Err(tokio::sync::broadcast::error::RecvError::Closed) => {
+                        break;
+                    }
+                }
+            }
+        });
+        let connect_result = client.connect().await;
+        if let Err(error) = connect_result {
+            relay_task.abort();
+            kb_shutdown_started_clients(&started_clients, &mut relay_tasks).await;
+            return Err(format!(
+                "cannot connect websocket client for endpoint '{}': {}",
+                endpoint.name, error
+            ));
+        }
+        started_clients.push(client);
+        relay_tasks.push(relay_task);
+    }
+    {
+        let mut runtime_guard = state.ws_runtime.lock().await;
+        if !runtime_guard.clients.is_empty() {
+            kb_shutdown_started_clients(&started_clients, &mut relay_tasks).await;
+            return Err("websocket clients were started concurrently".to_string());
+        }
+        runtime_guard.clients = started_clients;
+        runtime_guard.relay_tasks = relay_tasks;
+    }
+    let started_count = {
+        let runtime_guard = state.ws_runtime.lock().await;
+        runtime_guard.clients.len()
+    };
+    kb_emit_app_log(
+        &app_handle,
+        &format!("[app] {} websocket client(s) started", started_count),
+    );
+    Ok(started_count)
+}
+
+#[tauri::command]
+async fn stop_ws_clients(
+    app_handle: tauri::AppHandle,
+    state: tauri::State<'_, KbAppState>,
+) -> Result<usize, std::string::String> {
+    let (clients, mut relay_tasks) = {
+        let mut runtime_guard = state.ws_runtime.lock().await;
+        (
+            std::mem::take(&mut runtime_guard.clients),
+            std::mem::take(&mut runtime_guard.relay_tasks),
+        )
+    };
+    if clients.is_empty() {
+        kb_emit_app_log(&app_handle, "[app] websocket clients are already stopped");
+        return Ok(0);
+    }
+    kb_emit_app_log(
+        &app_handle,
+        &format!("[app] stopping {} websocket client(s)", clients.len()),
+    );
+    let stopped_count = clients.len();
+    for client in &clients {
+        let disconnect_result = client.disconnect().await;
+        if let Err(error) = disconnect_result {
+            kb_emit_app_log(
+                &app_handle,
+                &format!(
+                    "[app] disconnect error for endpoint '{}': {}",
+                    client.endpoint_name(),
+                    error
+                ),
+            );
+        }
+    }
+    for relay_task in relay_tasks.drain(..) {
+        relay_task.abort();
+    }
+    kb_emit_app_log(
+        &app_handle,
+        &format!("[app] {} websocket client(s) stopped", stopped_count),
+    );
+    Ok(stopped_count)
+}
+
+fn kb_emit_app_log(app_handle: &tauri::AppHandle, message: &str) {
+    let emit_result = app_handle.emit("kb-log", message.to_string());
+    if let Err(error) = emit_result {
+        tracing::error!("error emitting app log event: {error:?}");
+    }
+}
+
+fn kb_format_ws_event(event: &kb_lib::WsEvent) -> std::string::String {
+    match event {
+        kb_lib::WsEvent::Connected {
+            endpoint_name,
+            endpoint_url,
+        } => {
+            format!("[ws:{endpoint_name}] connected to {endpoint_url}")
+        }
+        kb_lib::WsEvent::TextMessage {
+            endpoint_name,
+            text,
+        } => {
+            format!("[ws:{endpoint_name}] text: {text}")
+        }
+        kb_lib::WsEvent::BinaryMessage {
+            endpoint_name,
+            data,
+        } => {
+            format!("[ws:{endpoint_name}] binary message ({} bytes)", data.len())
+        }
+        kb_lib::WsEvent::Ping {
+            endpoint_name,
+            data,
+        } => {
+            format!("[ws:{endpoint_name}] ping ({} bytes)", data.len())
+        }
+        kb_lib::WsEvent::Pong {
+            endpoint_name,
+            data,
+        } => {
+            format!("[ws:{endpoint_name}] pong ({} bytes)", data.len())
+        }
+        kb_lib::WsEvent::CloseReceived {
+            endpoint_name,
+            code,
+            reason,
+        } => {
+            format!(
+                "[ws:{endpoint_name}] close received code={:?} reason={:?}",
+                code, reason
+            )
+        }
+        kb_lib::WsEvent::Disconnected { endpoint_name } => {
+            format!("[ws:{endpoint_name}] disconnected")
+        }
+        kb_lib::WsEvent::Error {
+            endpoint_name,
+            error,
+        } => {
+            format!("[ws:{endpoint_name}] error: {error}")
+        }
+    }
+}
+
+async fn kb_shutdown_started_clients(
+    started_clients: &[kb_lib::WsClient],
+    relay_tasks: &mut std::vec::Vec<tauri::async_runtime::JoinHandle<()>>,
+) {
+    for client in started_clients {
+        let disconnect_result = client.disconnect().await;
+        if let Err(error) = disconnect_result {
+            tracing::error!(
+                endpoint_name = %client.endpoint_name(),
+                "cleanup disconnect error: {}",
+                error
+            );
+        }
+    }
+    for relay_task in relay_tasks.drain(..) {
+        relay_task.abort();
+    }
+}