// file: kb-onchain-transport/src/standard_ws.rs // version: 4 //! Typed contracts for every standard Solana WebSocket subscription. /// Effective endpoint capabilities for unstable standard subscriptions. /// /// Configuration enables an optimistic first attempt. A persistent session may disable one /// capability after the node explicitly reports that the method is absent or not enabled. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)] #[serde(rename_all = "camelCase")] pub struct StandardWsCapabilities { /// Whether `blockSubscribe` may be attempted on this endpoint. pub block_subscribe: bool, /// Whether `slotsUpdatesSubscribe` may be attempted on this endpoint. pub slots_updates_subscribe: bool, /// Whether `voteSubscribe` may be attempted on this endpoint. pub vote_subscribe: bool, } impl crate::StandardWsCapabilities { /// Builds initial capabilities from request kinds explicitly advertised by an endpoint. pub fn from_endpoint(endpoint: &kb_config::WsEndpointConfig) -> Self { return Self { block_subscribe: endpoint_declares_method(endpoint, "blockSubscribe"), slots_updates_subscribe: endpoint_declares_method(endpoint, "slotsUpdatesSubscribe"), vote_subscribe: endpoint_declares_method(endpoint, "voteSubscribe"), }; } fn allows(&self, method: &str) -> bool { return match method { "blockSubscribe" => self.block_subscribe, "slotsUpdatesSubscribe" => self.slots_updates_subscribe, "voteSubscribe" => self.vote_subscribe, _ => true, }; } pub(crate) fn disable_method(&mut self, method: &str) -> bool { let capability = match method { "blockSubscribe" => &mut self.block_subscribe, "slotsUpdatesSubscribe" => &mut self.slots_updates_subscribe, "voteSubscribe" => &mut self.vote_subscribe, _ => return false, }; if !*capability { return false; } *capability = false; return true; } pub(crate) fn is_unstable_method(method: &str) -> bool { return matches!(method, "blockSubscribe" | "slotsUpdatesSubscribe" | "voteSubscribe"); } } /// Account options accepted by `accountSubscribe`. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct WsAccountSubscribeConfig { /// Optional commitment level. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub commitment: std::option::Option, /// Optional account-data encoding. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub encoding: std::option::Option, /// Optional account-data byte slice. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub data_slice: std::option::Option, } impl crate::WsAccountSubscribeConfig { fn validate(&self) -> kb_core::Result<()> { return crate::RpcAccountInfoConfig { encoding: self.encoding, data_slice: self.data_slice, commitment: self.commitment, min_context_slot: std::option::Option::None, } .validate(); } } /// Filter accepted by `blockSubscribe`. #[derive(Clone, Debug, Eq, PartialEq)] pub enum WsBlockFilter { /// Subscribe to every block. All, /// Subscribe to blocks mentioning one account or Program ID. MentionsAccountOrProgram(std::string::String), } impl serde::Serialize for crate::WsBlockFilter { fn serialize( &self, serializer: Serializer, ) -> std::result::Result where Serializer: serde::Serializer, { return match self { Self::All => serializer.serialize_str("all"), Self::MentionsAccountOrProgram(pubkey) => { let value = serde_json::json!({ "mentionsAccountOrProgram": pubkey }); serde::Serialize::serialize(&value, serializer) }, }; } } impl<'de> serde::Deserialize<'de> for crate::WsBlockFilter { fn deserialize( deserializer: Deserializer, ) -> std::result::Result where Deserializer: serde::Deserializer<'de>, { let value = match ::deserialize(deserializer) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; if value == serde_json::Value::String("all".to_string()) { return std::result::Result::Ok(Self::All); } let pubkey = value .as_object() .filter(|object| return object.len() == 1) .and_then(|object| return object.get("mentionsAccountOrProgram")) .and_then(serde_json::Value::as_str); return match pubkey { std::option::Option::Some(pubkey) => { std::result::Result::Ok(Self::MentionsAccountOrProgram(pubkey.to_string())) }, std::option::Option::None => { std::result::Result::Err(::custom( "block filter must be 'all' or mentionsAccountOrProgram", )) }, }; } } /// Options accepted by `blockSubscribe`. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct WsBlockSubscribeConfig { /// Optional confirmed or finalized commitment. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub commitment: std::option::Option, /// Optional transaction encoding. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub encoding: std::option::Option, /// Optional transaction detail level. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub transaction_details: std::option::Option, /// Whether rewards are included. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub show_rewards: std::option::Option, /// Highest transaction version the caller can decode. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub max_supported_transaction_version: std::option::Option, } impl crate::WsBlockSubscribeConfig { fn validate(&self) -> kb_core::Result<()> { if self.commitment == std::option::Option::Some(crate::RpcCommitmentLevel::Processed) { return std::result::Result::Err(kb_core::Error::config( "blockSubscribe does not accept processed commitment", )); } return std::result::Result::Ok(()); } } /// Filter accepted by `logsSubscribe`. #[derive(Clone, Debug, Eq, PartialEq)] pub enum WsLogsFilter { /// Subscribe to non-simple-vote transaction logs. All, /// Subscribe to every transaction log, including simple votes. AllWithVotes, /// Subscribe to transactions mentioning one account or Program ID. Mentions(std::string::String), } impl serde::Serialize for crate::WsLogsFilter { fn serialize( &self, serializer: Serializer, ) -> std::result::Result where Serializer: serde::Serializer, { return match self { Self::All => serializer.serialize_str("all"), Self::AllWithVotes => serializer.serialize_str("allWithVotes"), Self::Mentions(pubkey) => { let value = serde_json::json!({ "mentions": [pubkey] }); serde::Serialize::serialize(&value, serializer) }, }; } } impl<'de> serde::Deserialize<'de> for crate::WsLogsFilter { fn deserialize( deserializer: Deserializer, ) -> std::result::Result where Deserializer: serde::Deserializer<'de>, { let value = match ::deserialize(deserializer) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; if value == serde_json::Value::String("all".to_string()) { return std::result::Result::Ok(Self::All); } if value == serde_json::Value::String("allWithVotes".to_string()) { return std::result::Result::Ok(Self::AllWithVotes); } let mentions = value .as_object() .filter(|object| return object.len() == 1) .and_then(|object| return object.get("mentions")) .and_then(serde_json::Value::as_array); let pubkey = mentions .and_then(|values| { if values.len() == 1 { return values.first(); } return std::option::Option::None; }) .and_then(serde_json::Value::as_str); return match pubkey { std::option::Option::Some(pubkey) => { std::result::Result::Ok(Self::Mentions(pubkey.to_string())) }, std::option::Option::None => { std::result::Result::Err(::custom( "logs mentions filter must contain exactly one public key", )) }, }; } } /// Options accepted by `logsSubscribe`. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct WsLogsSubscribeConfig { /// Optional commitment level. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub commitment: std::option::Option, } /// Options accepted by `programSubscribe`. #[derive(Clone, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct WsProgramSubscribeConfig { /// Optional account filters, evaluated in order. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub filters: std::option::Option>, /// Optional account-data encoding. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub encoding: std::option::Option, /// Optional account-data byte slice. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub data_slice: std::option::Option, /// Optional commitment level. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub commitment: std::option::Option, /// Whether the result includes the standard context wrapper. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub with_context: std::option::Option, } impl crate::WsProgramSubscribeConfig { fn validate(&self) -> kb_core::Result<()> { return crate::RpcProgramAccountsConfig { filters: self.filters.clone(), encoding: self.encoding, data_slice: self.data_slice, commitment: self.commitment, min_context_slot: std::option::Option::None, with_context: self.with_context, sort_results: std::option::Option::None, } .validate(); } } /// Options accepted by `signatureSubscribe`. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct WsSignatureSubscribeConfig { /// Optional commitment level. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub commitment: std::option::Option, /// Whether an early `receivedSignature` notification is requested. #[serde(skip_serializing_if = "std::option::Option::is_none")] pub enable_received_notification: std::option::Option, } /// Typed account subscription request. #[derive(Clone, Debug, Eq, PartialEq)] pub struct AccountSubscribeRequest { /// Account public key. pub pubkey: std::string::String, /// Optional commitment, encoding and slice options. pub config: std::option::Option, } /// Typed block subscription request. #[derive(Clone, Debug, Eq, PartialEq)] pub struct BlockSubscribeRequest { /// Block selection filter. pub filter: crate::WsBlockFilter, /// Optional commitment, encoding, details, rewards and version options. pub config: std::option::Option, } /// Typed logs subscription request. #[derive(Clone, Debug, Eq, PartialEq)] pub struct LogsSubscribeRequest { /// Log selection filter. pub filter: crate::WsLogsFilter, /// Optional commitment. pub config: std::option::Option, } /// Typed program-account subscription request. #[derive(Clone, Debug, Eq, PartialEq)] pub struct ProgramSubscribeRequest { /// Program public key. pub program_id: std::string::String, /// Optional filters, commitment, encoding, slice and context switch. pub config: std::option::Option, } /// Typed root subscription request. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct RootSubscribeRequest; /// Typed signature subscription request. #[derive(Clone, Debug, Eq, PartialEq)] pub struct SignatureSubscribeRequest { /// Transaction signature. pub signature: std::string::String, /// Optional commitment and received-notification switch. pub config: std::option::Option, } /// Typed slot subscription request. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct SlotSubscribeRequest; /// Typed unstable slot-updates subscription request. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct SlotsUpdatesSubscribeRequest; /// Typed unstable vote subscription request. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct VoteSubscribeRequest; /// One of the nine standard Solana WebSocket subscription requests. #[derive(Clone, Debug, Eq, PartialEq)] pub enum StandardWsRequest { /// `accountSubscribe` request. Account(crate::AccountSubscribeRequest), /// `blockSubscribe` request. Block(crate::BlockSubscribeRequest), /// `logsSubscribe` request. Logs(crate::LogsSubscribeRequest), /// `programSubscribe` request. Program(crate::ProgramSubscribeRequest), /// `rootSubscribe` request. Root(crate::RootSubscribeRequest), /// `signatureSubscribe` request. Signature(crate::SignatureSubscribeRequest), /// `slotSubscribe` request. Slot(crate::SlotSubscribeRequest), /// `slotsUpdatesSubscribe` request. SlotsUpdates(crate::SlotsUpdatesSubscribeRequest), /// `voteSubscribe` request. Vote(crate::VoteSubscribeRequest), } impl crate::StandardWsRequest { /// Returns the canonical subscription specification. pub fn specification(&self) -> &'static crate::StandardWsSubscriptionSpec { return match self { Self::Account(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[0], Self::Block(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[1], Self::Logs(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[2], Self::Program(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[3], Self::Root(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[4], Self::Signature(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[5], Self::Slot(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[6], Self::SlotsUpdates(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[7], Self::Vote(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[8], }; } /// Returns the exact subscribe method. pub const fn subscribe_method(&self) -> &'static str { return match self { Self::Account(_) => "accountSubscribe", Self::Block(_) => "blockSubscribe", Self::Logs(_) => "logsSubscribe", Self::Program(_) => "programSubscribe", Self::Root(_) => "rootSubscribe", Self::Signature(_) => "signatureSubscribe", Self::Slot(_) => "slotSubscribe", Self::SlotsUpdates(_) => "slotsUpdatesSubscribe", Self::Vote(_) => "voteSubscribe", }; } /// Validates capability gates and builds exact positional JSON-RPC parameters. pub fn params( &self, capabilities: crate::StandardWsCapabilities, ) -> kb_core::Result> { if !capabilities.allows(self.subscribe_method()) { return std::result::Result::Err(kb_core::Error::config(format!( "unstable WebSocket method '{}' is not enabled by the endpoint capability set", self.subscribe_method() ))); } return match self { Self::Account(request) => account_params(request), Self::Block(request) => block_params(request), Self::Logs(request) => logs_params(request), Self::Program(request) => program_params(request), Self::Root(_) | Self::Slot(_) | Self::SlotsUpdates(_) | Self::Vote(_) => { std::result::Result::Ok(std::vec::Vec::new()) }, Self::Signature(request) => signature_params(request), }; } } /// Typed account notification. #[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] pub struct WsAccountNotification { /// Remote subscription identifier. pub subscription: u64, /// Contextual account value. pub result: crate::RpcResponse, } /// One block update emitted by `blockSubscribe`. #[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] #[serde(rename_all = "camelCase")] pub struct WsBlockUpdate { /// Block slot. pub slot: u64, /// Block when the node produced one for the slot. #[serde(default)] pub block: std::option::Option, /// Node-provided block error when present. #[serde(default)] pub err: std::option::Option, } /// Typed block notification. #[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] pub struct WsBlockNotification { /// Remote subscription identifier. pub subscription: u64, /// Contextual block update. pub result: crate::RpcResponse, } /// One log notification value. #[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] pub struct WsLogsValue { /// Transaction signature. pub signature: std::string::String, /// Runtime transaction error, or `None` on success. pub err: std::option::Option, /// Runtime log messages. pub logs: std::vec::Vec, } /// Typed logs notification. #[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] pub struct WsLogsNotification { /// Remote subscription identifier. pub subscription: u64, /// Contextual log result. pub result: crate::RpcResponse, } /// Typed program-account notification. #[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] pub struct WsProgramNotification { /// Remote subscription identifier. pub subscription: u64, /// Contextual keyed account result. pub result: crate::RpcOptionalContext, } /// Typed root notification. #[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] pub struct WsRootNotification { /// Remote subscription identifier. pub subscription: u64, /// New root slot. pub result: u64, } /// Signature notification result. #[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] #[serde(untagged)] pub enum WsSignatureValue { /// Final signature status. Status { /// Runtime transaction error, or `None` on success. err: std::option::Option, }, /// Early node-receipt marker. Received(std::string::String), } /// Typed signature notification. #[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] pub struct WsSignatureNotification { /// Remote subscription identifier. pub subscription: u64, /// Contextual receipt or final status. pub result: crate::RpcResponse, } /// One slot progression notification. #[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] pub struct WsSlotInfo { /// Current slot. pub slot: u64, /// Parent slot. pub parent: u64, /// Current root slot. pub root: u64, } /// Typed slot notification. #[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] pub struct WsSlotNotification { /// Remote subscription identifier. pub subscription: u64, /// Slot progression value. pub result: crate::WsSlotInfo, } /// Transaction counters included by a frozen-slot update. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)] #[serde(rename_all = "camelCase")] pub struct WsSlotTransactionStats { /// Total transaction entries processed by the bank. pub num_transaction_entries: u64, /// Successful transactions. pub num_successful_transactions: u64, /// Failed transactions. pub num_failed_transactions: u64, /// Maximum transactions per entry. pub max_transactions_per_entry: u64, } /// One unstable slot lifecycle update. #[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] #[serde(tag = "type", rename_all = "camelCase")] pub enum WsSlotUpdate { /// First shred for a slot was received. FirstShredReceived { /// Slot. slot: u64, /// Unix timestamp in milliseconds. timestamp: i64, }, /// Every shred for a slot was received. Completed { /// Slot. slot: u64, /// Unix timestamp in milliseconds. timestamp: i64, }, /// A bank was created for a slot. CreatedBank { /// Slot. slot: u64, /// Parent slot. parent: u64, /// Unix timestamp in milliseconds. timestamp: i64, }, /// A bank was frozen. Frozen { /// Slot. slot: u64, /// Unix timestamp in milliseconds. timestamp: i64, /// Bank transaction counters. stats: crate::WsSlotTransactionStats, }, /// A slot was marked dead. Dead { /// Slot. slot: u64, /// Unix timestamp in milliseconds. timestamp: i64, /// Node-provided failure description. err: std::string::String, }, /// A slot reached optimistic confirmation. OptimisticConfirmation { /// Slot. slot: u64, /// Unix timestamp in milliseconds. timestamp: i64, }, /// A slot became root. Root { /// Slot. slot: u64, /// Unix timestamp in milliseconds. timestamp: i64, }, } /// Typed unstable slot-updates notification. #[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] pub struct WsSlotsUpdatesNotification { /// Remote subscription identifier. pub subscription: u64, /// Slot lifecycle update. pub result: crate::WsSlotUpdate, } /// One vote notification value. #[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] #[serde(rename_all = "camelCase")] pub struct WsVoteValue { /// Vote account public key. pub vote_pubkey: std::string::String, /// Voted slots. pub slots: std::vec::Vec, /// Vote hash. pub hash: std::string::String, /// Optional vote timestamp. #[serde(default)] pub timestamp: std::option::Option, /// Vote transaction signature. pub signature: std::string::String, } /// Typed unstable vote notification. #[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] pub struct WsVoteNotification { /// Remote subscription identifier. pub subscription: u64, /// Vote value. pub result: crate::WsVoteValue, } /// Typed notification emitted by one of the nine standard subscriptions. #[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] #[serde(tag = "kind", content = "notification", rename_all = "camelCase")] pub enum StandardWsNotification { /// `accountNotification`. Account(crate::WsAccountNotification), /// `blockNotification`. Block(crate::WsBlockNotification), /// `logsNotification`. Logs(crate::WsLogsNotification), /// `programNotification`. Program(crate::WsProgramNotification), /// `rootNotification`. Root(crate::WsRootNotification), /// `signatureNotification`. Signature(crate::WsSignatureNotification), /// `slotNotification`. Slot(crate::WsSlotNotification), /// `slotsUpdatesNotification`. SlotsUpdates(crate::WsSlotsUpdatesNotification), /// `voteNotification`. Vote(crate::WsVoteNotification), } impl crate::StandardWsNotification { /// Returns the remote subscription identifier. pub const fn subscription_id(&self) -> u64 { return match self { Self::Account(value) => value.subscription, Self::Block(value) => value.subscription, Self::Logs(value) => value.subscription, Self::Program(value) => value.subscription, Self::Root(value) => value.subscription, Self::Signature(value) => value.subscription, Self::Slot(value) => value.subscription, Self::SlotsUpdates(value) => value.subscription, Self::Vote(value) => value.subscription, }; } } /// Parses and validates one standard WebSocket notification. pub fn adapt_standard_ws_notification( notification: &crate::JsonRpcNotification, ) -> kb_core::Result { let subscription = notification.params.subscription; let result = ¬ification.params.result; return match notification.method.as_str() { "accountNotification" => { adapt_notification_result(subscription, result, crate::WsAccountNotification::new) .map(crate::StandardWsNotification::Account) }, "blockNotification" => { adapt_notification_result(subscription, result, crate::WsBlockNotification::new) .map(crate::StandardWsNotification::Block) }, "logsNotification" => { adapt_notification_result(subscription, result, crate::WsLogsNotification::new) .map(crate::StandardWsNotification::Logs) }, "programNotification" => { adapt_notification_result(subscription, result, crate::WsProgramNotification::new) .map(crate::StandardWsNotification::Program) }, "rootNotification" => { adapt_notification_result(subscription, result, crate::WsRootNotification::new) .map(crate::StandardWsNotification::Root) }, "signatureNotification" => { let adapted = adapt_notification_result( subscription, result, crate::WsSignatureNotification::new, ); return match adapted { std::result::Result::Ok(notification) => { if let crate::WsSignatureValue::Received(marker) = ¬ification.result.value { if marker != "receivedSignature" { return std::result::Result::Err(kb_core::Error::json(format!( "unknown signature notification marker '{marker}'" ))); } } std::result::Result::Ok(crate::StandardWsNotification::Signature(notification)) }, std::result::Result::Err(error) => std::result::Result::Err(error), }; }, "slotNotification" => { adapt_notification_result(subscription, result, crate::WsSlotNotification::new) .map(crate::StandardWsNotification::Slot) }, "slotsUpdatesNotification" => { adapt_notification_result(subscription, result, crate::WsSlotsUpdatesNotification::new) .map(crate::StandardWsNotification::SlotsUpdates) }, "voteNotification" => { adapt_notification_result(subscription, result, crate::WsVoteNotification::new) .map(crate::StandardWsNotification::Vote) }, method => std::result::Result::Err(kb_core::Error::json(format!( "unsupported standard WebSocket notification method '{method}'" ))), }; } impl crate::WsAccountNotification { fn new(subscription: u64, result: crate::RpcResponse) -> Self { return Self { subscription, result }; } } impl crate::WsBlockNotification { fn new(subscription: u64, result: crate::RpcResponse) -> Self { return Self { subscription, result }; } } impl crate::WsLogsNotification { fn new(subscription: u64, result: crate::RpcResponse) -> Self { return Self { subscription, result }; } } impl crate::WsProgramNotification { fn new(subscription: u64, result: crate::RpcOptionalContext) -> Self { return Self { subscription, result }; } } impl crate::WsRootNotification { fn new(subscription: u64, result: u64) -> Self { return Self { subscription, result }; } } impl crate::WsSignatureNotification { fn new(subscription: u64, result: crate::RpcResponse) -> Self { return Self { subscription, result }; } } impl crate::WsSlotNotification { fn new(subscription: u64, result: crate::WsSlotInfo) -> Self { return Self { subscription, result }; } } impl crate::WsSlotsUpdatesNotification { fn new(subscription: u64, result: crate::WsSlotUpdate) -> Self { return Self { subscription, result }; } } impl crate::WsVoteNotification { fn new(subscription: u64, result: crate::WsVoteValue) -> Self { return Self { subscription, result }; } } fn endpoint_declares_method(endpoint: &kb_config::WsEndpointConfig, method: &str) -> bool { let request_kind = crate::request_kind_from_method(method); for role in &endpoint.roles { if !role.enabled { continue; } for configured_kind in &role.request_kinds { if configured_kind == "*" || configured_kind == &request_kind { return true; } } } return false; } fn account_params( request: &crate::AccountSubscribeRequest, ) -> kb_core::Result> { let validation_result = crate::validate_solana_pubkey_text(&request.pubkey, "account pubkey"); if let std::result::Result::Err(error) = validation_result { return std::result::Result::Err(error); } let mut params = std::vec![serde_json::Value::String(request.pubkey.clone())]; if let std::option::Option::Some(config) = request.config { let config_result = config.validate(); if let std::result::Result::Err(error) = config_result { return std::result::Result::Err(error); } let value = match crate::serialize_parameter("accountSubscribe", &config) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; params.push(value); } return std::result::Result::Ok(params); } fn block_params( request: &crate::BlockSubscribeRequest, ) -> kb_core::Result> { if let crate::WsBlockFilter::MentionsAccountOrProgram(pubkey) = &request.filter { let validation_result = crate::validate_solana_pubkey_text(pubkey, "block mentions account or program"); if let std::result::Result::Err(error) = validation_result { return std::result::Result::Err(error); } } let filter = match crate::serialize_parameter("blockSubscribe", &request.filter) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let mut params = std::vec![filter]; if let std::option::Option::Some(config) = request.config { let config_result = config.validate(); if let std::result::Result::Err(error) = config_result { return std::result::Result::Err(error); } let value = match crate::serialize_parameter("blockSubscribe", &config) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; params.push(value); } return std::result::Result::Ok(params); } fn logs_params( request: &crate::LogsSubscribeRequest, ) -> kb_core::Result> { if let crate::WsLogsFilter::Mentions(pubkey) = &request.filter { let validation_result = crate::validate_solana_pubkey_text(pubkey, "logs mention"); if let std::result::Result::Err(error) = validation_result { return std::result::Result::Err(error); } } let filter = match crate::serialize_parameter("logsSubscribe", &request.filter) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let mut params = std::vec![filter]; if let std::option::Option::Some(config) = request.config { let value = match crate::serialize_parameter("logsSubscribe", &config) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; params.push(value); } return std::result::Result::Ok(params); } fn program_params( request: &crate::ProgramSubscribeRequest, ) -> kb_core::Result> { let validation_result = crate::validate_solana_pubkey_text(&request.program_id, "program id"); if let std::result::Result::Err(error) = validation_result { return std::result::Result::Err(error); } let mut params = std::vec![serde_json::Value::String(request.program_id.clone())]; if let std::option::Option::Some(config) = &request.config { let config_result = config.validate(); if let std::result::Result::Err(error) = config_result { return std::result::Result::Err(error); } let value = match crate::serialize_parameter("programSubscribe", config) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; params.push(value); } return std::result::Result::Ok(params); } fn signature_params( request: &crate::SignatureSubscribeRequest, ) -> kb_core::Result> { let validation_result = crate::validate_transaction_signature_text(&request.signature, "transaction signature"); if let std::result::Result::Err(error) = validation_result { return std::result::Result::Err(error); } let mut params = std::vec![serde_json::Value::String(request.signature.clone())]; if let std::option::Option::Some(config) = request.config { let value = match crate::serialize_parameter("signatureSubscribe", &config) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; params.push(value); } return std::result::Result::Ok(params); } fn adapt_notification_result( subscription: u64, value: &serde_json::Value, constructor: Constructor, ) -> kb_core::Result where ResultType: serde::de::DeserializeOwned, Constructor: FnOnce(u64, ResultType) -> NotificationType, { let result = match serde_json::from_value::(value.clone()) { std::result::Result::Ok(result) => result, std::result::Result::Err(error) => { return std::result::Result::Err(kb_core::Error::json(format!( "cannot decode standard WebSocket notification result: {error}" ))); }, }; return std::result::Result::Ok(constructor(subscription, result)); } #[cfg(test)] mod tests { const PUBKEY: &str = "11111111111111111111111111111111"; const SIGNATURE: &str = "1111111111111111111111111111111111111111111111111111111111111111"; #[test] fn every_standard_request_maps_to_the_compiled_registry() { let requests = [ crate::StandardWsRequest::Account(crate::AccountSubscribeRequest { pubkey: PUBKEY.to_string(), config: std::option::Option::None, }), crate::StandardWsRequest::Block(crate::BlockSubscribeRequest { filter: crate::WsBlockFilter::All, config: std::option::Option::None, }), crate::StandardWsRequest::Logs(crate::LogsSubscribeRequest { filter: crate::WsLogsFilter::All, config: std::option::Option::None, }), crate::StandardWsRequest::Program(crate::ProgramSubscribeRequest { program_id: PUBKEY.to_string(), config: std::option::Option::None, }), crate::StandardWsRequest::Root(crate::RootSubscribeRequest), crate::StandardWsRequest::Signature(crate::SignatureSubscribeRequest { signature: SIGNATURE.to_string(), config: std::option::Option::None, }), crate::StandardWsRequest::Slot(crate::SlotSubscribeRequest), crate::StandardWsRequest::SlotsUpdates(crate::SlotsUpdatesSubscribeRequest), crate::StandardWsRequest::Vote(crate::VoteSubscribeRequest), ]; let mut methods = std::collections::BTreeSet::new(); for request in requests { assert!(methods.insert(request.subscribe_method())); assert_eq!(request.specification().subscribe_method, request.subscribe_method()); } assert_eq!(methods.len(), crate::STANDARD_WS_SUBSCRIPTIONS.len()); } #[test] fn stable_request_options_are_independently_serialized() { let request = crate::StandardWsRequest::Account(crate::AccountSubscribeRequest { pubkey: PUBKEY.to_string(), config: std::option::Option::Some(crate::WsAccountSubscribeConfig { commitment: std::option::Option::Some(crate::RpcCommitmentLevel::Confirmed), encoding: std::option::Option::Some(crate::RpcAccountEncoding::Base64Zstd), data_slice: std::option::Option::Some(crate::RpcDataSlice { offset: 8, length: 16, }), }), }); let params = match request.params(crate::StandardWsCapabilities::default()) { std::result::Result::Ok(params) => params, std::result::Result::Err(error) => panic!("params failed: {error}"), }; assert_eq!(params[0], serde_json::Value::String(PUBKEY.to_string())); assert_eq!(params[1]["commitment"], "confirmed"); assert_eq!(params[1]["encoding"], "base64+zstd"); assert_eq!(params[1]["dataSlice"]["offset"], 8); } #[test] fn custom_filter_objects_reject_unknown_fields() { let block = serde_json::json!({ "mentionsAccountOrProgram": PUBKEY, "unexpected": true }); assert!(serde_json::from_value::(block).is_err()); let logs = serde_json::json!({ "mentions": [PUBKEY], "unexpected": true }); assert!(serde_json::from_value::(logs).is_err()); } #[test] fn logs_mentions_keeps_exact_single_key_shape() { let request = crate::StandardWsRequest::Logs(crate::LogsSubscribeRequest { filter: crate::WsLogsFilter::Mentions(PUBKEY.to_string()), config: std::option::Option::Some(crate::WsLogsSubscribeConfig { commitment: std::option::Option::Some(crate::RpcCommitmentLevel::Finalized), }), }); let params = match request.params(crate::StandardWsCapabilities::default()) { std::result::Result::Ok(params) => params, std::result::Result::Err(error) => panic!("params failed: {error}"), }; assert_eq!(params[0]["mentions"][0], PUBKEY); assert_eq!(params[1]["commitment"], "finalized"); } #[test] fn unstable_requests_require_explicit_capabilities() { let requests = [ crate::StandardWsRequest::Block(crate::BlockSubscribeRequest { filter: crate::WsBlockFilter::All, config: std::option::Option::None, }), crate::StandardWsRequest::SlotsUpdates(crate::SlotsUpdatesSubscribeRequest), crate::StandardWsRequest::Vote(crate::VoteSubscribeRequest), ]; for request in &requests { assert!(request.params(crate::StandardWsCapabilities::default()).is_err()); } let capabilities = crate::StandardWsCapabilities { block_subscribe: true, slots_updates_subscribe: true, vote_subscribe: true, }; for request in &requests { assert!(request.params(capabilities).is_ok()); } } #[test] fn unstable_capabilities_can_be_disabled_after_runtime_rejection() { let mut capabilities = crate::StandardWsCapabilities { block_subscribe: true, slots_updates_subscribe: true, vote_subscribe: true, }; assert!(capabilities.disable_method("blockSubscribe")); assert!(!capabilities.block_subscribe); assert!(!capabilities.disable_method("blockSubscribe")); assert!(capabilities.disable_method("slotsUpdatesSubscribe")); assert!(capabilities.disable_method("voteSubscribe")); assert!(!capabilities.disable_method("slotSubscribe")); assert!(crate::StandardWsCapabilities::is_unstable_method("blockSubscribe")); assert!(!crate::StandardWsCapabilities::is_unstable_method("slotSubscribe")); } #[test] fn block_processed_commitment_is_rejected() { let request = crate::StandardWsRequest::Block(crate::BlockSubscribeRequest { filter: crate::WsBlockFilter::All, config: std::option::Option::Some(crate::WsBlockSubscribeConfig { commitment: std::option::Option::Some(crate::RpcCommitmentLevel::Processed), ..crate::WsBlockSubscribeConfig::default() }), }); let capabilities = crate::StandardWsCapabilities { block_subscribe: true, slots_updates_subscribe: false, vote_subscribe: false, }; assert!(request.params(capabilities).is_err()); } #[test] fn notification_adapter_decodes_each_standard_shape() { let account = serde_json::json!({ "context": { "slot": 1 }, "value": { "lamports": 1, "owner": PUBKEY, "executable": false, "rentEpoch": 2, "space": 0, "data": ["", "base64"] } }); let keyed_account = serde_json::json!({ "context": { "slot": 1 }, "value": { "pubkey": PUBKEY, "account": { "lamports": 1, "owner": PUBKEY, "executable": false, "rentEpoch": 2, "space": 0, "data": ["", "base64"] } } }); let cases = [ ("accountNotification", account, "account"), ( "blockNotification", serde_json::json!({ "context": { "slot": 4 }, "value": { "slot": 4, "block": { "blockhash": PUBKEY, "previousBlockhash": PUBKEY, "parentSlot": 3 }, "err": null } }), "block", ), ( "logsNotification", serde_json::json!({ "context": { "slot": 5 }, "value": { "signature": SIGNATURE, "err": null, "logs": ["Program log"] } }), "logs", ), ("programNotification", keyed_account, "program"), ("rootNotification", serde_json::Value::from(42_u64), "root"), ( "signatureNotification", serde_json::json!({ "context": { "slot": 6 }, "value": { "err": null } }), "signature", ), ( "slotNotification", serde_json::json!({ "slot": 3, "parent": 2, "root": 1 }), "slot", ), ( "slotsUpdatesNotification", serde_json::json!({ "type": "completed", "slot": 3, "timestamp": 4 }), "slotsUpdates", ), ( "voteNotification", serde_json::json!({ "votePubkey": PUBKEY, "slots": [1, 2], "hash": PUBKEY, "timestamp": null, "signature": SIGNATURE }), "vote", ), ]; for (method, result, kind) in cases { let notification = crate::JsonRpcNotification { jsonrpc: "2.0".to_string(), method: method.to_string(), params: crate::JsonRpcNotificationParams { result, subscription: 7 }, }; let adapted = match crate::adapt_standard_ws_notification(¬ification) { std::result::Result::Ok(adapted) => adapted, std::result::Result::Err(error) => panic!("notification failed: {error}"), }; let serialized = match serde_json::to_value(adapted) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => panic!("serialization failed: {error}"), }; assert_eq!(serialized["kind"], kind); } } #[test] fn signature_received_marker_is_exact() { let valid = crate::JsonRpcNotification { jsonrpc: "2.0".to_string(), method: "signatureNotification".to_string(), params: crate::JsonRpcNotificationParams { result: serde_json::json!({ "context": { "slot": 1 }, "value": "receivedSignature" }), subscription: 9, }, }; assert!(crate::adapt_standard_ws_notification(&valid).is_ok()); let mut invalid = valid; invalid.params.result["value"] = serde_json::Value::String("received".to_string()); assert!(crate::adapt_standard_ws_notification(&invalid).is_err()); } }