// file: crates/ksp-onchain-transport-lib/src/ws_helius_transactions.rs // version: 5 const MAX_HELIUS_TRANSACTION_FILTER_ACCOUNTS: usize = 50_000; /// Helius `tokenAccounts` expansion mode accepted by `transactionSubscribe`. #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub enum HeliusTokenAccountsFilter { /// Disable token-account owner expansion explicitly; equivalent to omitting `tokenAccounts`. None, /// Match transactions where a token balance owned by an included account changes or its token account closes. BalanceChanged, /// Match transactions referencing any token account owned by an included account, even if the balance does not change. All, } impl HeliusTokenAccountsFilter { /// Returns the exact Helius WebSocket wire string. #[must_use] pub const fn as_str(self) -> &'static str { return match self { Self::None => "none", Self::BalanceChanged => "balanceChanged", Self::All => "all", }; } } /// Transaction encoding accepted by Helius `transactionSubscribe`. #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub enum HeliusTransactionSubscribeEncoding { /// Base58 encoded transaction bytes. Base58, /// Base64 encoded transaction bytes. Base64, /// Parsed JSON transaction representation. JsonParsed, } impl HeliusTransactionSubscribeEncoding { /// Returns the exact Helius WebSocket wire string. #[must_use] pub const fn as_str(self) -> &'static str { return match self { Self::Base58 => "base58", Self::Base64 => "base64", Self::JsonParsed => "jsonParsed", }; } } /// Helius-specific filter object accepted as the first `transactionSubscribe` parameter. /// /// Debug output intentionally exposes only filter presence, modes and account counts. Transaction signatures and account values are omitted so routine /// diagnostics cannot accidentally disclose the caller's complete provider filter payload. #[derive(Clone, Default, Eq, PartialEq)] pub struct HeliusTransactionSubscribeFilter { vote: std::option::Option, failed: std::option::Option, signature: std::option::Option, account_include: std::option::Option>, account_exclude: std::option::Option>, account_required: std::option::Option>, token_accounts: std::option::Option, } impl HeliusTransactionSubscribeFilter { /// Creates a complete Helius transaction filter while preserving omitted versus explicitly empty account arrays. #[must_use] #[allow(clippy::too_many_arguments)] pub fn new( vote: std::option::Option, failed: std::option::Option, signature: std::option::Option, account_include: std::option::Option>, account_exclude: std::option::Option>, account_required: std::option::Option>, token_accounts: std::option::Option, ) -> Self { return Self { vote, failed, signature, account_include, account_exclude, account_required, token_accounts }; } /// Returns the optional vote-transaction filter flag. #[must_use] pub const fn vote(&self) -> std::option::Option { return self.vote; } /// Returns the optional failed-transaction filter flag. #[must_use] pub const fn failed(&self) -> std::option::Option { return self.failed; } /// Returns the optional exact transaction signature filter. #[must_use] pub fn signature(&self) -> std::option::Option<&str> { return match self.signature.as_ref() { std::option::Option::Some(signature) => std::option::Option::Some(signature.as_str()), std::option::Option::None => std::option::Option::None, }; } /// Returns the optional OR-style account inclusion list. #[must_use] pub fn account_include(&self) -> std::option::Option<&[ksp_core_lib::Pubkey]> { return match self.account_include.as_ref() { std::option::Option::Some(accounts) => std::option::Option::Some(accounts.as_slice()), std::option::Option::None => std::option::Option::None, }; } /// Returns the optional account exclusion list. #[must_use] pub fn account_exclude(&self) -> std::option::Option<&[ksp_core_lib::Pubkey]> { return match self.account_exclude.as_ref() { std::option::Option::Some(accounts) => std::option::Option::Some(accounts.as_slice()), std::option::Option::None => std::option::Option::None, }; } /// Returns the optional AND-style required-account list. #[must_use] pub fn account_required(&self) -> std::option::Option<&[ksp_core_lib::Pubkey]> { return match self.account_required.as_ref() { std::option::Option::Some(accounts) => std::option::Option::Some(accounts.as_slice()), std::option::Option::None => std::option::Option::None, }; } /// Returns the optional Helius token-account owner-expansion mode. #[must_use] pub const fn token_accounts(&self) -> std::option::Option { return self.token_accounts; } fn validate(&self) -> ksp_core_lib::Result<()> { let include = validate_account_list("accountInclude", self.account_include.as_deref()); if let std::result::Result::Err(error) = include { return std::result::Result::Err(error); } let exclude = validate_account_list("accountExclude", self.account_exclude.as_deref()); if let std::result::Result::Err(error) = exclude { return std::result::Result::Err(error); } let required = validate_account_list("accountRequired", self.account_required.as_deref()); if let std::result::Result::Err(error) = required { return std::result::Result::Err(error); } return std::result::Result::Ok(()); } fn to_json_value(&self) -> serde_json::Value { let mut object = serde_json::Map::new(); if let std::option::Option::Some(vote) = self.vote { object.insert("vote".to_owned(), serde_json::Value::Bool(vote)); } if let std::option::Option::Some(failed) = self.failed { object.insert("failed".to_owned(), serde_json::Value::Bool(failed)); } if let std::option::Option::Some(signature) = self.signature.as_ref() { object.insert("signature".to_owned(), serde_json::Value::String(signature.clone())); } insert_account_list(&mut object, "accountInclude", self.account_include.as_deref()); insert_account_list(&mut object, "accountExclude", self.account_exclude.as_deref()); insert_account_list(&mut object, "accountRequired", self.account_required.as_deref()); if let std::option::Option::Some(token_accounts) = self.token_accounts { object.insert("tokenAccounts".to_owned(), serde_json::Value::String(token_accounts.as_str().to_owned())); } return serde_json::Value::Object(object); } } impl std::fmt::Debug for HeliusTransactionSubscribeFilter { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { return formatter .debug_struct("HeliusTransactionSubscribeFilter") .field("vote", &self.vote) .field("failed", &self.failed) .field("signature_present", &self.signature.is_some()) .field("account_include_count", &self.account_include.as_ref().map(std::vec::Vec::len)) .field("account_exclude_count", &self.account_exclude.as_ref().map(std::vec::Vec::len)) .field("account_required_count", &self.account_required.as_ref().map(std::vec::Vec::len)) .field("token_accounts", &self.token_accounts) .finish(); } } /// Optional Helius `transactionSubscribe` result-shaping configuration. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct HeliusTransactionSubscribeOptions { commitment: std::option::Option, encoding: std::option::Option, transaction_details: std::option::Option, show_rewards: std::option::Option, max_supported_transaction_version: std::option::Option, } impl HeliusTransactionSubscribeOptions { /// Creates a complete optional Helius transaction-subscription configuration. #[must_use] pub const fn new( commitment: std::option::Option, encoding: std::option::Option, transaction_details: std::option::Option, show_rewards: std::option::Option, max_supported_transaction_version: std::option::Option, ) -> Self { return Self { commitment, encoding, transaction_details, show_rewards, max_supported_transaction_version }; } /// Returns the optional commitment level. #[must_use] pub const fn commitment(&self) -> std::option::Option { return self.commitment; } /// Returns the optional Helius transaction encoding. #[must_use] pub const fn encoding(&self) -> std::option::Option { return self.encoding; } /// Returns the optional transaction detail level. #[must_use] pub const fn transaction_details(&self) -> std::option::Option { return self.transaction_details; } /// Returns whether rewards were explicitly requested. #[must_use] pub const fn show_rewards(&self) -> std::option::Option { return self.show_rewards; } /// Returns the highest transaction version the caller declares it can consume. #[must_use] pub const fn max_supported_transaction_version(&self) -> std::option::Option { return self.max_supported_transaction_version; } fn validate(&self) -> ksp_core_lib::Result<()> { let requires_version = matches!(self.transaction_details, std::option::Option::Some(crate::SolanaTransactionDetails::Full | crate::SolanaTransactionDetails::Accounts)); if requires_version && self.max_supported_transaction_version.is_none() { let detail = match self.transaction_details { std::option::Option::Some(detail) => detail.as_str(), std::option::Option::None => "omitted", }; return std::result::Result::Err( ksp_core_lib::Error::new( crate::ERROR_CODE_INVALID_RPC_PARAMETERS, "Helius transactionSubscribe requires maxSupportedTransactionVersion for full or accounts transaction details", ) .with_context("rpc_method", "transactionSubscribe") .with_context("field", "maxSupportedTransactionVersion") .with_context("transaction_details", detail), ); } return std::result::Result::Ok(()); } fn to_json_value(self) -> serde_json::Value { let mut object = serde_json::Map::new(); if let std::option::Option::Some(commitment) = self.commitment { object.insert("commitment".to_owned(), serde_json::Value::String(commitment.as_str().to_owned())); } if let std::option::Option::Some(encoding) = self.encoding { object.insert("encoding".to_owned(), serde_json::Value::String(encoding.as_str().to_owned())); } if let std::option::Option::Some(transaction_details) = self.transaction_details { object.insert("transactionDetails".to_owned(), serde_json::Value::String(transaction_details.as_str().to_owned())); } if let std::option::Option::Some(show_rewards) = self.show_rewards { object.insert("showRewards".to_owned(), serde_json::Value::Bool(show_rewards)); } if let std::option::Option::Some(version) = self.max_supported_transaction_version { object.insert("maxSupportedTransactionVersion".to_owned(), serde_json::Value::Number(version.into())); } return serde_json::Value::Object(object); } } /// Complete typed request contract for Helius `transactionSubscribe`. /// /// The request owns the exact provider filter and optional result-shaping object. Validation and serialization occur before actor registration so deterministic /// provider constraints fail without WebSocket I/O. #[derive(Clone, Eq, PartialEq)] pub struct HeliusTransactionSubscribeRequest { filter: crate::HeliusTransactionSubscribeFilter, options: std::option::Option, } impl HeliusTransactionSubscribeRequest { /// Creates one typed Helius transaction-subscription request. #[must_use] pub fn new(filter: crate::HeliusTransactionSubscribeFilter, options: std::option::Option) -> Self { return Self { filter, options }; } /// Returns the provider transaction filter. #[must_use] pub const fn filter(&self) -> &crate::HeliusTransactionSubscribeFilter { return &self.filter; } /// Returns the optional provider result-shaping configuration. #[must_use] pub const fn options(&self) -> std::option::Option<&crate::HeliusTransactionSubscribeOptions> { return self.options.as_ref(); } /// Validates deterministic Helius request constraints before any WebSocket I/O. pub fn validate(&self) -> ksp_core_lib::Result<()> { let filter = self.filter.validate(); if let std::result::Result::Err(error) = filter { return std::result::Result::Err(error); } if let std::option::Option::Some(options) = self.options { let options = options.validate(); if let std::result::Result::Err(error) = options { return std::result::Result::Err(error); } } return std::result::Result::Ok(()); } } impl std::fmt::Debug for HeliusTransactionSubscribeRequest { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { return formatter.debug_struct("HeliusTransactionSubscribeRequest").field("filter", &self.filter).field("options", &self.options).finish(); } } fn helius_transaction_subscribe_params(request: &crate::HeliusTransactionSubscribeRequest) -> ksp_core_lib::Result> { let validation = request.validate(); if let std::result::Result::Err(error) = validation { return std::result::Result::Err(error); } let mut params = std::vec![request.filter.to_json_value()]; if let std::option::Option::Some(options) = request.options { params.push(options.to_json_value()); } return std::result::Result::Ok(params); } /// Full/accounts-mode notification delivered by Helius `transactionSubscribe`. /// /// The nested transaction payload is deliberately retained as JSON because its exact Solana wire representation depends on the requested encoding and detail /// mode. KSP types the stable provider envelope while preserving the full nested payload without Program-specific decoding. #[derive(Clone, PartialEq)] pub struct HeliusFullTransactionNotification { transaction: serde_json::Value, signature: std::string::String, slot: u64, transaction_index: u64, } impl HeliusFullTransactionNotification { /// Returns the provider transaction/status payload without interpreting Program-specific contents. #[must_use] pub const fn transaction(&self) -> &serde_json::Value { return &self.transaction; } /// Returns the base58 transaction signature reported by Helius. #[must_use] pub fn signature(&self) -> &str { return self.signature.as_str(); } /// Returns the slot in which the transaction was processed. #[must_use] pub const fn slot(&self) -> u64 { return self.slot; } /// Returns the zero-based transaction position within the block. #[must_use] pub const fn transaction_index(&self) -> u64 { return self.transaction_index; } } impl std::fmt::Debug for HeliusFullTransactionNotification { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { return formatter .debug_struct("HeliusFullTransactionNotification") .field("transaction", &"") .field("signature", &"") .field("slot", &self.slot) .field("transaction_index", &self.transaction_index) .finish(); } } /// Signatures-mode notification delivered by Helius `transactionSubscribe`. #[derive(Clone, PartialEq)] pub struct HeliusTransactionSignatureNotification { signature: std::string::String, slot: u64, transaction_index: u64, err: crate::SolanaWireField, memo: crate::SolanaWireField, block_time: crate::SolanaWireField, confirmation_status: crate::SolanaWireField, } impl HeliusTransactionSignatureNotification { /// Returns the base58 transaction signature reported by Helius. #[must_use] pub fn signature(&self) -> &str { return self.signature.as_str(); } /// Returns the slot in which the transaction was processed. #[must_use] pub const fn slot(&self) -> u64 { return self.slot; } /// Returns the zero-based transaction position within the block. #[must_use] pub const fn transaction_index(&self) -> u64 { return self.transaction_index; } /// Returns the optional transaction error while preserving omitted/null/value wire states. #[must_use] pub const fn err(&self) -> &crate::SolanaWireField { return &self.err; } /// Returns the optional memo while preserving omitted/null/value wire states. #[must_use] pub const fn memo(&self) -> &crate::SolanaWireField { return &self.memo; } /// Returns the optional block time while preserving omitted/null/value wire states. #[must_use] pub const fn block_time(&self) -> &crate::SolanaWireField { return &self.block_time; } /// Returns the optional confirmation-status label while preserving omitted/null/value wire states. #[must_use] pub const fn confirmation_status(&self) -> &crate::SolanaWireField { return &self.confirmation_status; } } impl std::fmt::Debug for HeliusTransactionSignatureNotification { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { return formatter .debug_struct("HeliusTransactionSignatureNotification") .field("signature", &"") .field("slot", &self.slot) .field("transaction_index", &self.transaction_index) .field("err", &wire_field_debug_state(&self.err)) .field("memo", &wire_field_debug_state(&self.memo)) .field("block_time", &wire_field_debug_state(&self.block_time)) .field("confirmation_status", &wire_field_debug_state(&self.confirmation_status)) .finish(); } } fn wire_field_debug_state(field: &crate::SolanaWireField) -> &'static str { if field.is_omitted() { return "omitted"; } if field.is_null() { return "null"; } return "value"; } /// Typed Helius `transactionNotification` payload union. /// /// `Full` also covers the provider `accounts` detail mode because both contain the nested `transaction` member. `Signature` covers the lightweight /// signatures mode. `Unknown` preserves `none` mode and forward-compatible provider shapes instead of failing the logical subscription. #[derive(Clone, PartialEq)] #[non_exhaustive] pub enum HeliusTransactionNotification { /// Full/accounts notification carrying the nested transaction payload. Full(crate::HeliusFullTransactionNotification), /// Lightweight signatures notification. Signature(crate::HeliusTransactionSignatureNotification), /// Provider shape not currently typed by KSP, preserved losslessly. Unknown(serde_json::Value), } impl std::fmt::Debug for HeliusTransactionNotification { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { return match self { Self::Full(notification) => formatter.debug_tuple("Full").field(notification).finish(), Self::Signature(notification) => formatter.debug_tuple("Signature").field(notification).finish(), Self::Unknown(_) => formatter.debug_tuple("Unknown").field(&"").finish(), }; } } impl crate::HeliusLaserStreamWsSession { /// Opens one Helius `transactionSubscribe` logical subscription through the shared physical actor. /// /// The returned handle keeps a stable local identity across physical reconnects. Helius remote subscription IDs stay actor-private and are remapped after /// resubscribe. Calling [`crate::WsSubscription::unsubscribe`] removes the remote mapping before sending `transactionUnsubscribe`, so provider messages /// already in flight after cancellation are ignored without reactivating the logical subscription. pub async fn transaction_subscribe( &self, request: &crate::HeliusTransactionSubscribeRequest, ) -> ksp_core_lib::Result> { let params = helius_transaction_subscribe_params(request); let params = match params { std::result::Result::Ok(params) => params, std::result::Result::Err(error) => return std::result::Result::Err(error), }; return self .physical_session() .subscribe_typed(crate::WsSubscriptionKind::HeliusTransaction, params, |value| return decode_helius_transaction_notification(value)) .await; } } #[derive(serde::Deserialize)] #[serde(rename_all = "camelCase")] struct WireHeliusFullTransactionNotification { transaction: serde_json::Value, signature: std::string::String, slot: u64, transaction_index: u64, } #[derive(serde::Deserialize)] #[serde(rename_all = "camelCase")] struct WireHeliusTransactionSignatureNotification { signature: std::string::String, slot: u64, transaction_index: u64, #[serde(default)] err: crate::SolanaWireField, #[serde(default)] memo: crate::SolanaWireField, #[serde(default)] block_time: crate::SolanaWireField, #[serde(default)] confirmation_status: crate::SolanaWireField, } fn decode_helius_transaction_notification(value: serde_json::Value) -> ksp_core_lib::Result { if value.get("transaction").is_some() { let decoded = crate::decode_wire_json::("transactionNotification", value.clone()); if let std::result::Result::Ok(decoded) = decoded { return std::result::Result::Ok(crate::HeliusTransactionNotification::Full(crate::HeliusFullTransactionNotification { transaction: decoded.transaction, signature: decoded.signature, slot: decoded.slot, transaction_index: decoded.transaction_index, })); } } if value.get("signature").is_some() && value.get("slot").is_some() && value.get("transactionIndex").is_some() { let decoded = crate::decode_wire_json::("transactionNotification", value.clone()); if let std::result::Result::Ok(decoded) = decoded { return std::result::Result::Ok(crate::HeliusTransactionNotification::Signature(crate::HeliusTransactionSignatureNotification { signature: decoded.signature, slot: decoded.slot, transaction_index: decoded.transaction_index, err: decoded.err, memo: decoded.memo, block_time: decoded.block_time, confirmation_status: decoded.confirmation_status, })); } } return std::result::Result::Ok(crate::HeliusTransactionNotification::Unknown(value)); } fn validate_account_list(field: &'static str, accounts: std::option::Option<&[ksp_core_lib::Pubkey]>) -> ksp_core_lib::Result<()> { if let std::option::Option::Some(accounts) = accounts && accounts.len() > MAX_HELIUS_TRANSACTION_FILTER_ACCOUNTS { return std::result::Result::Err( ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RPC_PARAMETERS, "Helius transactionSubscribe account filter exceeds the provider limit") .with_context("rpc_method", "transactionSubscribe") .with_context("field", field) .with_context("actual_count", accounts.len().to_string()) .with_context("max_count", MAX_HELIUS_TRANSACTION_FILTER_ACCOUNTS.to_string()), ); } return std::result::Result::Ok(()); } fn insert_account_list( object: &mut serde_json::Map, field: &'static str, accounts: std::option::Option<&[ksp_core_lib::Pubkey]>, ) { if let std::option::Option::Some(accounts) = accounts { let values = accounts.iter().map(|account| return serde_json::Value::String(account.to_string())).collect::>(); object.insert(field.to_owned(), serde_json::Value::Array(values)); } return; } #[cfg(test)] #[path = "../unit_tests/ws_helius_transactions.rs"] mod tests;