delta_base_sdk/

rpc.rs

//! # RPC Communication with the delta Base Layer
//!
//! This module provides a client for communicating with delta base layer through
//! RPC (Remote Procedure Call). It enables applications to query network state, submit
//! transactions, retrieve vault data, and subscribe to network events.
//!
//! ## Overview
//!
//! The primary interface is the [`BaseRpcClient`] which handles connection
//! management, request serialization, and response parsing. The client provides
//! methods for:
//!
//! - [Submitting transactions to the network](BaseRpcClient::submit_transaction);
//! - Retrieving vault data ([`get_vault`](BaseRpcClient::get_vault),
//!   [`get_base_vault`](BaseRpcClient::get_base_vault),
//!   [`get_vaults`](BaseRpcClient::get_vaults));
//! - [Checking transaction status](BaseRpcClient::get_transaction_status); or
//! - [Subscribing to network events](BaseRpcClient::stream_base_layer_events).
//!
//! All RPC operations use the [`tonic`] library for gRPC communication, with automatic
//! serialization and deserialization between protocol buffer and native Rust types.
//!
//! ## Usage
//!
//! To use the RPC client, you need a running delta base layer at some known URL.
//!
//! ## Example
//!
//! ```rust,no_run
//! use delta_base_sdk::{
//!     core::Shard,
//!     crypto::{ed25519, Hash256, IntoSignatureEnum},
//!     rpc::{BaseRpcClient, RpcError},
//!     vaults::{Address, ReadableNativeBalance, Vault},
//! };
//! # use std::error::Error;
//! # use std::str::FromStr;
//! # use tokio_stream::StreamExt;
//!
//! async fn rpc_example() -> Result<(), Box<dyn Error>> {
//!     // Connect to base layer
//!     let client = BaseRpcClient::new("http://localhost:50051").await?;
//!
//!     // Get a vault
//!     let pubkey = ed25519::PrivKey::generate().pub_key();
//!     let vault = client.get_vault(Address::new(pubkey.owner(), 1)).await?;
//!     println!("Vault balance: {}", vault.balance());
//!
//!     // Check a transaction status
//!     // In a real app, this would be a valid signature
//!     let signature = ed25519::Signature::from_str("<MY SIGNATURE>")?
//!         .into_signature_enum(pubkey);
//!     let status = client.get_transaction_status(signature.hash_sha256()).await?;
//!     println!("Transaction status: {:?}", status);
//!
//!     // Query current epoch
//!     let epoch = client.get_epoch().await?;
//!     println!("Current epoch: {}", epoch);
//!
//!     // Subscribe to network events (requires tokio to run)
//!     let mut events = client.stream_base_layer_events(0).await?;
//!
//!     // Process the first 5 events
//!     let mut count = 0;
//!     while let Some(event) = events.next().await {
//!         if let Ok(event) = event {
//!             println!("Received event: {:?}", event);
//!             count += 1;
//!             if count >= 5 {
//!                 break;
//!             }
//!         }
//!     }
//!     Ok(())
//! }
//! ```

use crate::{
    core::Shard,
    crypto::HashDigest,
    events::BaseLayerEvent,
    sdl::StateDiffList,
    transactions::{
        types::TransactionType,
        SignedTransaction,
        TransactionStatus,
        TypeData,
    },
};
use http::{
    uri::{
        self,
        InvalidUri,
        Scheme,
    },
    Uri,
};
use primitives::{
    domain_agreement::DomainAgreement,
    type_aliases::{
        Epoch,
        TxId,
    },
    vault::{
        base::Vault as BaseVault,
        sharded::Vault as ShardedVault,
        Address,
        OwnerId,
    },
};
use proto_types::{
    error::HashSnafu,
    sdl::SdlStatus,
    services::{
        GetBaseVaultRequest,
        GetDomainAgreementRequest,
        GetDomainOwnerIdsRequest,
        GetEpochRequest,
        GetEpochResponse,
        GetSdlDataRequest,
        GetShardedVaultsRequest,
        GetTransactionStatusRequest,
        StreamBaseLayerEventRequest,
        ValidatorServiceClient,
    },
};
use snafu::{
    ResultExt,
    Snafu,
};
use std::collections::HashMap;
use tokio_stream::{
    Stream,
    StreamExt,
};
use tonic::{
    transport::Channel,
    Request,
};

/// Errors that can occur when using the [`BaseRpcClient`]
#[derive(Debug, Snafu)]
pub enum RpcError {
    /// Failed to establish a connection to base layer
    ///
    /// This error occurs when the client cannot connect to the specified
    /// base layer URL, typically due to network issues or because
    /// the node is not running.
    #[snafu(display("Could not connect: {source}"))]
    Connection {
        /// The underlying transport error
        source: tonic::transport::Error,
    },

    /// The RPC request was rejected by base layer
    ///
    /// This error occurs when the request is received by the base layer
    /// but cannot be processed, such as when requesting a non-existent
    /// vault or submitting an invalid transaction.
    #[snafu(display("Request failed with status: {status}"))]
    Request {
        /// The status returned by the base layer
        #[snafu(source(from(tonic::Status, Box::new)))]
        status: Box<tonic::Status>,
    },

    /// Failed to convert between protocol buffer and native format
    ///
    /// This error occurs when there's a problem converting between the
    /// wire format (protocol buffers) and the native Rust types, typically
    /// due to missing or malformed data.
    #[snafu(display("Could not convert type: {source}"))]
    IntoProto {
        /// The underlying conversion error
        source: proto_types::error::ProtoError,
    },

    /// Failed to parse base layer URL
    ///
    /// This error occurs when the URL provided to connect to the base layer is
    /// malformed or cannot be parsed.
    #[snafu(display("Could not parse URL: {source}"))]
    InvalidUrl {
        /// The underlying HTTP error
        source: http::Error,
    },
}

impl RpcError {
    /// Returns true if the underlying error is a gRPC not found status.
    pub fn is_not_found(&self) -> bool {
        matches!(self, Self::Request { status } if status.code() == tonic::Code::NotFound)
    }
}

/// Client for communicating with the delta base layer through RPC
///
/// [BaseRpcClient] provides a high-level interface for interacting with
/// delta base layers. It handles connection management, request formatting,
/// and response parsing for all base layer operations.
///
/// # Example
///
/// ```rust,no_run
/// use delta_base_sdk::rpc::BaseRpcClient;
///
/// async fn connect() -> Result<(), Box<dyn std::error::Error>> {
///     // Connect to local base layer provider
///     let client = BaseRpcClient::new("http://localhost:50051").await?;
///
///     // Connect to a remote node with HTTPS
///     let secure_client = BaseRpcClient::new("https://baselayer.example.com").await?;
///
///     // Connect with just the host (HTTPS will be added automatically)
///     let auto_scheme_client = BaseRpcClient::new("baselayer.example.com").await?;
///
///     Ok(())
/// }
/// ```
#[derive(Debug, Clone)]
pub struct BaseRpcClient {
    // The inner client is cloned in all calls but this is cheap and avoids mutable self refs:
    // https://docs.rs/tonic/latest/tonic/transport/struct.Channel.html#multiplexing-requests
    inner: ValidatorServiceClient<Channel>,
}

impl BaseRpcClient {
    /// Creates a new client by connecting to the specified base layer provider URL
    ///
    /// This method establishes a connection to delta base layer at the specified
    /// URL. If the URL does not include a scheme (http:// or https://), HTTPS will be
    /// used automatically.
    ///
    /// # Parameters
    ///
    /// * `url` - URL of base layer provider to connect to, either as a string or a URI
    ///
    /// # Errors
    ///
    /// Returns [RpcError] if:
    /// - The URL cannot be parsed; or
    /// - The connection cannot be established
    pub async fn new(url: impl TryInto<Uri, Error = InvalidUri> + Send) -> Result<Self, RpcError> {
        let uri = add_missing_scheme(url)?;
        let inner = ValidatorServiceClient::connect(uri.to_string())
            .await
            .context(ConnectionSnafu)?;
        Ok(Self { inner })
    }

    /// Submits a signed transaction to the base layer network
    ///
    /// This method sends a transaction to base layer for processing.
    /// The transaction must be properly signed with a valid signature from
    /// the transaction signer's private key.
    ///
    /// # Parameters
    ///
    /// * `tx` - Signed transaction to submit
    ///
    /// # Errors
    ///
    /// Returns [RpcError] if:
    /// - The connection to the base layer provider fails;
    /// - The base layer provider rejects the transaction; or
    /// - There's a protocol conversion error.
    pub async fn submit_transaction<T>(&self, tx: SignedTransaction<T>) -> Result<(), RpcError>
    where
        T: TransactionType + Into<TypeData> + Send,
    {
        self.inner
            .clone()
            .submit_transaction(Request::new(tx.into()))
            .await
            .context(RequestSnafu)?;
        Ok(())
    }

    /// Gets the base vault for an owner in the base layer shard
    ///
    /// This is a convenience method that calls [`get_vault`](BaseRpcClient::get_vault)
    /// with the base layer shard (0).
    ///
    /// # Parameters
    ///
    /// * `owner` - The ID of the vault owner
    ///
    /// # Errors
    ///
    /// Returns [RpcError] if:
    /// - The vault doesn't exist;
    /// - The connection to the base layer provider fails; or
    /// - There's a protocol conversion error.
    pub async fn get_base_vault(&self, owner: OwnerId) -> Result<BaseVault, RpcError> {
        let proto_vault = self
            .inner
            .clone()
            .get_base_vault(GetBaseVaultRequest {
                owner: owner.into(),
            })
            .await
            .context(RequestSnafu)?
            .into_inner();

        Ok(proto_vault.into())
    }

    /// Retrieves a domain vault data for a specific address
    ///
    /// # Parameters
    ///
    /// * `address` - of the vault to fetch
    ///
    /// # Errors
    ///
    /// Returns [RpcError] if:
    /// - The vault under the given address doesn't exist;
    /// - The connection to the base layer provider fails; or
    /// - There's a protocol conversion error.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use delta_base_sdk::{
    ///     crypto::ed25519,
    ///     rpc::BaseRpcClient,
    ///     vaults::{Address, ReadableNativeBalance},
    /// };
    ///
    /// async fn get_vault_example(client: &BaseRpcClient) -> Result<(), Box<dyn std::error::Error>> {
    ///     let address = Address::new(ed25519::PubKey::generate().owner(), 1);
    ///     let vault = client.get_vault(address).await?;
    ///     println!("Vault data: {:?}", vault);
    ///     println!("Vault balance: {}", vault.balance());
    ///     Ok(())
    /// }
    /// ```
    pub async fn get_vault(&self, address: Address) -> Result<ShardedVault, RpcError> {
        let mut vaults = self.get_vaults([address]).await?;
        vaults.remove(&address).ok_or_else(|| RpcError::Request {
            status: Box::new(tonic::Status::not_found("Vault not found")),
        })
    }

    /// Retrieves multiple sharded vaults by address
    ///
    /// This method fetches multiple vaults in a single request, up to a maximum of 100.
    ///
    /// # Parameters
    ///
    /// * `addresses` - An iterator of addresses to retrieve
    ///
    /// # Errors
    ///
    /// Returns [RpcError] if:
    /// - Any of the vaults don't exist;
    /// - The connection to the base layer provider fails; or
    /// - There's a protocol conversion error.
    pub async fn get_vaults(
        &self,
        addresses: impl IntoIterator<Item = Address>,
    ) -> Result<HashMap<Address, ShardedVault>, RpcError> {
        // Collect the iterator first since we'll need to consume it multiple times
        let addresses: Vec<Address> = addresses.into_iter().collect();
        if addresses.is_empty() {
            return Ok(HashMap::new())
        }

        let addresses_proto: Vec<_> = addresses.iter().map(|address| (*address).into()).collect();

        // Fetch vaults from server
        let vaults = self
            .inner
            .clone()
            .get_sharded_vaults(GetShardedVaultsRequest {
                addresses: addresses_proto,
            })
            .await
            .context(RequestSnafu)?
            .into_inner()
            .vaults;

        // Reconstruct addresses matching server response order
        addresses
            .into_iter()
            .zip(vaults)
            .map(|(address, vault_proto)| {
                vault_proto
                    .try_into()
                    .context(IntoProtoSnafu)
                    .map(|vault| (address, vault))
            })
            .collect()
    }

    /// Returns all the [OwnerId] that have a vault registered for the given shard on the base layer
    pub async fn get_domain_owner_ids(&self, shard: Shard) -> Result<Vec<OwnerId>, RpcError> {
        let response = self
            .inner
            .clone()
            .get_domain_owner_ids(GetDomainOwnerIdsRequest { shard })
            .await
            .context(RequestSnafu)?
            .into_inner();
        response
            .owner_ids
            .into_iter()
            .map(TryInto::try_into)
            .collect::<Result<Vec<OwnerId>, _>>()
            .context(HashSnafu)
            .context(IntoProtoSnafu)
    }

    /// Gets the Domain Agreement for the provided shard.
    ///
    /// # Parameters
    /// * `shard` - the shard to query the agreement for
    ///
    /// # Errors
    ///
    /// Returns [RpcError] if:
    /// - There's no active agreement;
    /// - The connection to the base layer RPC fails; or
    /// - There's a protocol conversion error.
    pub async fn get_domain_agreement(&self, shard: Shard) -> Result<DomainAgreement, RpcError> {
        self.inner
            .clone()
            .get_domain_agreement(GetDomainAgreementRequest { shard })
            .await
            .context(RequestSnafu)?
            .into_inner()
            .try_into()
            .context(IntoProtoSnafu)
    }

    /// Retrieves the current status of a transaction by its signature
    ///
    /// This method allows tracking the progress of a transaction through the
    /// delta network. After submitting a transaction, you can use this method
    /// to check if it has been processed, confirmed, or rejected.
    ///
    /// # Parameters
    ///
    /// * `tx_id` - [TxId] of the transaction to query
    ///
    /// # Errors
    ///
    /// Returns [RpcError] if:
    /// - The connection to the base layer provider fails; or
    /// - The signature is invalid or unknown to the network.
    ///
    /// # Note
    ///
    /// Transaction signatures are unique identifiers generated when signing a
    /// transaction. Retain the signature after submitting a transaction if you
    /// want to check its status later.
    pub async fn get_transaction_status(&self, tx_id: TxId) -> Result<TransactionStatus, RpcError> {
        let res = self
            .inner
            .clone()
            .get_transaction_status(GetTransactionStatusRequest {
                tx_id: tx_id.into(),
            })
            .await
            .context(RequestSnafu)?;
        res.into_inner().try_into().context(IntoProtoSnafu)
    }

    /// Retrieves the current epoch in delta
    ///
    /// This method returns the current epoch number. Epochs are protocol-defined
    /// periods in the delta protocol during which the base layer provider set
    /// remains constant.
    ///
    /// # Errors
    ///
    /// Returns [RpcError] if:
    /// - The connection to the base layer provider fails
    /// - The base layer provider cannot provide epoch information
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use delta_base_sdk::rpc::BaseRpcClient;
    ///
    /// async fn get_network_info() -> Result<(), Box<dyn std::error::Error>> {
    ///     let client = BaseRpcClient::new("http://localhost:50051").await?;
    ///
    ///     // Get information about the current epoch
    ///     let epoch = client.get_epoch().await?;
    ///
    ///     println!("Current epoch: {}", epoch);
    ///
    ///     Ok(())
    /// }
    /// ```
    pub async fn get_epoch(&self) -> Result<Epoch, RpcError> {
        let res: GetEpochResponse = self
            .inner
            .clone()
            .get_epoch(GetEpochRequest {})
            .await
            .context(RequestSnafu)?
            .into_inner();
        Ok(res.epoch)
    }

    /// Retrieves a [State Diff List (SDL)](StateDiffList) by its hash identifier
    ///
    /// A State Diff List (SDL) is a batch of state changes in the delta protocol.
    /// This method allows retrieving the complete contents of an SDL by its hash,
    /// enabling applications to inspect state changes or verify transaction results.
    ///
    /// # Parameters
    ///
    /// * `hash` - [Cryptographic hash](HashDigest) that uniquely identifies the SDL
    ///
    /// # Errors
    ///
    /// Returns [RpcError] if:
    /// - The SDL with the specified hash doesn't exist;
    /// - The connection to the base layer provider fails; or
    /// - There's a protocol conversion error.
    pub async fn get_sdl(&self, hash: HashDigest) -> Result<StateDiffList, RpcError> {
        let res = self
            .inner
            .clone()
            .get_sdl_data(GetSdlDataRequest {
                sdl_hash: hash.into(),
            })
            .await
            .context(RequestSnafu)?;
        res.into_inner().try_into().context(IntoProtoSnafu)
    }

    /// Retrieves the [status](SdlStatus) of a [State Diff List (SDL)](StateDiffList).
    ///
    /// # Parameters
    ///
    /// * `hash` - [Cryptographic hash](HashDigest) that uniquely identifies the SDL
    ///
    /// # Errors
    ///
    /// Returns [RpcError] if:
    /// - The SDL with the specified hash doesn't exist;
    /// - The connection to the base layer provider fails; or
    /// - There's a protocol conversion error.
    pub async fn get_sdl_status(&self, hash: HashDigest) -> Result<SdlStatus, RpcError> {
        let res = self
            .inner
            .clone()
            .get_sdl_status(GetSdlDataRequest {
                sdl_hash: hash.into(),
            })
            .await
            .context(RequestSnafu)?;

        res.into_inner().try_into().context(IntoProtoSnafu)
    }

    /// Subscribes to a stream of real-time events from the delta network
    ///
    /// This method establishes a streaming connection to receive events from the
    /// base layer as they occur. Events include epoch transitions, vault migrations,
    /// and other network state changes. The events are filtered by shard.
    ///
    /// # Parameters
    ///
    /// * `shard` - [Shard] number to receive events for
    ///
    /// # Errors
    ///
    /// Returns [RpcError] if:
    /// - The connection to the base layer provider fails;
    /// - The streaming request is rejected; or
    /// - Individual stream items may contain errors if there are issues during streaming.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use delta_base_sdk::{
    ///     events::BaseLayerEvent,
    ///     rpc::BaseRpcClient,
    /// };
    /// use tokio_stream::StreamExt;
    ///
    /// async fn monitor_events() -> Result<(), Box<dyn std::error::Error>> {
    ///     let client = BaseRpcClient::new("http://localhost:50051").await?;
    ///
    ///     // Subscribe to base layer events for shard 0 (the base layer shard)
    ///     let mut event_stream = client.stream_base_layer_events(0).await?;
    ///
    ///     println!("Listening for events...");
    ///
    ///     // Process events as they arrive
    ///     while let Some(event_result) = event_stream.next().await {
    ///         match event_result {
    ///             Ok(event) => {
    ///                 match event {
    ///                     BaseLayerEvent::NewEpoch(epoch) => {
    ///                         println!("New epoch started: {}", epoch);
    ///                     },
    ///                     BaseLayerEvent::SdlUpdate(update) => {
    ///                         println!("SDL update received: {:?}", update.hash);
    ///                     },
    ///                     BaseLayerEvent::VaultEmigrated(address) => {
    ///                         println!("Vault emigrated: {address:?}");
    ///                     },
    ///                     BaseLayerEvent::VaultImmigrated(address) => {
    ///                         println!("Vault immigrated: {address:?}");
    ///                     },
    ///                 }
    ///             },
    ///             Err(e) => {
    ///                 eprintln!("Error in event stream: {e}");
    ///                 // Decide whether to break or continue based on the error
    ///             }
    ///         }
    ///     }
    ///
    ///     Ok(())
    /// }
    /// ```
    ///
    /// # Note
    ///
    /// The returned stream implements the `Stream` trait from the `tokio_stream` crate.
    /// You need to bring the `StreamExt` trait into scope to use methods like `next()`.
    ///
    /// This is a long-lived connection that will continue to deliver events until
    /// the stream is dropped or the connection is closed. It's suitable for building
    /// reactive applications that need to respond to network events in real-time.
    pub async fn stream_base_layer_events(
        &self,
        shard: Shard,
    ) -> Result<impl Stream<Item = Result<BaseLayerEvent, RpcError>> + Send, RpcError> {
        let res = self
            .inner
            .clone()
            .stream_base_layer_events(StreamBaseLayerEventRequest { shard })
            .await
            .context(RequestSnafu)?;

        let stream = res.into_inner().map(|r| {
            r.context(RequestSnafu)
                .and_then(|u| u.try_into().context(IntoProtoSnafu))
        });

        Ok(stream)
    }
}

/// The default URI scheme to use when none is provided
const DEFAULT_SCHEME: Scheme = Scheme::HTTPS;

/// Adds the HTTPS scheme to a URL if it doesn't already have a scheme
///
/// This utility function ensures that all URLs have a scheme, adding HTTPS
/// as the default if none is specified. This allows users to provide just
/// the host and port (e.g., "localhost:50051") without having to specify
/// the scheme.
///
/// # Parameters
///
/// * `url` - URL to process, which may or may not have a scheme
///
/// # Errors
///
/// Returns [RpcError] if:
/// - The URL cannot be parsed as a valid URI
/// - The modified URI cannot be built
#[allow(clippy::result_large_err)]
fn add_missing_scheme(url: impl TryInto<Uri, Error = InvalidUri>) -> Result<Uri, RpcError> {
    let uri: Uri = url
        .try_into()
        .map_err(|e| e.into())
        .context(InvalidUrlSnafu)?;

    if uri.scheme().is_some() {
        Ok(uri)
    } else {
        let path = uri
            .path_and_query()
            .map(|p| p.as_str())
            .unwrap_or("/")
            .to_string();
        Into::<uri::Builder>::into(uri)
            .scheme(DEFAULT_SCHEME)
            .path_and_query(path)
            .build()
            .context(InvalidUrlSnafu)
    }
}

#[cfg(test)]
mod test {
    use super::*;

    #[test]
    fn test_add_missing_scheme() {
        // leaves existing scheme
        assert_eq!(
            "http://localhost:5005/",
            add_missing_scheme("http://localhost:5005")
                .unwrap()
                .to_string()
        );
        assert_eq!(
            "http://localhost:5005/path/yes",
            add_missing_scheme("http://localhost:5005/path/yes")
                .unwrap()
                .to_string()
        );
        assert_eq!(
            "yolo://net.delta.net/",
            add_missing_scheme("yolo://net.delta.net")
                .unwrap()
                .to_string()
        );

        // adds scheme if missing
        assert_eq!(
            "https://net.delta.net/",
            add_missing_scheme("net.delta.net").unwrap().to_string()
        );
        assert_eq!(
            "https://localhost:12345/",
            add_missing_scheme("localhost:12345").unwrap().to_string()
        );
    }
}