// file: crates/ksp-onchain-transport-lib/unit_tests/ws_helius_transactions.rs // version: 2 use futures_util::SinkExt; // rust-rules: trait-import use futures_util::StreamExt; // rust-rules: trait-import fn pubkey(value: &str) -> ksp_core_lib::Pubkey { return value.parse::().expect("fixture public key must parse"); } fn base_filter() -> crate::HeliusTransactionSubscribeFilter { return crate::HeliusTransactionSubscribeFilter::new( std::option::Option::Some(false), std::option::Option::Some(false), std::option::Option::Some("fixture-signature-secret-canary".to_owned()), std::option::Option::Some(std::vec![pubkey("11111111111111111111111111111111")]), std::option::Option::Some(std::vec![pubkey("SysvarC1ock11111111111111111111111111111111")]), std::option::Option::Some(std::vec![pubkey("Vote111111111111111111111111111111111111111")]), std::option::Option::Some(crate::HeliusTokenAccountsFilter::BalanceChanged), ); } fn assert_oversized_filter_rejected(filter: crate::HeliusTransactionSubscribeFilter) { let request = crate::HeliusTransactionSubscribeRequest::new(filter, std::option::Option::None); let error = request.validate().expect_err("50,001 Helius account filters must fail before I/O"); assert_eq!(error.code(), crate::ERROR_CODE_INVALID_RPC_PARAMETERS); assert!(error.to_string().contains("invalid_rpc_parameters")); assert!(!error.to_string().contains("11111111111111111111111111111111")); } fn helius_endpoint(url: &str) -> crate::WsEndpointSettings { return crate::WsEndpointSettings::new( "local_helius_transaction_fixture", true, crate::WsProviderName::new("helius"), crate::WsClusterName::new("local"), crate::WsProtocolKind::HeliusLaserStream, crate::WsEndpointUrl::parse(url).expect("local Helius WebSocket URL must parse"), crate::WsSessionSettings::default(), ); } async fn bind_local_listener() -> (tokio::net::TcpListener, std::string::String) { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("local listener must bind"); let address = listener.local_addr().expect("local listener must expose address"); return (listener, format!("ws://{address}")); } async fn read_request(websocket: &mut tokio_tungstenite::WebSocketStream) -> serde_json::Value { let message = websocket.next().await.expect("request message must exist").expect("request message must decode"); let text = message.to_text().expect("request must be text"); return serde_json::from_str(text).expect("request must contain JSON"); } async fn send_result(websocket: &mut tokio_tungstenite::WebSocketStream, request: &serde_json::Value, result: serde_json::Value) { let id = request.get("id").and_then(serde_json::Value::as_u64).expect("request id must be numeric"); let response = serde_json::json!({"jsonrpc":"2.0","id":id,"result":result}); websocket.send(tokio_tungstenite::tungstenite::Message::Text(response.to_string().into())).await.expect("local response must send"); return; } async fn wait_for_close_frame(websocket: &mut tokio_tungstenite::WebSocketStream) { loop { let message = websocket.next().await; match message { std::option::Option::Some(std::result::Result::Ok(tokio_tungstenite::tungstenite::Message::Close(_))) => return, std::option::Option::Some(std::result::Result::Ok(_)) => {}, std::option::Option::Some(std::result::Result::Err(_)) | std::option::Option::None => return, } } } #[test] fn helius_transaction_filter_and_option_enums_match_documented_wire_labels() { assert_eq!(crate::HeliusTokenAccountsFilter::None.as_str(), "none"); assert_eq!(crate::HeliusTokenAccountsFilter::BalanceChanged.as_str(), "balanceChanged"); assert_eq!(crate::HeliusTokenAccountsFilter::All.as_str(), "all"); assert_eq!(crate::HeliusTransactionSubscribeEncoding::Base58.as_str(), "base58"); assert_eq!(crate::HeliusTransactionSubscribeEncoding::Base64.as_str(), "base64"); assert_eq!(crate::HeliusTransactionSubscribeEncoding::JsonParsed.as_str(), "jsonParsed"); } #[test] fn helius_transaction_subscribe_request_serializes_complete_documented_filter_and_options() { let filter = base_filter(); let options = crate::HeliusTransactionSubscribeOptions::new( std::option::Option::Some(crate::SolanaCommitment::Confirmed), std::option::Option::Some(crate::HeliusTransactionSubscribeEncoding::JsonParsed), std::option::Option::Some(crate::SolanaTransactionDetails::Accounts), std::option::Option::Some(true), std::option::Option::Some(0), ); let request = crate::HeliusTransactionSubscribeRequest::new(filter, std::option::Option::Some(options)); let params = request.to_params().expect("complete documented Helius request must validate"); assert_eq!( params, serde_json::json!([ { "vote": false, "failed": false, "signature": "fixture-signature-secret-canary", "accountInclude": ["11111111111111111111111111111111"], "accountExclude": ["SysvarC1ock11111111111111111111111111111111"], "accountRequired": ["Vote111111111111111111111111111111111111111"], "tokenAccounts": "balanceChanged" }, { "commitment": "confirmed", "encoding": "jsonParsed", "transactionDetails": "accounts", "showRewards": true, "maxSupportedTransactionVersion": 0 } ]) ); assert_eq!(request.filter().vote(), std::option::Option::Some(false)); assert_eq!(request.filter().failed(), std::option::Option::Some(false)); assert_eq!(request.filter().signature(), std::option::Option::Some("fixture-signature-secret-canary")); assert_eq!(request.filter().account_include().map(<[ksp_core_lib::Pubkey]>::len), std::option::Option::Some(1)); assert_eq!(request.filter().account_exclude().map(<[ksp_core_lib::Pubkey]>::len), std::option::Option::Some(1)); assert_eq!(request.filter().account_required().map(<[ksp_core_lib::Pubkey]>::len), std::option::Option::Some(1)); assert_eq!(request.filter().token_accounts(), std::option::Option::Some(crate::HeliusTokenAccountsFilter::BalanceChanged)); let options = request.options().expect("options must remain available"); assert_eq!(options.commitment(), std::option::Option::Some(crate::SolanaCommitment::Confirmed)); assert_eq!(options.encoding(), std::option::Option::Some(crate::HeliusTransactionSubscribeEncoding::JsonParsed)); assert_eq!(options.transaction_details(), std::option::Option::Some(crate::SolanaTransactionDetails::Accounts)); assert_eq!(options.show_rewards(), std::option::Option::Some(true)); assert_eq!(options.max_supported_transaction_version(), std::option::Option::Some(0)); } #[test] fn helius_transaction_request_preserves_omitted_explicit_empty_and_explicit_none_states() { let omitted = crate::HeliusTransactionSubscribeRequest::new(crate::HeliusTransactionSubscribeFilter::default(), std::option::Option::None); assert_eq!(omitted.to_params().expect("fully omitted optional request must validate"), serde_json::json!([{}])); let explicit = crate::HeliusTransactionSubscribeRequest::new( crate::HeliusTransactionSubscribeFilter::new( std::option::Option::None, std::option::Option::None, std::option::Option::None, std::option::Option::Some(std::vec::Vec::new()), std::option::Option::Some(std::vec::Vec::new()), std::option::Option::Some(std::vec::Vec::new()), std::option::Option::Some(crate::HeliusTokenAccountsFilter::None), ), std::option::Option::Some(crate::HeliusTransactionSubscribeOptions::default()), ); assert_eq!( explicit.to_params().expect("explicit empty Helius request states must validate"), serde_json::json!([{"accountInclude":[],"accountExclude":[],"accountRequired":[],"tokenAccounts":"none"}, {}]) ); } #[test] fn helius_transaction_filter_enforces_each_documented_fifty_thousand_account_bound() { let key = pubkey("11111111111111111111111111111111"); let maximum = std::vec![key; 50_000]; let accepted = crate::HeliusTransactionSubscribeRequest::new( crate::HeliusTransactionSubscribeFilter::new( std::option::Option::None, std::option::Option::None, std::option::Option::None, std::option::Option::Some(maximum), std::option::Option::None, std::option::Option::None, std::option::Option::None, ), std::option::Option::None, ); assert!(accepted.validate().is_ok()); assert_oversized_filter_rejected(crate::HeliusTransactionSubscribeFilter::new( std::option::Option::None, std::option::Option::None, std::option::Option::None, std::option::Option::Some(std::vec![key; 50_001]), std::option::Option::None, std::option::Option::None, std::option::Option::None, )); assert_oversized_filter_rejected(crate::HeliusTransactionSubscribeFilter::new( std::option::Option::None, std::option::Option::None, std::option::Option::None, std::option::Option::None, std::option::Option::Some(std::vec![key; 50_001]), std::option::Option::None, std::option::Option::None, )); assert_oversized_filter_rejected(crate::HeliusTransactionSubscribeFilter::new( std::option::Option::None, std::option::Option::None, std::option::Option::None, std::option::Option::None, std::option::Option::None, std::option::Option::Some(std::vec![key; 50_001]), std::option::Option::None, )); } #[test] fn helius_transaction_details_require_max_supported_version_only_for_accounts_and_full() { for details in [crate::SolanaTransactionDetails::Full, crate::SolanaTransactionDetails::Accounts] { let options = crate::HeliusTransactionSubscribeOptions::new( std::option::Option::None, std::option::Option::None, std::option::Option::Some(details), std::option::Option::None, std::option::Option::None, ); let request = crate::HeliusTransactionSubscribeRequest::new(crate::HeliusTransactionSubscribeFilter::default(), std::option::Option::Some(options)); let error = request.validate().expect_err("full/accounts details must require maxSupportedTransactionVersion"); assert_eq!(error.code(), crate::ERROR_CODE_INVALID_RPC_PARAMETERS); } for details in [crate::SolanaTransactionDetails::Signatures, crate::SolanaTransactionDetails::None] { let options = crate::HeliusTransactionSubscribeOptions::new( std::option::Option::None, std::option::Option::None, std::option::Option::Some(details), std::option::Option::None, std::option::Option::None, ); let request = crate::HeliusTransactionSubscribeRequest::new(crate::HeliusTransactionSubscribeFilter::default(), std::option::Option::Some(options)); assert!(request.validate().is_ok()); } let full_with_version = crate::HeliusTransactionSubscribeOptions::new( std::option::Option::None, std::option::Option::None, std::option::Option::Some(crate::SolanaTransactionDetails::Full), std::option::Option::None, std::option::Option::Some(0), ); let request = crate::HeliusTransactionSubscribeRequest::new(crate::HeliusTransactionSubscribeFilter::default(), std::option::Option::Some(full_with_version)); assert!(request.validate().is_ok()); } #[test] fn helius_transaction_subscribe_and_unsubscribe_control_wire_is_exact() { assert_eq!(crate::helius_transaction_subscribe_method(), "transactionSubscribe"); assert_eq!(crate::helius_transaction_unsubscribe_method(), "transactionUnsubscribe"); assert_eq!( crate::decode_helius_transaction_subscribe_result(serde_json::json!(4_743_323_479_349_712_u64)).expect("numeric ack must decode"), 4_743_323_479_349_712 ); assert_eq!(crate::helius_transaction_unsubscribe_params(4_743_323_479_349_712), serde_json::json!([4_743_323_479_349_712_u64])); assert!(crate::decode_helius_transaction_unsubscribe_result(serde_json::json!(true)).expect("boolean true must decode")); assert!(!crate::decode_helius_transaction_unsubscribe_result(serde_json::json!(false)).expect("boolean false must decode")); assert_eq!( crate::decode_helius_transaction_subscribe_result(serde_json::json!("not-an-id")).expect_err("non-numeric subscribe ack must fail").code(), crate::ERROR_CODE_INVALID_RESPONSE ); assert_eq!( crate::decode_helius_transaction_unsubscribe_result(serde_json::json!(1)).expect_err("non-boolean unsubscribe ack must fail").code(), crate::ERROR_CODE_INVALID_RESPONSE ); } #[test] fn helius_transaction_filter_debug_omits_signature_and_account_values() { let filter = base_filter(); let request = crate::HeliusTransactionSubscribeRequest::new(filter, std::option::Option::None); let debug = format!("{request:?}"); assert!(debug.contains("signature_present")); assert!(debug.contains("account_include_count")); assert!(!debug.contains("fixture-signature-secret-canary")); assert!(!debug.contains("11111111111111111111111111111111")); assert!(!debug.contains("SysvarC1ock11111111111111111111111111111111")); assert!(!debug.contains("Vote111111111111111111111111111111111111111")); } #[tokio::test(flavor = "current_thread")] async fn helius_transaction_control_wire_round_trips_through_shared_physical_actor() { let (listener, url) = bind_local_listener().await; let server = tokio::spawn(async move { let (stream, _) = listener.accept().await.expect("local server must accept client"); let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed"); let subscribe = read_request(&mut websocket).await; assert_eq!(subscribe["method"], serde_json::json!("transactionSubscribe")); assert_eq!( subscribe["params"], serde_json::json!([ {"failed":false,"accountInclude":["11111111111111111111111111111111"],"tokenAccounts":"balanceChanged"}, {"commitment":"confirmed","encoding":"jsonParsed","transactionDetails":"full","showRewards":false,"maxSupportedTransactionVersion":0} ]) ); send_result(&mut websocket, &subscribe, serde_json::json!(4242)).await; let unsubscribe = read_request(&mut websocket).await; assert_eq!(unsubscribe["method"], serde_json::json!("transactionUnsubscribe")); assert_eq!(unsubscribe["params"], serde_json::json!([4242])); send_result(&mut websocket, &unsubscribe, serde_json::json!(true)).await; wait_for_close_frame(&mut websocket).await; }); let session = crate::HeliusLaserStreamWsSession::connect(helius_endpoint(url.as_str())).await.expect("Helius facade must connect"); let filter = crate::HeliusTransactionSubscribeFilter::new( std::option::Option::None, std::option::Option::Some(false), std::option::Option::None, std::option::Option::Some(std::vec![pubkey("11111111111111111111111111111111")]), std::option::Option::None, std::option::Option::None, std::option::Option::Some(crate::HeliusTokenAccountsFilter::BalanceChanged), ); let options = crate::HeliusTransactionSubscribeOptions::new( std::option::Option::Some(crate::SolanaCommitment::Confirmed), std::option::Option::Some(crate::HeliusTransactionSubscribeEncoding::JsonParsed), std::option::Option::Some(crate::SolanaTransactionDetails::Full), std::option::Option::Some(false), std::option::Option::Some(0), ); let request = crate::HeliusTransactionSubscribeRequest::new(filter, std::option::Option::Some(options)); let params = request.to_params().expect("typed Helius request must validate before I/O"); let subscribe_result = session .physical_session() .execute_json_rpc(crate::helius_transaction_subscribe_method(), params) .await .expect("transactionSubscribe acknowledgement must arrive"); let remote_id = crate::decode_helius_transaction_subscribe_result(subscribe_result).expect("transactionSubscribe id must decode"); assert_eq!(remote_id, 4242); let unsubscribe_result = session .physical_session() .execute_json_rpc(crate::helius_transaction_unsubscribe_method(), crate::helius_transaction_unsubscribe_params(remote_id)) .await .expect("transactionUnsubscribe acknowledgement must arrive"); assert!(crate::decode_helius_transaction_unsubscribe_result(unsubscribe_result).expect("transactionUnsubscribe boolean must decode")); session.close().await.expect("Helius fixture session must close"); server.await.expect("local Helius transaction server must finish"); }