diff --git a/kb-app-demo-desktop/frontend/main.html b/kb-app-demo-desktop/frontend/main.html index 0734ca5..8aba313 100644 --- a/kb-app-demo-desktop/frontend/main.html +++ b/kb-app-demo-desktop/frontend/main.html @@ -1,5 +1,5 @@ - + @@ -22,6 +22,7 @@ Démos diff --git a/kb-app-demo-desktop/frontend/ts/main.ts b/kb-app-demo-desktop/frontend/ts/main.ts index a31caea..2c48769 100644 --- a/kb-app-demo-desktop/frontend/ts/main.ts +++ b/kb-app-demo-desktop/frontend/ts/main.ts @@ -37,6 +37,14 @@ document.addEventListener("DOMContentLoaded", async () => { } }); +const openDemoWsLink = document.querySelector("#openDemoWsLink"); +if (openDemoWsLink) { + openDemoWsLink.addEventListener("click", event => { + event.preventDefault(); + void invoke("open_demo_ws_window"); + }); +} + const openDemoHttpLink = document.querySelector("#openDemoHttpLink"); if (openDemoHttpLink) { openDemoHttpLink.addEventListener("click", event => { diff --git a/kb-app-demo-desktop/src/app_state.rs b/kb-app-demo-desktop/src/app_state.rs index 0125f83..89014a2 100644 --- a/kb-app-demo-desktop/src/app_state.rs +++ b/kb-app-demo-desktop/src/app_state.rs @@ -1,5 +1,5 @@ // file: kb-app-demo-desktop/src/app_state.rs -// version: 2 +// version: 3 //! Shared Tauri application state and startup initialization. @@ -10,6 +10,8 @@ pub(crate) struct AppState { active_profile: kb_config::ProfileConfig, logging_guard: std::sync::Mutex, http_pool: kb_onchain_transport::HttpEndpointPool, + ws_pool: kb_onchain_transport::WsEndpointPool, + demo_ws_session: tokio::sync::Mutex>>, demo_backfill_running: std::sync::atomic::AtomicBool, demo_backfill_cancel_requested: std::sync::atomic::AtomicBool, } @@ -31,8 +33,11 @@ impl crate::AppState { std::result::Result::Ok(guard) => guard, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - let http_pool = match kb_onchain_transport::HttpEndpointPool::from_profile(&active_profile) - { + let http_pool = match kb_onchain_transport::HttpEndpointPool::from_profile(&active_profile) { + std::result::Result::Ok(pool) => pool, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let ws_pool = match kb_onchain_transport::WsEndpointPool::from_profile(&active_profile) { std::result::Result::Ok(pool) => pool, std::result::Result::Err(error) => return std::result::Result::Err(error), }; @@ -42,6 +47,8 @@ impl crate::AppState { active_profile, logging_guard: std::sync::Mutex::new(logging_guard), http_pool, + ws_pool, + demo_ws_session: tokio::sync::Mutex::new(std::option::Option::None), demo_backfill_running: std::sync::atomic::AtomicBool::new(false), demo_backfill_cancel_requested: std::sync::atomic::AtomicBool::new(false), }); @@ -67,6 +74,20 @@ impl crate::AppState { return &self.http_pool; } + /// Returns the configured WebSocket endpoint pool. + pub(crate) fn ws_pool(&self) -> &kb_onchain_transport::WsEndpointPool { + return &self.ws_pool; + } + + /// Returns the persistent WebSocket demo session slot. + pub(crate) fn demo_ws_session( + &self, + ) -> &tokio::sync::Mutex< + std::option::Option>, + > { + return &self.demo_ws_session; + } + /// Returns the single-campaign execution flag used by the backfill window. pub(crate) fn demo_backfill_running(&self) -> &std::sync::atomic::AtomicBool { return &self.demo_backfill_running; diff --git a/kb-app-demo-desktop/src/demo_ws.rs b/kb-app-demo-desktop/src/demo_ws.rs new file mode 100644 index 0000000..6ccbda4 --- /dev/null +++ b/kb-app-demo-desktop/src/demo_ws.rs @@ -0,0 +1,869 @@ +// file: kb-app-demo-desktop/src/demo_ws.rs +// version: 14 + +//! Standard Solana WebSocket demo commands backed by `kb_onchain_transport::WsSession`. + +use tauri::Emitter; // rust-rules: trait-import +use tauri::Manager; // rust-rules: trait-import +use ts_rs::TS; // rust-rules: derive-import + +/// One selectable role shown by the WebSocket demo. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_ws/DemoWsRoleOption.ts" +)] +pub(crate) struct DemoWsRoleOption { + /// Endpoint role code. + pub(crate) role: std::string::String, + /// Request kinds accepted by this role. + pub(crate) request_kinds: std::vec::Vec, +} + +/// One selectable WebSocket JSON-RPC method shown by the WebSocket demo. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_ws/DemoWsMethodOption.ts" +)] +pub(crate) struct DemoWsMethodOption { + /// JSON-RPC subscribe method name. + pub(crate) method: std::string::String, + /// Matching unsubscribe method name. + pub(crate) unsubscribe_method: std::string::String, + /// Derived request kind used for endpoint routing. + pub(crate) request_kind: std::string::String, + /// Human-readable method label. + pub(crate) label: std::string::String, + /// Whether this method needs the target field. + pub(crate) requires_target: bool, + /// Whether this method needs the filter JSON field. + pub(crate) requires_filter_json: bool, + /// Whether this method supports an optional configuration object. + pub(crate) supports_config_json: bool, +} + +/// WebSocket demo options derived from configuration and the standard registry. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_ws/DemoWsOptionsPayload.ts" +)] +pub(crate) struct DemoWsOptionsPayload { + /// Selectable roles. + pub(crate) roles: std::vec::Vec, + /// Selectable methods. + pub(crate) methods: std::vec::Vec, +} + +/// Request payload for one WebSocket subscription demo command. +#[derive(Clone, Debug, serde::Deserialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_ws/DemoWsRequest.ts" +)] +pub(crate) struct DemoWsRequest { + /// Required endpoint role used by the WebSocket pool. + pub(crate) role: std::string::String, + /// JSON-RPC WebSocket subscribe method name. + pub(crate) method: std::string::String, + /// Optional target string used by account/program/signature subscriptions. + pub(crate) target: std::option::Option, + /// Optional JSON filter string used by logsSubscribe or blockSubscribe. + pub(crate) filter_json: std::option::Option, + /// Optional JSON configuration string. + pub(crate) config_json: std::option::Option, +} + +/// Response payload for one WebSocket subscription response. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_ws/DemoWsExecutionPayload.ts" +)] +pub(crate) struct DemoWsExecutionPayload { + /// Selected endpoint name. + pub(crate) endpoint_name: std::string::String, + /// Selected provider name. + pub(crate) provider: std::string::String, + /// Selected endpoint URL. + pub(crate) endpoint_url: std::string::String, + /// Required role used by the selection. + pub(crate) role: std::string::String, + /// JSON-RPC method name. + pub(crate) method: std::string::String, + /// Derived request kind. + pub(crate) request_kind: std::string::String, + /// Parsed response kind. + pub(crate) response_kind: std::string::String, + /// Remote subscription id returned by the node. + #[ts(type = "number | null")] + pub(crate) subscription_id: std::option::Option, + /// Pretty JSON response text. + pub(crate) response_json: std::string::String, +} + +/// Current WebSocket demo session status. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_ws/DemoWsStatusPayload.ts" +)] +pub(crate) struct DemoWsStatusPayload { + /// Whether a session is currently connected. + pub(crate) connected: bool, + /// Selected endpoint name. + pub(crate) endpoint_name: std::option::Option, + /// Selected endpoint URL. + pub(crate) endpoint_url: std::option::Option, + /// Last subscription method. + pub(crate) method: std::option::Option, + /// Last unsubscribe method. + pub(crate) unsubscribe_method: std::option::Option, + /// Last remote subscription id. + #[ts(type = "number | null")] + pub(crate) subscription_id: std::option::Option, + /// Number of active subscriptions on the current connection. + pub(crate) subscription_count: u32, + /// Active subscriptions kept by the session. + pub(crate) subscriptions: std::vec::Vec, +} + +/// One active WebSocket subscription shown by the demo UI. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_ws/DemoWsSubscriptionStatusPayload.ts" +)] +pub(crate) struct DemoWsSubscriptionStatusPayload { + /// Remote subscription id. + #[ts(type = "number")] + pub(crate) subscription_id: u64, + /// Subscribe method used to create this subscription. + pub(crate) method: std::string::String, + /// Matching unsubscribe method. + pub(crate) unsubscribe_method: std::string::String, +} + +/// Request payload for one explicit WebSocket unsubscribe command. +#[derive(Clone, Debug, serde::Deserialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_ws/DemoWsUnsubscribeRequest.ts" +)] +pub(crate) struct DemoWsUnsubscribeRequest { + /// Remote subscription id to unsubscribe. + #[ts(type = "number")] + pub(crate) subscription_id: u64, +} + +/// WebSocket event payload emitted to the demo window. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_ws/DemoWsMessagePayload.ts" +)] +pub(crate) struct DemoWsMessagePayload { + /// Message kind. + pub(crate) kind: std::string::String, + /// Pretty JSON or text payload. + pub(crate) payload_json: std::string::String, +} + +const DEMO_WS_UI_RATE_WINDOW_MS: u64 = 1000; +const DEMO_WS_UI_MAX_MESSAGES_PER_WINDOW: u32 = 12; +const DEMO_WS_UI_MAX_PAYLOAD_CHARS: usize = 8_000; +const DEMO_WS_RECONNECT_ATTEMPTS: u32 = 3; +const DEMO_WS_RECONNECT_INITIAL_DELAY_MS: u64 = 250; +const DEMO_WS_RECONNECT_MAX_DELAY_MS: u64 = 2_000; + +struct DemoWsUiRateLimiter { + window_started_at: std::time::Instant, + emitted_in_window: u32, + dropped_in_window: u32, +} + +impl DemoWsUiRateLimiter { + fn new() -> Self { + return Self { + window_started_at: std::time::Instant::now(), + emitted_in_window: 0, + dropped_in_window: 0, + }; + } + + fn should_emit(&mut self, app_handle: &tauri::AppHandle) -> bool { + let now = std::time::Instant::now(); + if now.duration_since(self.window_started_at) + >= std::time::Duration::from_millis(DEMO_WS_UI_RATE_WINDOW_MS) + { + if self.dropped_in_window > 0 { + emit_demo_ws_message( + app_handle, + "throttled", + format!( + "{} WebSocket notification(s) masquée(s) pendant la dernière fenêtre UI.", + self.dropped_in_window + ), + ); + } + self.window_started_at = now; + self.emitted_in_window = 0; + self.dropped_in_window = 0; + } + if self.emitted_in_window >= DEMO_WS_UI_MAX_MESSAGES_PER_WINDOW { + self.dropped_in_window = self.dropped_in_window.saturating_add(1); + return false; + } + self.emitted_in_window = self.emitted_in_window.saturating_add(1); + return true; + } +} + +pub(crate) async fn demo_ws_status_inner( + state: &crate::AppState, +) -> std::result::Result { + let session = { + let guard = state.demo_ws_session().lock().await; + guard.clone() + }; + return match session { + std::option::Option::Some(session) => { + std::result::Result::Ok(status_from_snapshot(session.snapshot().await)) + }, + std::option::Option::None => std::result::Result::Ok(disconnected_status()), + }; +} + +pub(crate) async fn demo_ws_connect_inner( + app_handle: tauri::AppHandle, + state: &crate::AppState, + request: crate::DemoWsRequest, +) -> std::result::Result { + let role = request.role.trim().to_string(); + if role.is_empty() { + return std::result::Result::Err("demo WebSocket role must not be empty".to_string()); + } + let method = request.method.trim().to_string(); + if method.is_empty() { + return std::result::Result::Err("demo WebSocket method must not be empty".to_string()); + } + let standard_request = match build_standard_ws_request( + &method, + request.target, + request.filter_json, + request.config_json, + ) { + std::result::Result::Ok(request) => request, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let selected_client = match state.ws_pool().select_client_for_role_and_method(&role, &method) { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), + }; + let session = match ensure_demo_ws_session(&app_handle, state, selected_client.clone()).await { + std::result::Result::Ok(session) => session, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let acknowledgement = match session.subscribe(standard_request).await { + std::result::Result::Ok(acknowledgement) => acknowledgement, + std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), + }; + let response_value = match acknowledgement.response.to_value() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), + }; + let response_json = match serde_json::to_string_pretty(&response_value) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), + }; + emit_demo_ws_status(&app_handle, session.snapshot().await); + return std::result::Result::Ok(crate::DemoWsExecutionPayload { + endpoint_name: selected_client.endpoint_name().to_string(), + provider: selected_client.provider().to_string(), + endpoint_url: selected_client.endpoint_url().to_string(), + role, + method: method.clone(), + request_kind: kb_onchain_transport::request_kind_from_method(&method), + response_kind: acknowledgement.response.kind_name().to_string(), + subscription_id: acknowledgement.subscription.remote_subscription_id, + response_json, + }); +} + +async fn ensure_demo_ws_session( + app_handle: &tauri::AppHandle, + state: &crate::AppState, + selected_client: kb_onchain_transport::WsClient, +) -> std::result::Result, std::string::String> { + let existing = { + let guard = state.demo_ws_session().lock().await; + guard.clone() + }; + if let std::option::Option::Some(session) = existing { + let snapshot = session.snapshot().await; + if snapshot.state != kb_onchain_transport::WsSessionState::Disconnected { + if snapshot.endpoint_url != selected_client.endpoint_url() { + return std::result::Result::Err(format!( + "demo WebSocket session already uses endpoint '{}'; disconnect before selecting '{}'", + snapshot.endpoint_name, + selected_client.endpoint_name() + )); + } + return std::result::Result::Ok(session); + } + } + let reconnect_policy = if selected_client.endpoint_config().auto_reconnect { + match kb_onchain_transport::WsReconnectPolicy::bounded( + DEMO_WS_RECONNECT_ATTEMPTS, + DEMO_WS_RECONNECT_INITIAL_DELAY_MS, + DEMO_WS_RECONNECT_MAX_DELAY_MS, + ) { + std::result::Result::Ok(policy) => policy, + std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), + } + } else { + kb_onchain_transport::WsReconnectPolicy::disabled() + }; + let capabilities = + kb_onchain_transport::StandardWsCapabilities::from_endpoint(selected_client.endpoint_config()); + let session = + match kb_onchain_transport::WsSession::connect(selected_client, capabilities, reconnect_policy).await { + std::result::Result::Ok(session) => std::sync::Arc::new(session), + std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), + }; + spawn_demo_ws_event_bridge(app_handle.clone(), session.clone()); + { + let mut guard = state.demo_ws_session().lock().await; + *guard = std::option::Option::Some(session.clone()); + } + return std::result::Result::Ok(session); +} + +pub(crate) async fn demo_ws_unsubscribe_inner( + app_handle: tauri::AppHandle, + state: &crate::AppState, + remote_subscription_id: u64, +) -> std::result::Result { + let session = { + let guard = state.demo_ws_session().lock().await; + guard.clone() + }; + let session = match session { + std::option::Option::Some(session) => session, + std::option::Option::None => { + return std::result::Result::Err("demo WebSocket session is not connected".to_string()); + }, + }; + let unsubscribe_result = session.unsubscribe(remote_subscription_id).await; + if let std::result::Result::Err(error) = unsubscribe_result { + return std::result::Result::Err(error.to_string()); + } + let snapshot = session.snapshot().await; + emit_demo_ws_status(&app_handle, snapshot.clone()); + return std::result::Result::Ok(status_from_snapshot(snapshot)); +} + +pub(crate) async fn demo_ws_disconnect_inner( + state: &crate::AppState, +) -> std::result::Result { + let session = { + let mut guard = state.demo_ws_session().lock().await; + guard.take() + }; + if let std::option::Option::Some(session) = session { + if let std::result::Result::Err(error) = session.disconnect().await { + return std::result::Result::Err(error.to_string()); + } + } + return std::result::Result::Ok(disconnected_status()); +} + +/// Disconnects the demo WebSocket session from application lifecycle hooks. +pub(crate) async fn disconnect_demo_ws_app_state(state: &crate::AppState, _wait_for_relay: bool) { + let disconnect_result = demo_ws_disconnect_inner(state).await; + if let std::result::Result::Err(error) = disconnect_result { + tracing::debug!(target: crate::TRACING_TARGET, "demo ws lifecycle disconnect failed: {error}"); + } +} + +fn spawn_demo_ws_event_bridge( + app_handle: tauri::AppHandle, + session: std::sync::Arc, +) { + let mut receiver = session.subscribe_events(); + tauri::async_runtime::spawn(async move { + let mut rate_limiter = DemoWsUiRateLimiter::new(); + loop { + let event = receiver.recv().await; + let event = match event { + std::result::Result::Ok(event) => event, + std::result::Result::Err(tokio::sync::broadcast::error::RecvError::Lagged( + count, + )) => { + emit_demo_ws_message( + &app_handle, + "lagged", + format!("{count} événement(s) WebSocket interne(s) perdu(s)"), + ); + continue; + }, + std::result::Result::Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + }; + match event { + kb_onchain_transport::WsSessionEvent::Notification { notification, .. } => { + if rate_limiter.should_emit(&app_handle) { + let payload = match serde_json::to_string_pretty(¬ification) { + std::result::Result::Ok(payload) => payload, + std::result::Result::Err(error) => { + format!("cannot serialize typed WebSocket notification: {error}") + }, + }; + emit_demo_ws_message( + &app_handle, + "notification", + truncate_payload(payload), + ); + } + }, + kb_onchain_transport::WsSessionEvent::Diagnostic { code, message } => { + emit_demo_ws_message(&app_handle, &code, truncate_payload(message)); + }, + kb_onchain_transport::WsSessionEvent::Reconnecting { attempt, maximum_attempts } => { + emit_demo_ws_message( + &app_handle, + "reconnecting", + format!("tentative {attempt}/{maximum_attempts}"), + ); + emit_demo_ws_status(&app_handle, session.snapshot().await); + }, + kb_onchain_transport::WsSessionEvent::Reconnected { reconnect_count } => { + emit_demo_ws_message( + &app_handle, + "reconnected", + format!("reconnexion numéro {reconnect_count}"), + ); + emit_demo_ws_status(&app_handle, session.snapshot().await); + }, + kb_onchain_transport::WsSessionEvent::Connected + | kb_onchain_transport::WsSessionEvent::SubscriptionAdded(_) + | kb_onchain_transport::WsSessionEvent::SubscriptionRemapped { .. } + | kb_onchain_transport::WsSessionEvent::SubscriptionRemoved(_) => { + emit_demo_ws_status(&app_handle, session.snapshot().await); + }, + kb_onchain_transport::WsSessionEvent::Disconnected => { + emit_demo_ws_status(&app_handle, session.snapshot().await); + break; + }, + } + } + }); +} + +fn emit_demo_ws_status(app_handle: &tauri::AppHandle, snapshot: kb_onchain_transport::WsSessionSnapshot) { + let window = match app_handle.get_webview_window("demo_ws") { + std::option::Option::Some(window) => window, + std::option::Option::None => return, + }; + let status = status_from_snapshot(snapshot); + if let std::result::Result::Err(error) = window.emit("demo-ws-status", status) { + tracing::debug!(target: crate::TRACING_TARGET, "cannot emit demo ws status: {error}"); + } +} + +fn emit_demo_ws_message( + app_handle: &tauri::AppHandle, + kind: &str, + payload_json: std::string::String, +) { + let window = match app_handle.get_webview_window("demo_ws") { + std::option::Option::Some(window) => window, + std::option::Option::None => return, + }; + if let std::result::Result::Err(error) = window.emit( + "demo-ws-message", + crate::DemoWsMessagePayload { kind: kind.to_string(), payload_json }, + ) { + tracing::debug!(target: crate::TRACING_TARGET, "cannot emit demo ws message: {error}"); + } +} + +fn status_from_snapshot(snapshot: kb_onchain_transport::WsSessionSnapshot) -> crate::DemoWsStatusPayload { + let mut subscriptions = std::vec::Vec::new(); + for subscription in &snapshot.subscriptions { + if let std::option::Option::Some(remote_subscription_id) = + subscription.remote_subscription_id + { + subscriptions.push(crate::DemoWsSubscriptionStatusPayload { + subscription_id: remote_subscription_id, + method: subscription.subscribe_method.clone(), + unsubscribe_method: subscription.unsubscribe_method.clone(), + }); + } + } + let (method, unsubscribe_method, subscription_id) = match subscriptions.last() { + std::option::Option::Some(subscription) => ( + std::option::Option::Some(subscription.method.clone()), + std::option::Option::Some(subscription.unsubscribe_method.clone()), + std::option::Option::Some(subscription.subscription_id), + ), + std::option::Option::None => { + (std::option::Option::None, std::option::Option::None, std::option::Option::None) + }, + }; + return crate::DemoWsStatusPayload { + connected: snapshot.state != kb_onchain_transport::WsSessionState::Disconnected, + endpoint_name: std::option::Option::Some(snapshot.endpoint_name), + endpoint_url: std::option::Option::Some(snapshot.endpoint_url), + method, + unsubscribe_method, + subscription_id, + subscription_count: match u32::try_from(subscriptions.len()) { + std::result::Result::Ok(count) => count, + std::result::Result::Err(_) => u32::MAX, + }, + subscriptions, + }; +} + +fn disconnected_status() -> crate::DemoWsStatusPayload { + return crate::DemoWsStatusPayload { + connected: false, + endpoint_name: std::option::Option::None, + endpoint_url: std::option::Option::None, + method: std::option::Option::None, + unsubscribe_method: std::option::Option::None, + subscription_id: std::option::Option::None, + subscription_count: 0, + subscriptions: std::vec::Vec::new(), + }; +} + +pub(crate) fn build_ws_role_options( + snapshots: std::vec::Vec, +) -> std::vec::Vec { + let mut roles = std::collections::BTreeMap::< + std::string::String, + std::collections::BTreeSet, + >::new(); + for snapshot in snapshots { + for role in snapshot.roles { + if !role.enabled { + continue; + } + let role_entry = roles.entry(role.role).or_default(); + for request_kind in role.request_kinds { + role_entry.insert(request_kind); + } + } + } + let mut options = std::vec::Vec::new(); + for (role, request_kinds) in roles { + options.push(crate::DemoWsRoleOption { + role, + request_kinds: request_kinds.into_iter().collect(), + }); + } + return options; +} + +pub(crate) fn build_ws_method_options() -> std::vec::Vec { + let labels = std::collections::BTreeMap::from([ + ("accountSubscribe", ("Compte", true, false, true)), + ("blockSubscribe", ("Bloc", false, true, true)), + ("logsSubscribe", ("Logs", false, true, true)), + ("programSubscribe", ("Programme", true, false, true)), + ("rootSubscribe", ("Roots", false, false, false)), + ("signatureSubscribe", ("Signature", true, false, true)), + ("slotSubscribe", ("Slots", false, false, false)), + ("slotsUpdatesSubscribe", ("Mises à jour slots", false, false, false)), + ("voteSubscribe", ("Votes", false, false, false)), + ]); + let mut options = std::vec::Vec::new(); + for specification in &kb_onchain_transport::STANDARD_WS_SUBSCRIPTIONS { + let metadata = labels.get(specification.subscribe_method); + let (label, requires_target, requires_filter_json, supports_config_json) = match metadata { + std::option::Option::Some(metadata) => *metadata, + std::option::Option::None => continue, + }; + options.push(crate::DemoWsMethodOption { + method: specification.subscribe_method.to_string(), + unsubscribe_method: specification.unsubscribe_method.to_string(), + request_kind: kb_onchain_transport::request_kind_from_method(specification.subscribe_method), + label: label.to_string(), + requires_target, + requires_filter_json, + supports_config_json, + }); + } + return options; +} + +fn build_standard_ws_request( + method: &str, + target: std::option::Option, + filter_json: std::option::Option, + config_json: std::option::Option, +) -> std::result::Result { + return match method { + "accountSubscribe" => { + let target = match required_target(target, method) { + std::result::Result::Ok(target) => target, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let config = match parse_optional_json_as::( + config_json, + "configJson", + ) { + std::result::Result::Ok(config) => config, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if let std::result::Result::Err(error) = + reject_present(filter_json, "filterJson", method) + { + return std::result::Result::Err(error); + } + std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Account( + kb_onchain_transport::AccountSubscribeRequest { pubkey: target, config }, + )) + }, + "blockSubscribe" => { + if let std::result::Result::Err(error) = reject_present(target, "target", method) { + return std::result::Result::Err(error); + } + let filter = + match parse_required_json_as::(filter_json, "filterJson") { + std::result::Result::Ok(filter) => filter, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let config = match parse_optional_json_as::( + config_json, + "configJson", + ) { + std::result::Result::Ok(config) => config, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Block( + kb_onchain_transport::BlockSubscribeRequest { filter, config }, + )) + }, + "logsSubscribe" => { + if let std::result::Result::Err(error) = reject_present(target, "target", method) { + return std::result::Result::Err(error); + } + let filter = + match parse_required_json_as::(filter_json, "filterJson") { + std::result::Result::Ok(filter) => filter, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let config = match parse_optional_json_as::( + config_json, + "configJson", + ) { + std::result::Result::Ok(config) => config, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Logs(kb_onchain_transport::LogsSubscribeRequest { + filter, + config, + })) + }, + "programSubscribe" => { + let target = match required_target(target, method) { + std::result::Result::Ok(target) => target, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if let std::result::Result::Err(error) = + reject_present(filter_json, "filterJson", method) + { + return std::result::Result::Err(error); + } + let config = match parse_optional_json_as::( + config_json, + "configJson", + ) { + std::result::Result::Ok(config) => config, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Program( + kb_onchain_transport::ProgramSubscribeRequest { program_id: target, config }, + )) + }, + "signatureSubscribe" => { + let target = match required_target(target, method) { + std::result::Result::Ok(target) => target, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if let std::result::Result::Err(error) = + reject_present(filter_json, "filterJson", method) + { + return std::result::Result::Err(error); + } + let config = match parse_optional_json_as::( + config_json, + "configJson", + ) { + std::result::Result::Ok(config) => config, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Signature( + kb_onchain_transport::SignatureSubscribeRequest { signature: target, config }, + )) + }, + "rootSubscribe" => { + if let std::result::Result::Err(error) = + reject_parameterless_inputs(target, filter_json, config_json, method) + { + return std::result::Result::Err(error); + } + std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Root(kb_onchain_transport::RootSubscribeRequest)) + }, + "slotSubscribe" => { + if let std::result::Result::Err(error) = + reject_parameterless_inputs(target, filter_json, config_json, method) + { + return std::result::Result::Err(error); + } + std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Slot(kb_onchain_transport::SlotSubscribeRequest)) + }, + "slotsUpdatesSubscribe" => { + if let std::result::Result::Err(error) = + reject_parameterless_inputs(target, filter_json, config_json, method) + { + return std::result::Result::Err(error); + } + std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::SlotsUpdates( + kb_onchain_transport::SlotsUpdatesSubscribeRequest, + )) + }, + "voteSubscribe" => { + if let std::result::Result::Err(error) = + reject_parameterless_inputs(target, filter_json, config_json, method) + { + return std::result::Result::Err(error); + } + std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Vote(kb_onchain_transport::VoteSubscribeRequest)) + }, + _ => std::result::Result::Err(format!("unsupported standard WebSocket method '{method}'")), + }; +} + +fn required_target( + target: std::option::Option, + method: &str, +) -> std::result::Result { + let target = match target { + std::option::Option::Some(target) => target.trim().to_string(), + std::option::Option::None => std::string::String::new(), + }; + if target.is_empty() { + return std::result::Result::Err(format!("method '{method}' requires target")); + } + return std::result::Result::Ok(target); +} + +fn reject_parameterless_inputs( + target: std::option::Option, + filter_json: std::option::Option, + config_json: std::option::Option, + method: &str, +) -> std::result::Result<(), std::string::String> { + if let std::result::Result::Err(error) = reject_present(target, "target", method) { + return std::result::Result::Err(error); + } + if let std::result::Result::Err(error) = reject_present(filter_json, "filterJson", method) { + return std::result::Result::Err(error); + } + if let std::result::Result::Err(error) = reject_present(config_json, "configJson", method) { + return std::result::Result::Err(error); + } + return std::result::Result::Ok(()); +} + +fn reject_present( + value: std::option::Option, + field: &str, + method: &str, +) -> std::result::Result<(), std::string::String> { + if let std::option::Option::Some(value) = value { + if !value.trim().is_empty() { + return std::result::Result::Err(format!("method '{method}' does not accept {field}")); + } + } + return std::result::Result::Ok(()); +} + +fn parse_required_json_as( + text: std::option::Option, + field: &str, +) -> std::result::Result { + let text = match text { + std::option::Option::Some(text) if !text.trim().is_empty() => text, + _ => return std::result::Result::Err(format!("{field} is required")), + }; + return match serde_json::from_str::(&text) { + std::result::Result::Ok(value) => std::result::Result::Ok(value), + std::result::Result::Err(error) => { + std::result::Result::Err(format!("invalid {field}: {error}")) + }, + }; +} + +fn parse_optional_json_as( + text: std::option::Option, + field: &str, +) -> std::result::Result, std::string::String> { + let text = match text { + std::option::Option::Some(text) if !text.trim().is_empty() => text, + _ => return std::result::Result::Ok(std::option::Option::None), + }; + return match serde_json::from_str::(&text) { + std::result::Result::Ok(value) => std::result::Result::Ok(std::option::Option::Some(value)), + std::result::Result::Err(error) => { + std::result::Result::Err(format!("invalid {field}: {error}")) + }, + }; +} + +fn truncate_payload(payload: std::string::String) -> std::string::String { + if payload.chars().count() <= DEMO_WS_UI_MAX_PAYLOAD_CHARS { + return payload; + } + let mut truncated = payload + .chars() + .take(DEMO_WS_UI_MAX_PAYLOAD_CHARS) + .collect::(); + truncated.push_str("\n… payload tronqué côté Rust …"); + return truncated; +} + +#[cfg(test)] +mod tests { + use ts_rs::TS; // rust-rules: derive-import + + #[test] + fn tauri_numeric_bindings_use_json_compatible_numbers() { + let config = ts_rs::Config::default(); + let declarations = [ + ::decl(&config), + ::decl(&config), + ::decl(&config), + ::decl(&config), + ::decl(&config), + ::decl(&config), + ::decl(&config), + ]; + for declaration in declarations { + assert!(!declaration.contains("bigint")); + } + } +} diff --git a/kb-app-demo-desktop/src/frontend_log.rs b/kb-app-demo-desktop/src/frontend_log.rs index 2a81881..cded103 100644 --- a/kb-app-demo-desktop/src/frontend_log.rs +++ b/kb-app-demo-desktop/src/frontend_log.rs @@ -1,5 +1,5 @@ // file: kb-app-demo-desktop/src/frontend_log.rs -// version: 2 +// version: 3 //! Frontend logging bridge used by Tauri WebView scripts. @@ -23,21 +23,11 @@ pub(crate) struct FrontendLogPayload { pub(crate) fn emit_frontend_log(payload: crate::FrontendLogPayload) { let target = normalize_frontend_target(payload.target.as_str()); match payload.level.trim().to_ascii_lowercase().as_str() { - "trace" => { - tracing::trace!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message) - }, - "debug" => { - tracing::debug!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message) - }, - "warn" => { - tracing::warn!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message) - }, - "error" => { - tracing::error!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message) - }, - _ => { - tracing::info!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message) - }, + "trace" => tracing::trace!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message), + "debug" => tracing::debug!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message), + "warn" => tracing::warn!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message), + "error" => tracing::error!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message), + _ => tracing::info!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message), } } @@ -55,6 +45,9 @@ fn normalize_frontend_target(target: &str) -> std::string::String { if trimmed == "kb-app-demo-desktop.frontend.demo_http" { return trimmed.to_string(); } + if trimmed == "kb-app-demo-desktop.frontend.demo_ws" { + return trimmed.to_string(); + } return "kb-app-demo-desktop.frontend".to_string(); } @@ -85,5 +78,10 @@ mod tests { super::normalize_frontend_target("kb-app-demo-desktop.frontend.demo_http"), "kb-app-demo-desktop.frontend.demo_http" ); + assert_eq!( + super::normalize_frontend_target("kb-app-demo-desktop.frontend.demo_ws"), + "kb-app-demo-desktop.frontend.demo_ws" + ); } + } diff --git a/kb-app-demo-desktop/src/lib.rs b/kb-app-demo-desktop/src/lib.rs index d0ec96e..0588153 100644 --- a/kb-app-demo-desktop/src/lib.rs +++ b/kb-app-demo-desktop/src/lib.rs @@ -11,6 +11,7 @@ mod app_state; mod constants; mod demo_backfill; mod demo_http; +mod demo_ws; mod frontend_log; mod main_window; mod splash; @@ -57,6 +58,38 @@ pub(crate) use self::demo_http::build_http_method_options; pub(crate) use self::demo_http::build_http_role_options; /// Executes one raw HTTP JSON-RPC request. pub(crate) use self::demo_http::demo_http_execute_request_inner; +/// WebSocket execution response payload. +pub(crate) use self::demo_ws::DemoWsExecutionPayload; +/// WebSocket message emitted to the frontend. +pub(crate) use self::demo_ws::DemoWsMessagePayload; +/// One selectable WebSocket method. +pub(crate) use self::demo_ws::DemoWsMethodOption; +/// WebSocket demo options payload. +pub(crate) use self::demo_ws::DemoWsOptionsPayload; +/// WebSocket subscription request. +pub(crate) use self::demo_ws::DemoWsRequest; +/// One selectable WebSocket role. +pub(crate) use self::demo_ws::DemoWsRoleOption; +/// Current WebSocket session status. +pub(crate) use self::demo_ws::DemoWsStatusPayload; +/// One active WebSocket subscription status. +pub(crate) use self::demo_ws::DemoWsSubscriptionStatusPayload; +/// WebSocket unsubscribe request. +pub(crate) use self::demo_ws::DemoWsUnsubscribeRequest; +/// Builds the standard WebSocket method inventory. +pub(crate) use self::demo_ws::build_ws_method_options; +/// Builds selectable WebSocket roles from pool snapshots. +pub(crate) use self::demo_ws::build_ws_role_options; +/// Connects and subscribes through the persistent WebSocket session. +pub(crate) use self::demo_ws::demo_ws_connect_inner; +/// Disconnects the persistent WebSocket session. +pub(crate) use self::demo_ws::demo_ws_disconnect_inner; +/// Returns the current WebSocket session status. +pub(crate) use self::demo_ws::demo_ws_status_inner; +/// Unsubscribes one active WebSocket subscription. +pub(crate) use self::demo_ws::demo_ws_unsubscribe_inner; +/// Disconnects the application WebSocket session during shutdown. +pub(crate) use self::demo_ws::disconnect_demo_ws_app_state; /// Frontend logging payload. pub(crate) use self::frontend_log::FrontendLogPayload; /// Emits one normalized frontend log event. diff --git a/kb-app-demo-desktop/src/tauri.rs b/kb-app-demo-desktop/src/tauri.rs index 6e094fb..cb0a973 100644 --- a/kb-app-demo-desktop/src/tauri.rs +++ b/kb-app-demo-desktop/src/tauri.rs @@ -1,5 +1,5 @@ // file: kb-app-demo-desktop/src/tauri.rs -// version: 5 +// version: 6 //! Tauri runtime assembly and private command wrappers. @@ -45,6 +45,13 @@ pub fn run() -> kb_core::Result<()> { demo_http_list_pool_clients, demo_http_options, demo_http_execute_request, + open_demo_ws_window, + demo_ws_list_pool_clients, + demo_ws_options, + demo_ws_status, + demo_ws_connect, + demo_ws_unsubscribe, + demo_ws_disconnect, ]); builder = builder.plugin(tracing_builder.build::()); builder = builder.setup(|app| { @@ -89,6 +96,13 @@ pub fn run() -> kb_core::Result<()> { std::option::Option::None, std::option::Option::None, ); + crate::emit_splash_order( + &splash_window, + "add_log", + std::option::Option::Some("Pool WebSocket initialisé"), + std::option::Option::None, + std::option::Option::None, + ); crate::emit_splash_order( &splash_window, "fadein", @@ -143,9 +157,9 @@ pub fn run() -> kb_core::Result<()> { let run_result = builder.run(tauri::generate_context!()); return match run_result { std::result::Result::Ok(()) => std::result::Result::Ok(()), - std::result::Result::Err(error) => std::result::Result::Err(kb_core::Error::tauri( - format!("cannot run desktop demo application: {error:?}"), - )), + std::result::Result::Err(error) => std::result::Result::Err(kb_core::Error::tauri(format!( + "cannot run desktop demo application: {error:?}" + ))), }; } @@ -162,7 +176,9 @@ fn install_default_rustls_provider() -> kb_core::Result<()> { }; } -fn into_ipc_result(result: kb_core::Result) -> std::result::Result { +fn into_ipc_result( + result: kb_core::Result, +) -> std::result::Result { return match result { std::result::Result::Ok(value) => std::result::Result::Ok(value), std::result::Result::Err(error) => std::result::Result::Err(error.to_string()), @@ -181,7 +197,9 @@ fn load_project_readme() -> std::result::Result) -> bool { - let running = state.demo_backfill_running().load(std::sync::atomic::Ordering::Acquire); + let running = state + .demo_backfill_running() + .load(std::sync::atomic::Ordering::Acquire); state .demo_backfill_cancel_requested() .store(true, std::sync::atomic::Ordering::Release); @@ -240,12 +258,14 @@ fn demo_backfill_options( default_max_pages: 20, default_max_concurrent_requests: 4, default_max_retries: 2, - running: state.demo_backfill_running().load(std::sync::atomic::Ordering::Acquire), + running: state + .demo_backfill_running() + .load(std::sync::atomic::Ordering::Acquire), }; } -#[allow(clippy::question_mark_used)] #[tauri::command] +#[allow(clippy::question_mark_used)] async fn demo_backfill_execute( app_handle: tauri::AppHandle, state: tauri::State<'_, crate::AppState>, @@ -260,7 +280,9 @@ async fn demo_backfill_execute( if acquire_result.is_err() { return std::result::Result::Err("a backfill campaign is already running".to_string()); } - let _run_guard = crate::DemoBackfillRunGuard { running: state.demo_backfill_running() }; + let _run_guard = crate::DemoBackfillRunGuard { + running: state.demo_backfill_running(), + }; state .demo_backfill_cancel_requested() .store(false, std::sync::atomic::Ordering::Release); @@ -345,18 +367,124 @@ fn demo_http_list_pool_clients( } #[tauri::command] -fn demo_http_options(state: tauri::State<'_, crate::AppState>) -> crate::DemoHttpOptionsPayload { +fn demo_http_options( + state: tauri::State<'_, crate::AppState>, +) -> crate::DemoHttpOptionsPayload { return crate::DemoHttpOptionsPayload { roles: crate::build_http_role_options(state.http_pool().snapshot()), methods: crate::build_http_method_options(), }; } -#[allow(clippy::question_mark_used)] #[tauri::command] +#[allow(clippy::question_mark_used)] async fn demo_http_execute_request( state: tauri::State<'_, crate::AppState>, request: crate::DemoHttpRequest, ) -> std::result::Result { return crate::demo_http_execute_request_inner(state, request).await; } + +/// Opens or focuses the WebSocket demo window. +#[tauri::command] +fn open_demo_ws_window( + app_handle: tauri::AppHandle, +) -> std::result::Result<(), std::string::String> { + tracing::info!(target: crate::TRACING_TARGET, "open_demo_ws_window"); + if let std::option::Option::Some(window) = app_handle.get_webview_window("demo_ws") { + if let std::result::Result::Err(error) = window.show() { + return std::result::Result::Err( + kb_core::Error::tauri(format!("cannot show demo_ws window: {error}")).to_string(), + ); + } + if let std::result::Result::Err(error) = window.set_focus() { + return std::result::Result::Err( + kb_core::Error::tauri(format!("cannot focus demo_ws window: {error}")).to_string(), + ); + } + return std::result::Result::Ok(()); + } + let build_result = tauri::WebviewWindowBuilder::new( + &app_handle, + "demo_ws", + tauri::WebviewUrl::App("demo_ws.html".into()), + ) + .title("Khadhroony Bot3 - WebSocket standard") + .inner_size(1200.0, 760.0) + .min_inner_size(920.0, 560.0) + .resizable(true) + .visible(true) + .build(); + return match build_result { + std::result::Result::Ok(window) => match window.set_focus() { + std::result::Result::Ok(()) => std::result::Result::Ok(()), + std::result::Result::Err(error) => std::result::Result::Err( + kb_core::Error::tauri(format!("cannot focus created demo_ws window: {error}")) + .to_string(), + ), + }, + std::result::Result::Err(error) => std::result::Result::Err( + kb_core::Error::tauri(format!("cannot create demo_ws window: {error}")).to_string(), + ), + }; +} + +/// Lists WebSocket endpoints available through the configured pool. +#[tauri::command] +fn demo_ws_list_pool_clients( + state: tauri::State<'_, crate::AppState>, +) -> std::result::Result, std::string::String> { + return std::result::Result::Ok(state.ws_pool().snapshot()); +} + +/// Lists selectable WebSocket roles and methods for the demo UI. +#[tauri::command] +fn demo_ws_options( + state: tauri::State<'_, crate::AppState>, +) -> std::result::Result { + return std::result::Result::Ok(crate::DemoWsOptionsPayload { + roles: crate::build_ws_role_options(state.ws_pool().snapshot()), + methods: crate::build_ws_method_options(), + }); +} + +/// Returns the current WebSocket demo session status. +#[tauri::command] +#[allow(clippy::question_mark_used)] +async fn demo_ws_status( + state: tauri::State<'_, crate::AppState>, +) -> std::result::Result { + return crate::demo_ws_status_inner(state.inner()).await; +} + +/// Connects if needed, then subscribes through `kb_onchain_transport::WsSession`. +#[tauri::command] +#[allow(clippy::question_mark_used)] +async fn demo_ws_connect( + app_handle: tauri::AppHandle, + state: tauri::State<'_, crate::AppState>, + request: crate::DemoWsRequest, +) -> std::result::Result { + return crate::demo_ws_connect_inner(app_handle, state.inner(), request).await; +} + +/// Unsubscribes one subscription while keeping the WebSocket connection open. +#[tauri::command] +#[allow(clippy::question_mark_used)] +async fn demo_ws_unsubscribe( + app_handle: tauri::AppHandle, + state: tauri::State<'_, crate::AppState>, + request: crate::DemoWsUnsubscribeRequest, +) -> std::result::Result { + return crate::demo_ws_unsubscribe_inner(app_handle, state.inner(), request.subscription_id) + .await; +} + +/// Disconnects the current persistent WebSocket demo session. +#[tauri::command] +#[allow(clippy::question_mark_used)] +async fn demo_ws_disconnect( + state: tauri::State<'_, crate::AppState>, +) -> std::result::Result { + return crate::demo_ws_disconnect_inner(state.inner()).await; +}