0.3.4

kb_app/src/demo_ws.rs

@@ -0,0 +1,848 @@
+// file: kb_app/src/demo_ws.rs
+
+//! Demo WebSocket window commands and runtime state.
+//!
+//! This module isolates the manual WebSocket subscription test bench from the
+//! main application window.
+
+use tauri::Emitter;
+use tauri::Manager;
+
+/// Endpoint summary sent to the demo frontend.
+#[derive(Clone, Debug, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct KbDemoWsEndpointSummary {
+    name: std::string::String,
+    resolved_url: std::string::String,
+    provider: std::string::String,
+    enabled: bool,
+    roles: std::vec::Vec<std::string::String>,
+}
+
+/// Current demo window runtime status.
+#[derive(Clone, Debug, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct KbDemoWsStatusPayload {
+    connection_state: std::string::String,
+    endpoint_name: std::option::Option<std::string::String>,
+    endpoint_url: std::option::Option<std::string::String>,
+    current_subscription_id: std::option::Option<u64>,
+    current_subscribe_method: std::option::Option<std::string::String>,
+    current_unsubscribe_method: std::option::Option<std::string::String>,
+    current_notification_method: std::option::Option<std::string::String>,
+}
+
+/// Subscribe request sent by the demo frontend.
+#[derive(Clone, Debug, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct KbDemoWsSubscribeRequest {
+    method: std::string::String,
+    mode: std::string::String,
+    target: std::option::Option<std::string::String>,
+    filter_json: std::option::Option<std::string::String>,
+    config_json: std::option::Option<std::string::String>,
+}
+
+/// Runtime state for the demo websocket window.
+#[derive(Debug)]
+pub(crate) struct KbDemoWsRuntimeState {
+    client: std::option::Option<kb_lib::WsClient>,
+    relay_task: std::option::Option<tauri::async_runtime::JoinHandle<()>>,
+    keepalive_task: std::option::Option<tauri::async_runtime::JoinHandle<()>>,
+    endpoint_name: std::option::Option<std::string::String>,
+    endpoint_url: std::option::Option<std::string::String>,
+    connection_state: kb_lib::KbConnectionState,
+    current_subscription: std::option::Option<kb_lib::WsSubscriptionInfo>,
+}
+
+impl KbDemoWsRuntimeState {
+    /// Creates a new empty runtime state.
+    pub(crate) fn new() -> Self {
+        Self {
+            client: None,
+            relay_task: None,
+            keepalive_task: None,
+            endpoint_name: None,
+            endpoint_url: None,
+            connection_state: kb_lib::KbConnectionState::Disconnected,
+            current_subscription: None,
+        }
+    }
+
+    fn to_status_payload(&self) -> KbDemoWsStatusPayload {
+        let current_subscription_id = self
+            .current_subscription
+            .as_ref()
+            .map(|subscription| subscription.subscription_id);
+        let current_subscribe_method = self
+            .current_subscription
+            .as_ref()
+            .map(|subscription| subscription.subscribe_method.clone());
+        let current_unsubscribe_method = self
+            .current_subscription
+            .as_ref()
+            .map(|subscription| subscription.unsubscribe_method.clone());
+        let current_notification_method = self
+            .current_subscription
+            .as_ref()
+            .map(|subscription| subscription.notification_method.clone());
+        KbDemoWsStatusPayload {
+            connection_state: kb_connection_state_to_string(self.connection_state),
+            endpoint_name: self.endpoint_name.clone(),
+            endpoint_url: self.endpoint_url.clone(),
+            current_subscription_id,
+            current_subscribe_method,
+            current_unsubscribe_method,
+            current_notification_method,
+        }
+    }
+
+    fn clear(&mut self) {
+        self.client = None;
+        self.relay_task = None;
+        self.keepalive_task = None;
+        self.endpoint_name = None;
+        self.endpoint_url = None;
+        self.connection_state = kb_lib::KbConnectionState::Disconnected;
+        self.current_subscription = None;
+    }
+}
+
+/// Shows and focuses the preconfigured `demo_ws` window.
+#[tauri::command]
+pub(crate) fn open_demo_ws_window(app_handle: tauri::AppHandle) -> Result<(), std::string::String> {
+    let existing_window_option = app_handle.get_webview_window("demo_ws");
+    let demo_window = match existing_window_option {
+        Some(demo_window) => demo_window,
+        None => {
+            let builder = tauri::WebviewWindowBuilder::new(
+                &app_handle,
+                "demo_ws",
+                tauri::WebviewUrl::App("demo_ws.html".into()),
+            )
+            .title("Demo Ws Subscribe")
+            .inner_size(1400.0, 768.0)
+            .min_inner_size(800.0, 600.0)
+            .center()
+            .visible(true)
+            .transparent(false)
+            .decorations(true);
+            let build_result = builder.build();
+            match build_result {
+                Ok(window) => window,
+                Err(error) => {
+                    return Err(format!("cannot create demo_ws window: {error:?}"));
+                }
+            }
+        }
+    };
+    let show_result = demo_window.show();
+    if let Err(error) = show_result {
+        return Err(format!("cannot show demo_ws window: {error:?}"));
+    }
+    let focus_result = demo_window.set_focus();
+    if let Err(error) = focus_result {
+        return Err(format!("cannot focus demo_ws window: {error:?}"));
+    }
+    Ok(())
+}
+
+/// Returns the list of configured websocket endpoints.
+#[tauri::command]
+pub(crate) async fn demo_ws_list_endpoints(
+    state: tauri::State<'_, crate::KbAppState>,
+) -> Result<std::vec::Vec<KbDemoWsEndpointSummary>, std::string::String> {
+    let mut endpoints = std::vec::Vec::new();
+    for endpoint in &state.config.solana.ws_endpoints {
+        let resolved_url_result = endpoint.resolved_url();
+        let resolved_url = match resolved_url_result {
+            Ok(resolved_url) => resolved_url,
+            Err(_) => endpoint.url.clone(),
+        };
+        endpoints.push(KbDemoWsEndpointSummary {
+            name: endpoint.name.clone(),
+            resolved_url,
+            provider: endpoint.provider.clone(),
+            enabled: endpoint.enabled,
+            roles: endpoint.roles.clone(),
+        });
+    }
+    Ok(endpoints)
+}
+
+/// Returns the current demo websocket runtime status.
+#[tauri::command]
+pub(crate) async fn demo_ws_get_status(
+    state: tauri::State<'_, crate::KbAppState>,
+) -> Result<KbDemoWsStatusPayload, std::string::String> {
+    let runtime_guard = state.demo_ws_runtime.lock().await;
+    Ok(runtime_guard.to_status_payload())
+}
+
+/// Connects the demo websocket runtime to the selected endpoint.
+#[tauri::command]
+pub(crate) async fn demo_ws_connect(
+    app_handle: tauri::AppHandle,
+    state: tauri::State<'_, crate::KbAppState>,
+    endpoint_name: std::string::String,
+) -> Result<KbDemoWsStatusPayload, std::string::String> {
+    let endpoint_option = state.config.find_ws_endpoint(&endpoint_name);
+    let endpoint = match endpoint_option {
+        Some(endpoint) => endpoint.clone(),
+        None => {
+            return Err(format!("unknown websocket endpoint '{}'", endpoint_name));
+        }
+    };
+    let runtime_arc = state.demo_ws_runtime.clone();
+    {
+        let runtime_guard = runtime_arc.lock().await;
+        if runtime_guard.client.is_some() {
+            return Err("demo websocket client is already connected or connecting".to_string());
+        }
+    }
+    let client_result = kb_lib::WsClient::new(endpoint.clone());
+    let client = match client_result {
+        Ok(client) => client,
+        Err(error) => {
+            return Err(format!("cannot create websocket client: {error}"));
+        }
+    };
+    {
+        let mut runtime_guard = runtime_arc.lock().await;
+        runtime_guard.endpoint_name = Some(endpoint.name.clone());
+        runtime_guard.endpoint_url = Some(client.endpoint_url().to_string());
+        runtime_guard.connection_state = kb_lib::KbConnectionState::Connecting;
+        runtime_guard.current_subscription = None;
+    }
+    kb_emit_demo_ws_status(&app_handle, &runtime_arc).await;
+    kb_emit_demo_ws_log(
+        &app_handle,
+        &format!(
+            "[demo] connecting endpoint '{}' ({})",
+            endpoint.name,
+            client.endpoint_url()
+        ),
+    );
+    let mut event_receiver = client.subscribe_events();
+    let relay_runtime = runtime_arc.clone();
+    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) => {
+                    kb_apply_demo_ws_event_to_runtime(&relay_runtime, &event).await;
+                    kb_emit_demo_ws_log(&relay_app_handle, &kb_format_demo_ws_event(&event));
+                    kb_emit_demo_ws_status(&relay_app_handle, &relay_runtime).await;
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
+                    kb_emit_demo_ws_log(
+                        &relay_app_handle,
+                        &format!(
+                            "[demo] event receiver lagged and skipped {} event(s)",
+                            skipped
+                        ),
+                    );
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Closed) => {
+                    break;
+                }
+            }
+        }
+    });
+    let keepalive_client = client.clone();
+    let keepalive_app_handle = app_handle.clone();
+    let keepalive_task = tauri::async_runtime::spawn(async move {
+        kb_demo_ws_keepalive_loop(&keepalive_app_handle, &keepalive_client).await;
+    });
+    let connect_result = client.connect().await;
+    if let Err(error) = connect_result {
+        relay_task.abort();
+        keepalive_task.abort();
+        {
+            let mut runtime_guard = runtime_arc.lock().await;
+            runtime_guard.clear();
+        }
+        kb_emit_demo_ws_status(&app_handle, &runtime_arc).await;
+        return Err(format!("cannot connect websocket client: {error}"));
+    }
+    {
+        let mut runtime_guard = runtime_arc.lock().await;
+        runtime_guard.client = Some(client);
+        runtime_guard.relay_task = Some(relay_task);
+        runtime_guard.keepalive_task = Some(keepalive_task);
+        runtime_guard.endpoint_name = Some(endpoint.name.clone());
+        runtime_guard.endpoint_url = Some(endpoint.resolved_url().unwrap_or(endpoint.url));
+        runtime_guard.connection_state = kb_lib::KbConnectionState::Connected;
+    }
+    kb_emit_demo_ws_status(&app_handle, &runtime_arc).await;
+    let runtime_guard = runtime_arc.lock().await;
+    Ok(runtime_guard.to_status_payload())
+}
+
+/// Disconnects the demo websocket runtime.
+#[tauri::command]
+pub(crate) async fn demo_ws_disconnect(
+    app_handle: tauri::AppHandle,
+    state: tauri::State<'_, crate::KbAppState>,
+) -> Result<KbDemoWsStatusPayload, std::string::String> {
+    let runtime_arc = state.demo_ws_runtime.clone();
+    {
+        let mut runtime_guard = runtime_arc.lock().await;
+        runtime_guard.connection_state = kb_lib::KbConnectionState::Disconnecting;
+    }
+    kb_emit_demo_ws_status(&app_handle, &runtime_arc).await;
+    let (client_option, relay_task_option, keepalive_task_option) = {
+        let mut runtime_guard = runtime_arc.lock().await;
+        (
+            runtime_guard.client.take(),
+            runtime_guard.relay_task.take(),
+            runtime_guard.keepalive_task.take(),
+        )
+    };
+    if let Some(keepalive_task) = keepalive_task_option {
+        keepalive_task.abort();
+    }
+    if let Some(client) = &client_option {
+        let disconnect_result = client.disconnect().await;
+        if let Err(error) = disconnect_result {
+            kb_emit_demo_ws_log(
+                &app_handle,
+                &format!("[demo] disconnect error: {}", error),
+            );
+        }
+    }
+    if let Some(relay_task) = relay_task_option {
+        relay_task.abort();
+    }
+    {
+        let mut runtime_guard = runtime_arc.lock().await;
+        runtime_guard.clear();
+    }
+    kb_emit_demo_ws_status(&app_handle, &runtime_arc).await;
+    let runtime_guard = runtime_arc.lock().await;
+    Ok(runtime_guard.to_status_payload())
+}
+
+/// Sends one demo subscription request.
+#[tauri::command]
+pub(crate) async fn demo_ws_subscribe(
+    state: tauri::State<'_, crate::KbAppState>,
+    request: KbDemoWsSubscribeRequest,
+) -> Result<u64, std::string::String> {
+    let client_option = {
+        let runtime_guard = state.demo_ws_runtime.lock().await;
+        if runtime_guard.current_subscription.is_some() {
+            return Err("a subscription is already active, unsubscribe it first".to_string());
+        }
+        runtime_guard.client.clone()
+    };
+    let client = match client_option {
+        Some(client) => client,
+        None => {
+            return Err("demo websocket client is not connected".to_string());
+        }
+    };
+
+    kb_execute_demo_ws_subscribe(&client, &request).await
+}
+
+/// Sends one unsubscribe request for the current active subscription.
+#[tauri::command]
+pub(crate) async fn demo_ws_unsubscribe_current(
+    state: tauri::State<'_, crate::KbAppState>,
+) -> Result<u64, std::string::String> {
+    let (client_option, subscription_option) = {
+        let runtime_guard = state.demo_ws_runtime.lock().await;
+        (
+            runtime_guard.client.clone(),
+            runtime_guard.current_subscription.clone(),
+        )
+    };
+    let client = match client_option {
+        Some(client) => client,
+        None => {
+            return Err("demo websocket client is not connected".to_string());
+        }
+    };
+    let subscription = match subscription_option {
+        Some(subscription) => subscription,
+        None => {
+            return Err("no active subscription is currently registered".to_string());
+        }
+    };
+    let params = vec![serde_json::Value::from(subscription.subscription_id)];
+    let send_result = client
+        .send_json_rpc_request(subscription.unsubscribe_method.clone(), params)
+        .await;
+    match send_result {
+        Ok(request_id) => Ok(request_id),
+        Err(error) => Err(format!("cannot send unsubscribe request: {error}")),
+    }
+}
+
+async fn kb_execute_demo_ws_subscribe(
+    client: &kb_lib::WsClient,
+    request: &KbDemoWsSubscribeRequest,
+) -> Result<u64, std::string::String> {
+    let method = request.method.trim();
+    let mode = request.mode.trim();
+    if method == "account" {
+        let target = kb_required_target(request, "account pubkey")?;
+        if mode == "typed" {
+            let config = kb_parse_optional_json_typed::<
+                solana_rpc_client_api::config::RpcAccountInfoConfig,
+            >(&request.config_json, "account typed config")?;
+            let result = client.account_subscribe_typed(target, config).await;
+            return result.map_err(|error| format!("account typed subscribe failed: {error}"));
+        }
+        let config = kb_parse_optional_json_value(&request.config_json, "account raw config")?;
+        let result = client.account_subscribe_raw(target, config).await;
+        return result.map_err(|error| format!("account raw subscribe failed: {error}"));
+    }
+    if method == "block" {
+        if mode == "typed" {
+            let filter = kb_parse_required_json_typed::<
+                solana_rpc_client_api::config::RpcBlockSubscribeFilter,
+            >(&request.filter_json, "block typed filter")?;
+
+            let config = kb_parse_optional_json_typed::<
+                solana_rpc_client_api::config::RpcBlockSubscribeConfig,
+            >(&request.config_json, "block typed config")?;
+            let result = client.block_subscribe_typed(filter, config).await;
+            return result.map_err(|error| format!("block typed subscribe failed: {error}"));
+        }
+        let filter = kb_parse_required_json_value(&request.filter_json, "block raw filter")?;
+        let config = kb_parse_optional_json_value(&request.config_json, "block raw config")?;
+        let result = client.block_subscribe_raw(filter, config).await;
+        return result.map_err(|error| format!("block raw subscribe failed: {error}"));
+    }
+    if method == "logs" {
+        if mode == "typed" {
+            let filter = kb_parse_required_json_typed::<
+                solana_rpc_client_api::config::RpcTransactionLogsFilter,
+            >(&request.filter_json, "logs typed filter")?;
+            let config = kb_parse_optional_json_typed::<
+                solana_rpc_client_api::config::RpcTransactionLogsConfig,
+            >(&request.config_json, "logs typed config")?;
+            let result = client.logs_subscribe_typed(filter, config).await;
+            return result.map_err(|error| format!("logs typed subscribe failed: {error}"));
+        }
+        let filter = kb_parse_required_json_value(&request.filter_json, "logs raw filter")?;
+        let config = kb_parse_optional_json_value(&request.config_json, "logs raw config")?;
+        let result = client.logs_subscribe_raw(filter, config).await;
+        return result.map_err(|error| format!("logs raw subscribe failed: {error}"));
+    }
+    if method == "program" {
+        let target = kb_required_target(request, "program id")?;
+        if mode == "typed" {
+            let config = kb_parse_optional_json_typed::<
+                solana_rpc_client_api::config::RpcProgramAccountsConfig,
+            >(&request.config_json, "program typed config")?;
+            let result = client.program_subscribe_typed(target, config).await;
+            return result.map_err(|error| format!("program typed subscribe failed: {error}"));
+        }
+        let config = kb_parse_optional_json_value(&request.config_json, "program raw config")?;
+        let result = client.program_subscribe_raw(target, config).await;
+        return result.map_err(|error| format!("program raw subscribe failed: {error}"));
+    }
+    if method == "root" {
+        let result = client.root_subscribe().await;
+        return result.map_err(|error| format!("root subscribe failed: {error}"));
+    }
+    if method == "signature" {
+        let target = kb_required_target(request, "signature")?;
+        if mode == "typed" {
+            let config = kb_parse_optional_json_typed::<
+                solana_rpc_client_api::config::RpcSignatureSubscribeConfig,
+            >(&request.config_json, "signature typed config")?;
+            let result = client.signature_subscribe_typed(target, config).await;
+            return result.map_err(|error| format!("signature typed subscribe failed: {error}"));
+        }
+        let config = kb_parse_optional_json_value(&request.config_json, "signature raw config")?;
+        let result = client.signature_subscribe_raw(target, config).await;
+        return result.map_err(|error| format!("signature raw subscribe failed: {error}"));
+    }
+    if method == "slot" {
+        let result = client.slot_subscribe().await;
+        return result.map_err(|error| format!("slot subscribe failed: {error}"));
+    }
+    if method == "slotsUpdates" {
+        let result = client.slots_updates_subscribe().await;
+        return result.map_err(|error| format!("slotsUpdates subscribe failed: {error}"));
+    }
+    if method == "vote" {
+        let result = client.vote_subscribe().await;
+        return result.map_err(|error| format!("vote subscribe failed: {error}"));
+    }
+    Err(format!("unsupported demo subscribe method '{}'", method))
+}
+
+fn kb_required_target(
+    request: &KbDemoWsSubscribeRequest,
+    label: &str,
+) -> Result<std::string::String, std::string::String> {
+    let target_option = request.target.as_ref();
+    let target = match target_option {
+        Some(target) => target.trim(),
+        None => {
+            return Err(format!("{} is required", label));
+        }
+    };
+    if target.is_empty() {
+        return Err(format!("{} is required", label));
+    }
+    Ok(target.to_string())
+}
+
+fn kb_parse_optional_json_value(
+    input: &std::option::Option<std::string::String>,
+    label: &str,
+) -> Result<std::option::Option<serde_json::Value>, std::string::String> {
+    match input {
+        Some(input) => {
+            if input.trim().is_empty() {
+                return Ok(None);
+            }
+            let parse_result = serde_json::from_str::<serde_json::Value>(input);
+            match parse_result {
+                Ok(value) => Ok(Some(value)),
+                Err(error) => Err(format!("cannot parse {}: {}", label, error)),
+            }
+        }
+        None => Ok(None),
+    }
+}
+
+fn kb_parse_required_json_value(
+    input: &std::option::Option<std::string::String>,
+    label: &str,
+) -> Result<serde_json::Value, std::string::String> {
+    let input_option = input.as_ref();
+    let input = match input_option {
+        Some(input) => input.trim(),
+        None => {
+            return Err(format!("{} is required", label));
+        }
+    };
+    if input.is_empty() {
+        return Err(format!("{} is required", label));
+    }
+    let parse_result = serde_json::from_str::<serde_json::Value>(input);
+    match parse_result {
+        Ok(value) => Ok(value),
+        Err(error) => Err(format!("cannot parse {}: {}", label, error)),
+    }
+}
+
+fn kb_parse_optional_json_typed<T>(
+    input: &std::option::Option<std::string::String>,
+    label: &str,
+) -> Result<std::option::Option<T>, std::string::String>
+where
+    T: serde::de::DeserializeOwned,
+{
+    match input {
+        Some(input) => {
+            if input.trim().is_empty() {
+                return Ok(None);
+            }
+            let parse_result = serde_json::from_str::<T>(input);
+            match parse_result {
+                Ok(value) => Ok(Some(value)),
+                Err(error) => Err(format!("cannot parse {}: {}", label, error)),
+            }
+        }
+        None => Ok(None),
+    }
+}
+
+fn kb_parse_required_json_typed<T>(
+    input: &std::option::Option<std::string::String>,
+    label: &str,
+) -> Result<T, std::string::String>
+where
+    T: serde::de::DeserializeOwned,
+{
+    let input_option = input.as_ref();
+    let input = match input_option {
+        Some(input) => input.trim(),
+        None => {
+            return Err(format!("{} is required", label));
+        }
+    };
+    if input.is_empty() {
+        return Err(format!("{} is required", label));
+    }
+    let parse_result = serde_json::from_str::<T>(input);
+    match parse_result {
+        Ok(value) => Ok(value),
+        Err(error) => Err(format!("cannot parse {}: {}", label, error)),
+    }
+}
+
+async fn kb_apply_demo_ws_event_to_runtime(
+    runtime_arc: &std::sync::Arc<tokio::sync::Mutex<KbDemoWsRuntimeState>>,
+    event: &kb_lib::WsEvent,
+) {
+    let mut runtime_guard = runtime_arc.lock().await;
+    match event {
+        kb_lib::WsEvent::Connected {
+            endpoint_name,
+            endpoint_url,
+        } => {
+            runtime_guard.connection_state = kb_lib::KbConnectionState::Connected;
+            runtime_guard.endpoint_name = Some(endpoint_name.clone());
+            runtime_guard.endpoint_url = Some(endpoint_url.clone());
+        }
+        kb_lib::WsEvent::SubscriptionRegistered { subscription, .. } => {
+            runtime_guard.current_subscription = Some(subscription.clone());
+        }
+        kb_lib::WsEvent::SubscriptionUnregistered {
+            subscription_id, ..
+        } => {
+            let current_subscription_id = runtime_guard
+                .current_subscription
+                .as_ref()
+                .map(|subscription| subscription.subscription_id);
+
+            if current_subscription_id == Some(*subscription_id) {
+                runtime_guard.current_subscription = None;
+            }
+        }
+        kb_lib::WsEvent::Disconnected { .. } => {
+            runtime_guard.client = None;
+            runtime_guard.relay_task = None;
+            runtime_guard.keepalive_task = None;
+            runtime_guard.connection_state = kb_lib::KbConnectionState::Disconnected;
+            runtime_guard.current_subscription = None;
+        }
+        _ => {}
+    }
+}
+
+async fn kb_emit_demo_ws_status(
+    app_handle: &tauri::AppHandle,
+    runtime_arc: &std::sync::Arc<tokio::sync::Mutex<KbDemoWsRuntimeState>>,
+) {
+    let status_payload = {
+        let runtime_guard = runtime_arc.lock().await;
+        runtime_guard.to_status_payload()
+    };
+    let demo_window_option = app_handle.get_webview_window("demo_ws");
+    let demo_window = match demo_window_option {
+        Some(demo_window) => demo_window,
+        None => {
+            return;
+        }
+    };
+    let emit_result = demo_window.emit("demo-ws-status", status_payload);
+    if let Err(error) = emit_result {
+        tracing::error!("error emitting demo-ws-status: {error:?}");
+    }
+}
+
+fn kb_emit_demo_ws_log(app_handle: &tauri::AppHandle, line: &str) {
+    tracing::debug!("{}", line);
+    let demo_window_option = app_handle.get_webview_window("demo_ws");
+    let demo_window = match demo_window_option {
+        Some(demo_window) => demo_window,
+        None => {
+            return;
+        }
+    };
+    let emit_result = demo_window.emit("demo-ws-log", line.to_string());
+    if let Err(error) = emit_result {
+        tracing::error!("error emitting demo-ws-log: {error:?}");
+    }
+}
+
+fn kb_connection_state_to_string(state: kb_lib::KbConnectionState) -> std::string::String {
+    match state {
+        kb_lib::KbConnectionState::Disconnected => "Disconnected".to_string(),
+        kb_lib::KbConnectionState::Connecting => "Connecting".to_string(),
+        kb_lib::KbConnectionState::Connected => "Connected".to_string(),
+        kb_lib::KbConnectionState::Disconnecting => "Disconnecting".to_string(),
+    }
+}
+
+fn kb_format_demo_ws_event(event: &kb_lib::WsEvent) -> std::string::String {
+    match event {
+        kb_lib::WsEvent::Connected {
+            endpoint_name,
+            endpoint_url,
+        } => {
+            format!("[demo:{endpoint_name}] connected to {endpoint_url}")
+        }
+        kb_lib::WsEvent::TextMessage {
+            endpoint_name,
+            text,
+        } => {
+            format!(
+                "[demo:{endpoint_name}] text: {}",
+                kb_shorten_log_text(text, 1200)
+            )
+        }
+        kb_lib::WsEvent::JsonRpcMessage {
+            endpoint_name,
+            message,
+        } => {
+            let rendered = format!("{message:?}");
+            format!(
+                "[demo:{endpoint_name}] json-rpc: {}",
+                kb_shorten_log_text(&rendered, 1800)
+            )
+        }
+        kb_lib::WsEvent::JsonRpcParseError {
+            endpoint_name,
+            text,
+            error,
+        } => {
+            format!(
+                "[demo:{endpoint_name}] json-rpc parse error: {} | raw={}",
+                error,
+                kb_shorten_log_text(text, 1200)
+            )
+        }
+        kb_lib::WsEvent::SubscriptionRegistered {
+            endpoint_name,
+            subscription,
+        } => {
+            format!(
+                "[demo:{endpoint_name}] subscription registered request_id={} subscription_id={} subscribe={} unsubscribe={} notification={}",
+                subscription.request_id,
+                subscription.subscription_id,
+                subscription.subscribe_method,
+                subscription.unsubscribe_method,
+                subscription.notification_method
+            )
+        }
+        kb_lib::WsEvent::SubscriptionNotification {
+            endpoint_name,
+            subscription,
+            notification,
+            method_matches_registry,
+        } => {
+            let result_text = notification.params.result.to_string();
+            let typed_suffix = match kb_lib::parse_kb_solana_ws_typed_notification(notification) {
+                Ok(typed_notification) => {
+                    let rendered = format!("{typed_notification:?}");
+                    format!(" | typed={}", kb_shorten_log_text(&rendered, 1200))
+                }
+                Err(_) => std::string::String::new(),
+            };
+            format!(
+                "[demo:{endpoint_name}] tracked notification subscription_id={} method={} expected={} matches={} result={}{}",
+                subscription.subscription_id,
+                notification.method,
+                subscription.notification_method,
+                method_matches_registry,
+                kb_shorten_log_text(&result_text, 1600),
+                typed_suffix
+            )
+        }
+        kb_lib::WsEvent::JsonRpcNotificationWithoutSubscription {
+            endpoint_name,
+            notification,
+        } => {
+            let result_text = notification.params.result.to_string();
+            let typed_suffix = match kb_lib::parse_kb_solana_ws_typed_notification(notification) {
+                Ok(typed_notification) => {
+                    let rendered = format!("{typed_notification:?}");
+                    format!(" | typed={}", kb_shorten_log_text(&rendered, 1200))
+                }
+                Err(_) => std::string::String::new(),
+            };
+            format!(
+                "[demo:{endpoint_name}] untracked notification method={} subscription={} result={}{}",
+                notification.method,
+                notification.params.subscription,
+                kb_shorten_log_text(&result_text, 1600),
+                typed_suffix
+            )
+        }
+        kb_lib::WsEvent::SubscriptionUnregistered {
+            endpoint_name,
+            subscription_id,
+            unsubscribe_method,
+            was_active,
+        } => {
+            format!(
+                "[demo:{endpoint_name}] subscription unregistered subscription_id={} unsubscribe_method={} was_active={}",
+                subscription_id, unsubscribe_method, was_active
+            )
+        }
+        kb_lib::WsEvent::BinaryMessage {
+            endpoint_name,
+            data,
+        } => {
+            format!(
+                "[demo:{endpoint_name}] binary message ({} bytes)",
+                data.len()
+            )
+        }
+        kb_lib::WsEvent::Ping {
+            endpoint_name,
+            data,
+        } => {
+            format!("[demo:{endpoint_name}] ping ({} bytes)", data.len())
+        }
+        kb_lib::WsEvent::Pong {
+            endpoint_name,
+            data,
+        } => {
+            format!("[demo:{endpoint_name}] pong ({} bytes)", data.len())
+        }
+        kb_lib::WsEvent::CloseReceived {
+            endpoint_name,
+            code,
+            reason,
+        } => {
+            format!(
+                "[demo:{endpoint_name}] close received code={:?} reason={:?}",
+                code, reason
+            )
+        }
+        kb_lib::WsEvent::Disconnected { endpoint_name } => {
+            format!("[demo:{endpoint_name}] disconnected")
+        }
+        kb_lib::WsEvent::Error {
+            endpoint_name,
+            error,
+        } => {
+            format!("[demo:{endpoint_name}] error: {error}")
+        }
+    }
+}
+
+fn kb_shorten_log_text(input: &str, max_chars: usize) -> std::string::String {
+    let char_count = input.chars().count();
+    if char_count <= max_chars {
+        return input.to_string();
+    }
+
+    let shortened: std::string::String = input.chars().take(max_chars).collect();
+    format!("{shortened} …[truncated {} chars]", char_count - max_chars)
+}
+
+async fn kb_demo_ws_keepalive_loop(app_handle: &tauri::AppHandle, client: &kb_lib::WsClient) {
+    loop {
+        tokio::time::sleep(std::time::Duration::from_secs(30)).await;
+        let state = client.connection_state().await;
+        if state != kb_lib::KbConnectionState::Connected {
+            break;
+        }
+        let send_result = client.send_ping(b"demo-keepalive".to_vec()).await;
+        if let Err(error) = send_result {
+            kb_emit_demo_ws_log(
+                app_handle,
+                &format!(
+                    "[demo:{}] keepalive ping failed: {}",
+                    client.endpoint_name(),
+                    error
+                ),
+            );
+            break;
+        }
+    }
+}