khadhroony-solana-project/crates/ksp-onchain-transport-lib/src/ws_transactions.rs

// file: crates/ksp-onchain-transport-lib/src/ws_transactions.rs
// version: 4

/// Optional configuration accepted by standard Solana `signatureSubscribe`.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct SolanaSignatureSubscribeConfig {
    commitment: std::option::Option<crate::SolanaCommitment>,
    enable_received_notification: std::option::Option<bool>,
}

impl SolanaSignatureSubscribeConfig {
    /// Creates an explicit signature-subscription configuration.
    #[must_use]
    pub const fn new(commitment: std::option::Option<crate::SolanaCommitment>, enable_received_notification: std::option::Option<bool>) -> Self {
        return Self { commitment, enable_received_notification };
    }

    /// Returns the optional commitment level.
    #[must_use]
    pub const fn commitment(&self) -> std::option::Option<crate::SolanaCommitment> {
        return self.commitment;
    }

    /// Returns whether the server should emit the early `receivedSignature` notification when explicitly configured.
    #[must_use]
    pub const fn enable_received_notification(&self) -> std::option::Option<bool> {
        return self.enable_received_notification;
    }

    fn is_empty(&self) -> bool {
        return self.commitment.is_none() && self.enable_received_notification.is_none();
    }

    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(enable_received_notification) = self.enable_received_notification {
            object.insert("enableReceivedNotification".to_owned(), serde_json::Value::Bool(enable_received_notification));
        }
        return serde_json::Value::Object(object);
    }
}

/// Typed value carried by standard Solana `signatureNotification` messages.
#[derive(Clone, Debug, PartialEq)]
pub enum SolanaSignatureNotification {
    /// Early notification emitted when the RPC node first receives the signature and `enableReceivedNotification` is enabled.
    ReceivedSignature,
    /// Terminal processing notification emitted when the configured commitment is reached.
    Processed {
        /// Nullable transaction-error wire value; `None` means the transaction succeeded at the requested commitment.
        err: std::option::Option<serde_json::Value>,
    },
}

impl SolanaSignatureNotification {
    /// Returns whether this notification terminates the server-side one-shot subscription.
    #[must_use]
    pub const fn is_terminal(&self) -> bool {
        return match self {
            Self::ReceivedSignature => false,
            Self::Processed { .. } => true,
        };
    }

    /// Returns the transaction-error wire value for a terminal processing notification when present.
    #[must_use]
    pub const fn err(&self) -> std::option::Option<&serde_json::Value> {
        return match self {
            Self::ReceivedSignature | Self::Processed { err: std::option::Option::None } => std::option::Option::None,
            Self::Processed { err: std::option::Option::Some(err) } => std::option::Option::Some(err),
        };
    }
}

/// Filter accepted by the standard Solana `logsSubscribe` WebSocket method.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum SolanaLogsSubscribeFilter {
    /// Subscribe to all transactions except simple vote transactions.
    All,
    /// Subscribe to all transactions including simple vote transactions.
    AllWithVotes,
    /// Subscribe only to transactions mentioning exactly one public key.
    Mentions(ksp_core_lib::Pubkey),
}

impl SolanaLogsSubscribeFilter {
    fn to_json_value(&self) -> serde_json::Value {
        return match self {
            Self::All => serde_json::Value::String("all".to_owned()),
            Self::AllWithVotes => serde_json::Value::String("allWithVotes".to_owned()),
            Self::Mentions(pubkey) => serde_json::json!({"mentions": [pubkey.to_string()]}),
        };
    }
}

/// Typed value carried by a contextual Solana `logsNotification`.
#[derive(Clone, Debug, PartialEq)]
pub struct SolanaLogsNotification {
    signature: std::string::String,
    err: std::option::Option<serde_json::Value>,
    logs: std::vec::Vec<std::string::String>,
}

impl SolanaLogsNotification {
    /// Returns the base58 transaction signature exactly as reported by the RPC node.
    #[must_use]
    pub fn signature(&self) -> &str {
        return self.signature.as_str();
    }

    /// Returns the nullable transaction-error wire value without interpreting Program/runtime error semantics.
    #[must_use]
    pub const fn err(&self) -> std::option::Option<&serde_json::Value> {
        return self.err.as_ref();
    }

    /// Returns the ordered transaction log messages.
    #[must_use]
    pub fn logs(&self) -> &[std::string::String] {
        return self.logs.as_slice();
    }
}

impl crate::WsSession {
    /// Subscribes to one Solana transaction signature through standard `signatureSubscribe`.
    ///
    /// The server automatically terminates this subscription after the terminal processed notification. When
    /// `enableReceivedNotification` is enabled, an earlier `ReceivedSignature` value may be delivered first without closing the logical handle.
    pub async fn signature_subscribe(
        &self,
        signature: &str,
        config: std::option::Option<&crate::SolanaSignatureSubscribeConfig>,
    ) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaRpcResponse<crate::SolanaSignatureNotification>>> {
        let mut params = std::vec![serde_json::Value::String(signature.to_owned())];
        if let std::option::Option::Some(config) = config
            && !config.is_empty()
        {
            params.push((*config).to_json_value());
        }
        return self
            .subscribe_typed_with_completion(
                crate::WsSubscriptionKind::Signature,
                params,
                |value| return decode_signature_notification("signatureSubscribe", value),
                |notification| return notification.value().is_terminal(),
            )
            .await;
    }

    /// Subscribes to Solana transaction logs through standard `logsSubscribe`.
    pub async fn logs_subscribe(
        &self,
        filter: &crate::SolanaLogsSubscribeFilter,
        config: std::option::Option<&crate::SolanaCommitmentConfig>,
    ) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaRpcResponse<crate::SolanaLogsNotification>>> {
        let mut params = std::vec![filter.to_json_value()];
        if let std::option::Option::Some(config) = config
            && config.commitment().is_some()
        {
            params.push(config.to_json_value());
        }
        return self.subscribe_typed(crate::WsSubscriptionKind::Logs, params, |value| return decode_logs_notification("logsSubscribe", value)).await;
    }
}

#[derive(serde::Deserialize)]
#[serde(untagged)]
enum WireSignatureNotification {
    Received(std::string::String),
    Processed(WireSignatureProcessed),
}

#[derive(serde::Deserialize)]
struct WireSignatureProcessed {
    err: serde_json::Value,
}

#[derive(serde::Deserialize)]
struct WireRpcResponseSignature {
    context: serde_json::Value,
    value: WireSignatureNotification,
}

#[derive(serde::Deserialize)]
struct WireRpcResponse {
    context: serde_json::Value,
    value: WireLogsNotification,
}

#[derive(serde::Deserialize)]
struct WireLogsNotification {
    signature: std::string::String,
    err: serde_json::Value,
    logs: std::vec::Vec<std::string::String>,
}

fn decode_signature_notification(method: &str, value: serde_json::Value) -> ksp_core_lib::Result<crate::SolanaRpcResponse<crate::SolanaSignatureNotification>> {
    let decoded = crate::decode_wire_json::<WireRpcResponseSignature>(method, value);
    let wire = match decoded {
        std::result::Result::Ok(wire) => wire,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let context = crate::SolanaRpcContext::decode_wire(method, wire.context);
    let context = match context {
        std::result::Result::Ok(context) => context,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let notification = match wire.value {
        WireSignatureNotification::Received(value) if value == "receivedSignature" => crate::SolanaSignatureNotification::ReceivedSignature,
        WireSignatureNotification::Received(_) => {
            return std::result::Result::Err(
                ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, "signatureSubscribe notification contains an unknown string variant")
                    .with_context("rpc_method", method),
            );
        },
        WireSignatureNotification::Processed(processed) => {
            let err = match processed.err {
                serde_json::Value::Null => std::option::Option::None,
                value => std::option::Option::Some(value),
            };
            crate::SolanaSignatureNotification::Processed { err }
        },
    };
    return std::result::Result::Ok(crate::SolanaRpcResponse::new(context, notification));
}

fn decode_logs_notification(method: &str, value: serde_json::Value) -> ksp_core_lib::Result<crate::SolanaRpcResponse<crate::SolanaLogsNotification>> {
    let decoded = crate::decode_wire_json::<WireRpcResponse>(method, value);
    let wire = match decoded {
        std::result::Result::Ok(wire) => wire,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let context = crate::SolanaRpcContext::decode_wire(method, wire.context);
    let context = match context {
        std::result::Result::Ok(context) => context,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let err = match wire.value.err {
        serde_json::Value::Null => std::option::Option::None,
        value => std::option::Option::Some(value),
    };
    let notification = crate::SolanaLogsNotification { signature: wire.value.signature, err, logs: wire.value.logs };
    return std::result::Result::Ok(crate::SolanaRpcResponse::new(context, notification));
}

impl crate::SolanaStandardWsSession {
    /// Subscribes to one Solana transaction signature through standard `signatureSubscribe`.
    pub async fn signature_subscribe(
        &self,
        signature: &str,
        config: std::option::Option<&crate::SolanaSignatureSubscribeConfig>,
    ) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaRpcResponse<crate::SolanaSignatureNotification>>> {
        return self.physical_session().signature_subscribe(signature, config).await;
    }

    /// Subscribes to Solana transaction logs through standard `logsSubscribe`.
    pub async fn logs_subscribe(
        &self,
        filter: &crate::SolanaLogsSubscribeFilter,
        config: std::option::Option<&crate::SolanaCommitmentConfig>,
    ) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaRpcResponse<crate::SolanaLogsNotification>>> {
        return self.physical_session().logs_subscribe(filter, config).await;
    }
}

impl crate::HeliusLaserStreamWsSession {
    /// Subscribes to one transaction signature through the standard `signatureSubscribe` wire supported by Helius LaserStream WebSocket.
    pub async fn signature_subscribe(
        &self,
        signature: &str,
        config: std::option::Option<&crate::SolanaSignatureSubscribeConfig>,
    ) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaRpcResponse<crate::SolanaSignatureNotification>>> {
        return self.physical_session().signature_subscribe(signature, config).await;
    }

    /// Subscribes to transaction logs through the standard `logsSubscribe` wire supported by Helius LaserStream WebSocket.
    pub async fn logs_subscribe(
        &self,
        filter: &crate::SolanaLogsSubscribeFilter,
        config: std::option::Option<&crate::SolanaCommitmentConfig>,
    ) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaRpcResponse<crate::SolanaLogsNotification>>> {
        return self.physical_session().logs_subscribe(filter, config).await;
    }
}

#[cfg(test)]
#[path = "../unit_tests/ws_transactions.rs"]
mod tests;