rialo-cdk 0.2.0-alpha.0

// Copyright (c) Subzero Labs, Inc.
// SPDX-License-Identifier: Apache-2.0

use std::str::FromStr;

use async_trait::async_trait;
use base64::Engine as _;
#[cfg(feature = "rpc-client")]
use reqwest::{Client, Url};
#[cfg(feature = "rpc-client")]
use rialo_api_types::{
    messages::{
        get_account_info::{GetAccountInfoConfig, GetAccountInfoRequest},
        get_accounts_by_owner::{GetAccountsByOwnerConfig, GetAccountsByOwnerRequest},
        get_epoch_info::{GetEpochInfoConfig, GetEpochInfoRequest},
        get_fee_for_message::GetFeeForMessageRequest,
        get_minimum_balance_for_rent_exemption::GetMinimumBalanceForRentExemptionRequest,
        get_multiple_accounts::GetMultipleAccountsRequest,
        get_secret_sharing_pubkey::GetSecretSharingPubkeyResponse,
        get_signature_statuses::GetSignatureStatusesRequest,
        get_signatures_for_address::GetSignaturesForAddressRequest,
        get_transactions::{
            GetTransactionsConfig, GetTransactionsRequest, GetTransactionsResponse,
        },
        submit_epoch_change::{SubmitEpochChangeRequest, SubmitEpochChangeResponse},
    },
    parameters::*,
};
#[cfg(feature = "distributed-tracing")]
use rialo_telemetry::{apply_trace_headers_to_reqwest, inject_trace_headers};
#[cfg(feature = "rpc-client")]
use serde_json::{json, Value};

#[cfg(feature = "rpc-client")]
use crate::constants::RPC_VERSION;
#[cfg(feature = "rpc-client")]
use crate::error::{Result, RialoError};
pub use crate::rpc::no_wasm::*;
#[cfg(feature = "rpc-client")]
use crate::rpc::traits::{GenericRpcCaller, RpcClient};
#[cfg(feature = "rpc-client")]
use crate::rpc::types::*;

/// `HttpRpcClient` implements JSON-RPC communication with the Rialo blockchain
/// over HTTP.
///
/// This client handles serialization, deserialization, and error handling for all
/// RPC requests to the blockchain.
#[cfg(feature = "rpc-client")]
pub struct HttpRpcClient {
    /// The URL of the RPC endpoint
    url: String,
    /// HTTP client used for making requests
    client: Client,
    /// Atomic counter for generating unique request IDs
    id: std::sync::atomic::AtomicU64,
}

#[cfg(feature = "rpc-client")]
impl HttpRpcClient {
    /// Creates a new HTTP RPC client connected to the specified URL.
    ///
    /// Certificate validation is automatically disabled for localhost URLs
    /// (localhost, 127.0.0.1, ::1) and enabled for all other URLs.
    ///
    /// # Arguments
    ///
    /// * `url` - The URL of the Rialo RPC endpoint
    ///
    /// # Returns
    ///
    /// A new `HttpRpcClient` instance
    pub fn new(url: String) -> Self {
        let danger_accept_invalid_certs = Self::is_localhost(&url);
        Self::new_with_config(url, danger_accept_invalid_certs)
    }

    /// Creates a new HTTP RPC client with explicit TLS configuration.
    ///
    /// # Arguments
    ///
    /// * `url` - The URL of the Rialo RPC endpoint
    /// * `danger_accept_invalid_certs` - Whether to accept invalid TLS certificates
    ///   (DANGEROUS - only use for testing with self-signed certificates).
    ///   Note: This parameter is ignored on wasm32 targets.
    ///
    /// # Returns
    ///
    /// A new `HttpRpcClient` instance
    pub fn new_with_config(
        url: String,
        #[allow(unused_variables)] danger_accept_invalid_certs: bool,
    ) -> Self {
        let client = {
            #[cfg(not(target_arch = "wasm32"))]
            let client = Client::builder()
                .danger_accept_invalid_certs(danger_accept_invalid_certs)
                .build()
                .expect("Failed to build HTTP client");

            #[cfg(target_arch = "wasm32")]
            let client = Client::builder()
                .build()
                .expect("Failed to build HTTP client");

            client
        };

        Self {
            url,
            client,
            id: std::sync::atomic::AtomicU64::new(1),
        }
    }

    /// Checks if a URL points to localhost.
    ///
    /// Returns true for:
    /// - localhost
    /// - 127.0.0.1
    /// - ::1 (IPv6 localhost)
    ///
    /// # Arguments
    ///
    /// * `url` - The URL to check
    ///
    /// # Returns
    ///
    /// `true` if the URL is a localhost URL, `false` otherwise
    pub fn is_localhost(url: &str) -> bool {
        if let Ok(parsed) = Url::parse(url) {
            if let Some(host) = parsed.host_str() {
                return host == "localhost"
                    || host == "127.0.0.1"
                    || host == "::1"
                    || host == "[::1]";
            }
        }
        false
    }

    /// Generates the next unique request ID for RPC calls.
    ///
    /// # Returns
    ///
    /// A unique u64 ID for the next RPC request
    fn next_id(&self) -> u64 {
        self.id.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
    }
}

#[cfg(feature = "rpc-client")]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl GenericRpcCaller for HttpRpcClient {
    /// Sends a generic RPC request to the Rialo blockchain and parses the response.
    ///
    /// # Arguments
    ///
    /// * `method` - The RPC method name to call
    /// * `params` - The parameters to pass to the method
    ///
    /// # Returns
    ///
    /// The deserialized response result or an error if the request failed
    ///
    /// # Errors
    ///
    /// Returns `RialoError::Rpc` if the request fails, receives an error response,
    /// or if the response cannot be properly parsed
    #[tracing::instrument(skip(self, params))]
    async fn call<T, R>(&self, method: &str, params: T) -> Result<R>
    where
        T: serde::Serialize + NoWasmSend,
        R: for<'de> serde::Deserialize<'de> + NoWasmSend,
    {
        let id = self.next_id();
        let request = json!({
            "jsonrpc": RPC_VERSION,
            "id": id,
            "method": method,
            "params": params,
        });

        log::trace!(
            "Calling the equivalent of: curl -X POST {} -H 'Content-Type: application/json' -d '{}'",
            self.url,
            serde_json::to_string(&request).unwrap_or_default()
        );

        // Build request with trace context propagation
        #[allow(unused_mut)]
        let mut request_builder = self.client.post(&self.url).json(&request);

        #[cfg(feature = "distributed-tracing")]
        {
            let trace_headers = inject_trace_headers();
            request_builder = apply_trace_headers_to_reqwest(request_builder, trace_headers);
        }

        let response = request_builder.send().await?;

        if !response.status().is_success() {
            let status = response.status();
            let body = response
                .text()
                .await
                .unwrap_or_else(|_| "Unable to read response body".to_string());
            log::error!("HTTP request failed with status {status} and body: {body}");

            // Try to parse JSON-RPC error from the response body
            let error_details = if let Ok(json) = serde_json::from_str::<Value>(&body) {
                if let Some(error) = json.get("error") {
                    let code = error.get("code").and_then(|c| c.as_i64()).unwrap_or(0);
                    let message = error
                        .get("message")
                        .and_then(|m| m.as_str())
                        .unwrap_or("Unknown error")
                        .to_string();
                    crate::error::RpcErrorDetails::new(Some(status.as_u16()), code, message)
                } else {
                    crate::error::RpcErrorDetails::new(
                        Some(status.as_u16()),
                        status.as_u16() as i64,
                        body,
                    )
                }
            } else {
                crate::error::RpcErrorDetails::new(
                    Some(status.as_u16()),
                    status.as_u16() as i64,
                    body,
                )
            };

            return Err(RialoError::Rpc(error_details));
        }

        let json: Value = response.json().await?;

        if let Some(error) = json.get("error") {
            // Enhanced error handling - extract more details from the error object
            let code = error.get("code").and_then(|c| c.as_i64()).unwrap_or(0);
            let mut message = error
                .get("message")
                .and_then(|m| m.as_str())
                .unwrap_or("Unknown error")
                .to_string();

            // Include data field in message if present
            if let Some(data) = error.get("data") {
                if !data.to_string().is_empty() && *data != "null" {
                    message.push_str(&format!(" (data: {data})"));
                }
            }

            let error_details = crate::error::RpcErrorDetails::new(None, code, message);
            return Err(RialoError::Rpc(error_details));
        }

        let result = json.get("result").ok_or_else(|| {
            let error_details = crate::error::RpcErrorDetails::from_message(
                "Missing 'result' field in response".to_string(),
            );
            RialoError::Rpc(error_details)
        })?;

        serde_json::from_value(result.clone()).map_err(|e| {
            let error_details =
                crate::error::RpcErrorDetails::from_message(format!("Failed to parse result: {e}"));
            RialoError::Rpc(error_details)
        })
    }
}

#[cfg(feature = "rpc-client")]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl RpcClient for HttpRpcClient {
    /// Sends an RPC request with parameters as a JSON value.
    ///
    /// # Arguments
    ///
    /// * `method` - The RPC method name to call
    /// * `params` - The parameters as a JSON value
    ///
    /// # Returns
    ///
    /// The response as a JSON value or an error
    async fn call_with_json(
        &self,
        method: &str,
        params: serde_json::Value,
    ) -> Result<serde_json::Value> {
        self.call(method, params).await
    }

    /// Gets the balance of a specific account.
    ///
    /// # Arguments
    ///
    /// * `pubkey` - The public key of the account as a string
    ///
    /// # Returns
    ///
    /// The account balance in kelvin
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails or the response format is invalid
    async fn get_balance(&self, pubkey: &Pubkey) -> Result<u64> {
        let params = GetBalanceRequest::new(pubkey.to_string());
        let result: GetBalanceResponse = self.call("getBalance", [params]).await?;

        Ok(result.value)
    }

    /// Submits a signed transaction to the blockchain.
    ///
    /// # Arguments
    ///
    /// * `transaction` - The serialized transaction as a byte array
    /// * `options` - The options for the transaction.
    ///   If None, we default to {skip_preflight: true}
    ///
    /// # Returns
    ///
    /// The transaction signature as a string
    ///
    /// # Errors
    ///
    /// Returns an error if the transaction submission fails
    async fn send_transaction(
        &self,
        transaction: &[u8],
        options: Option<SendTransactionOptions>,
    ) -> Result<Signature> {
        let options = options.unwrap_or(SendTransactionOptions {
            skip_preflight: true,
            ..Default::default()
        });
        #[allow(clippy::disallowed_methods)]
        let encoded = base64::engine::general_purpose::STANDARD.encode(transaction);
        let params = SendTransactionRequest::new(encoded, options.into());
        let signature_str: String = self.call("sendTransaction", [params]).await?;
        Signature::from_str(&signature_str).map_err(|e| {
            let error_details = crate::error::RpcErrorDetails::from_message(format!(
                "Invalid signature format: {e}"
            ));
            RialoError::Rpc(error_details)
        })
    }

    /// Fetches account information for a given public key.
    ///
    /// # Arguments
    ///
    /// * `pubkey` - The public key of the account as a string
    ///
    /// # Returns
    ///
    /// Account information including owner, data, and kelvin
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails or the response format is invalid
    async fn get_account_info(&self, pubkey: &Pubkey) -> Result<AccountInfo> {
        let params = GetAccountInfoRequest {
            address: pubkey.to_string(),
            config: Some(GetAccountInfoConfig {
                encoding: Some("base64".to_string()),
                data_slice: None,
                min_context_slot: None,
            }),
        };
        let result: Value = self.call("getAccountInfo", [params]).await?;

        serde_json::from_value(
            result
                .get("value")
                .ok_or_else(|| {
                    let error_details = crate::error::RpcErrorDetails::from_message(
                        "Missing 'value' in getAccountInfo response".to_string(),
                    );
                    RialoError::Rpc(error_details)
                })?
                .clone(),
        )
        .map_err(|e| {
            let error_details = crate::error::RpcErrorDetails::from_message(format!(
                "Failed to parse account info: {e}"
            ));
            RialoError::Rpc(error_details)
        })
    }

    /// Gets the current block height from the blockchain.
    ///
    /// # Returns
    ///
    /// The current finalized block height.
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails.
    async fn get_block_height(&self) -> Result<u64> {
        self.get_block_height_with_config(None).await
    }

    async fn get_block_height_with_config(&self, min_context_slot: Option<u64>) -> Result<u64> {
        let config = RequestBlockHeightConfig { min_context_slot };

        // Send as array to match RPC server expectations
        self.call("getBlockHeight", [config]).await
    }

    /// Retrieves transaction details by signature.
    ///
    /// # Arguments
    ///
    /// * `signature` - The transaction signature as a string
    ///
    /// # Returns
    ///
    /// The transaction details
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails or the transaction is not found
    async fn get_transaction(&self, signature: &Signature) -> Result<TransactionResponse> {
        let params = GetTransactionRequest::new(signature.to_string(), Some("json".to_string()));
        let response: GetTransactionResponse = self.call("getTransaction", [params]).await?;

        // Convert from API type to CDK compatibility type
        Ok(TransactionResponse {
            block_height: response.block_height,
            block_hash: response.block_hash,
            meta: response.meta,
            transaction: response.transaction,
            block_time: response.block_time,
        })
    }

    /// Gets the total transaction count in the ledger.
    ///
    /// # Returns
    ///
    /// The current transaction count
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails
    async fn get_transaction_count(&self) -> Result<u64> {
        let params = GetTransactionCountRequest::new();
        let response: GetTransactionCountResponse =
            self.call("getTransactionCount", [params]).await?;
        Ok(response.value)
    }

    /// Calculates the minimum balance required for rent exemption.
    ///
    /// # Arguments
    ///
    /// * `size` - The size of the account data in bytes
    ///
    /// # Returns
    ///
    /// The minimum balance in kelvin required for rent exemption
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails
    async fn get_minimum_balance_for_rent_exemption(&self, size: u64) -> Result<u64> {
        let params = GetMinimumBalanceForRentExemptionRequest { data_length: size };
        self.call("getMinimumBalanceForRentExemption", [params])
            .await
    }

    /// Retrieves the status of multiple transaction signatures.
    ///
    /// # Arguments
    ///
    /// * `signatures` - A slice of transaction signatures as strings
    ///
    /// # Returns
    ///
    /// A vector of signature statuses
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails or the response format is invalid
    async fn get_signature_statuses(
        &self,
        signatures: &[Signature],
    ) -> Result<Vec<SignatureStatus>> {
        let signature_strings: Vec<String> = signatures.iter().map(|s| s.to_string()).collect();
        let params = GetSignatureStatusesRequest {
            signatures: signature_strings,
            config: None,
        };
        let response: GetSignatureStatusesResponse =
            self.call("getSignatureStatuses", [params]).await?;

        // Convert from API response format to CDK compatibility format
        let signature_statuses: Vec<SignatureStatus> = response
            .value
            .into_iter()
            .map(|status| SignatureStatus {
                slot: response.context.slot,
                err: status.err.map(|e| e.to_string()),
                executed: status.executed,
            })
            .collect();

        Ok(signature_statuses)
    }

    /// Gets subscriptions for a given subscriber.
    ///
    /// # Arguments
    ///
    /// * `subscriber` - The public key of the subscriber
    /// * `nonce` - The nonce identifying the specific subscription
    ///
    /// # Returns
    ///
    /// The subscription for the subscriber
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails or the response format is invalid
    async fn get_subscription(
        &self,
        subscriber: &Pubkey,
        nonce: &str,
    ) -> Result<GetSubscriptionResponse> {
        let params =
            GetSubscriptionRequest::new(Some(subscriber.to_string()), Some(nonce.to_string()));
        self.call("getSubscription", [params]).await
    }

    /// Gets triggered transactions for a subscription account.
    ///
    /// # Arguments
    ///
    /// * `subscription_account` - The public key of the subscription account
    /// * `limit` - Optional limit on the number of transactions to return
    ///
    /// # Returns
    ///
    /// A response containing triggered transactions for the subscription
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails or the response format is invalid
    async fn get_triggered_transactions(
        &self,
        subscription_account: &Pubkey,
        limit: Option<u32>,
    ) -> Result<GetTriggeredTransactionsResponse> {
        // Use array params to match node handler expectations:
        // [subscription_pubkey: string, limit: string?]
        let mut params: Vec<Value> = vec![Value::String(subscription_account.to_string())];
        if let Some(lim) = limit {
            params.push(Value::String(lim.to_string()));
        }

        let result: Value = self.call("getTriggeredTransactions", params).await?;

        serde_json::from_value(result).map_err(|e| {
            let error_details = crate::error::RpcErrorDetails::from_message(format!(
                "Failed to parse triggered transactions response: {e}"
            ));
            RialoError::Rpc(error_details)
        })
    }

    /// Requests an airdrop of tokens to the specified account.
    ///
    /// # Arguments
    ///
    /// * `pubkey` - The public key of the receiving account
    /// * `amount` - The amount of kelvin to request
    ///
    /// # Returns
    ///
    /// The transaction signature of the airdrop
    ///
    /// # Errors
    ///
    /// Returns an error if the airdrop request fails
    async fn request_airdrop(&self, pubkey: &Pubkey, amount: u64) -> Result<Signature> {
        let params = RequestAirdropRequest::new(pubkey.to_string(), amount);
        let signature_str: String = self.call("requestAirdrop", [params]).await?;
        Signature::from_str(&signature_str).map_err(|e| {
            let error_details = crate::error::RpcErrorDetails::from_message(format!(
                "Invalid signature format: {e}"
            ));
            RialoError::Rpc(error_details)
        })
    }

    /// Gets epoch information.
    ///
    /// # Returns
    ///
    /// Epoch information including slot indices and transaction count
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails
    async fn get_epoch_info(&self) -> Result<EpochInfoResponse> {
        let params = GetEpochInfoRequest {
            config: Some(GetEpochInfoConfig {
                min_context_slot: None,
            }),
        };
        self.call("getEpochInfo", [params]).await
    }

    /// Gets the fee for a given message.
    ///
    /// # Arguments
    ///
    /// * `message` - The message to calculate fees for
    ///
    /// # Returns
    ///
    /// The fee required for the message
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails
    async fn get_fee_for_message(&self, message: &str) -> Result<GetFeeForMessageResponse> {
        let params = GetFeeForMessageRequest {
            message: message.to_string(),
        };
        self.call("getFeeForMessage", [params]).await
    }

    /// Gets information for multiple accounts.
    ///
    /// # Arguments
    ///
    /// * `pubkeys` - A slice of public keys to query
    ///
    /// # Returns
    ///
    /// Account information for the requested accounts
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails
    async fn get_multiple_accounts(&self, pubkeys: &[Pubkey]) -> Result<MultipleAccountsResponse> {
        let pubkey_strings: Vec<String> = pubkeys.iter().map(|pk| pk.to_string()).collect();
        let params = GetMultipleAccountsRequest {
            addresses: pubkey_strings,
        };
        let response: MultipleAccountsResponse = self.call("getMultipleAccounts", [params]).await?;
        Ok(response)
    }

    /// Gets workflow lineage information.
    ///
    /// # Arguments
    ///
    /// * `request` - The workflow lineage request parameters
    ///
    /// # Returns
    ///
    /// Workflow lineage information
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails
    async fn get_workflow_lineage(
        &self,
        request: &GetWorkflowLineageRequest,
    ) -> Result<GetWorkflowLineageResponse> {
        let params = request.clone();
        self.call("getWorkflowLineage", [params]).await
    }

    /// Gets signatures for a given address.
    ///
    /// # Arguments
    ///
    /// * `address` - The address to query signatures for
    /// * `config` - Optional configuration for the request
    ///
    /// # Returns
    ///
    /// Signatures for the given address
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails
    async fn get_signatures_for_address(
        &self,
        address: &Pubkey,
        config: Option<GetSignaturesForAddressConfig>,
    ) -> Result<GetSignaturesForAddressResponse> {
        let params = GetSignaturesForAddressRequest {
            address: address.to_string(),
            config,
        };
        self.call("getSignaturesForAddress", [params]).await
    }

    async fn get_transactions(
        &self,
        config: Option<GetTransactionsConfig>,
    ) -> Result<GetTransactionsResponse> {
        let params = GetTransactionsRequest { config };
        self.call("getTransactions", [params]).await
    }

    /// Checks if a blockhash is valid.
    ///
    /// # Arguments
    ///
    /// * `blockhash` - The blockhash to validate
    ///
    /// # Returns
    ///
    /// Whether the blockhash is valid
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails
    async fn is_blockhash_valid(&self, blockhash: &Hash) -> Result<IsBlockhashValidResponse> {
        let params = IsBlockhashValidRequest::new(blockhash.to_string());
        let result: Value = self.call("isBlockhashValid", [params]).await?;

        let slot = result
            .get("context")
            .and_then(|ctx| ctx.get("slot"))
            .and_then(|s| s.as_u64())
            .unwrap_or(0);

        let is_valid = result
            .get("value")
            .and_then(|v| v.as_bool())
            .unwrap_or(false);

        Ok(IsBlockhashValidResponse::new(slot, is_valid))
    }

    /// Gets the health status of the node.
    ///
    /// # Returns
    ///
    /// The health status of the node, typically "ok" when healthy
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails
    async fn get_health(&self) -> Result<GetHealthResponse> {
        let params = GetHealthRequest::new();
        self.call("getHealth", [params]).await
    }

    /// Gets the health status of the validator.
    ///
    /// # Returns
    ///
    /// The validator health status, typically "ok" when healthy
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails
    async fn get_validator_health(&self) -> Result<GetValidatorHealthResponse> {
        let params = GetValidatorHealthRequest::new();
        self.call("getValidatorHealth", [params]).await
    }

    /// Gets the full nodes connected to the validator or full node.
    ///
    /// # Returns
    ///
    /// A list of full nodes identified by a `NetworkPublicKey` and how long
    /// they have been connected in milliseconds.
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails
    async fn get_connected_full_nodes(&self) -> Result<GetConnectedFullNodesResponse> {
        let params = GetConnectedFullNodesRequest::new();
        self.call("getConnectedFullNodes", [params]).await
    }

    async fn get_secret_sharing_pubkey(&self) -> Result<GetSecretSharingPubkeyResponse> {
        self.call("getSecretSharingPubkey", [Value::Null]).await
    }

    async fn send_admin_transaction(
        &self,
        params: SubmitEpochChangeRequest,
    ) -> Result<SubmitEpochChangeResponse> {
        self.call("submitEpochChange", [params]).await
    }

    /// Gets accounts by owner.
    ///
    /// # Arguments
    ///
    /// * `owner` - The public key of the owner to query accounts for
    ///
    /// # Returns
    ///
    /// A response containing accounts matching the filter criteria
    ///
    /// # Errors
    ///
    /// Returns an error if the RPC call fails
    async fn get_accounts_by_owner(&self, owner: &Pubkey) -> Result<GetAccountsByOwnerResponse> {
        let params = GetAccountsByOwnerRequest::with_config(
            owner.to_string(),
            rialo_api_types::messages::get_accounts_by_owner::AccountFilter::ProgramAccounts,
            GetAccountsByOwnerConfig::base64(),
        );
        self.call("getAccountsByOwner", [params]).await
    }
}