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

// file: crates/ksp-onchain-transport-lib/src/grpc_unary.rs
// version: 1

const MAX_YELLOWSTONE_BLOCKHASH_LENGTH_BYTES: usize = 256;
const MAX_YELLOWSTONE_VERSION_LENGTH_BYTES: usize = 512;
const PATH_GET_BLOCK_HEIGHT: &str = "/geyser.Geyser/GetBlockHeight";
const PATH_GET_LATEST_BLOCKHASH: &str = "/geyser.Geyser/GetLatestBlockhash";
const PATH_GET_SLOT: &str = "/geyser.Geyser/GetSlot";
const PATH_GET_VERSION: &str = "/geyser.Geyser/GetVersion";
const PATH_IS_BLOCKHASH_VALID: &str = "/geyser.Geyser/IsBlockhashValid";
const PATH_PING: &str = "/geyser.Geyser/Ping";
const PATH_SUBSCRIBE_REPLAY_INFO: &str = "/geyser.Geyser/SubscribeReplayInfo";

/// Standard Solana Yellowstone unary client layered on one KSP-owned physical gRPC channel.
///
/// This N2 facade deliberately exposes no raw Tonic client and no upstream protobuf types. Streaming `Subscribe` is not part of `0.2.9-pre.003`.
#[derive(Clone)]
pub struct SolanaYellowstoneGrpcUnaryClient {
    channel: tonic::transport::Channel,
    metadata: std::vec::Vec<crate::YellowstoneGrpcMetadataEntry>,
    unary_timeout: std::time::Duration,
    max_inbound_message_size_bytes: usize,
    max_outbound_message_size_bytes: usize,
}

impl SolanaYellowstoneGrpcUnaryClient {
    /// Creates the internal unary facade state from a KSP-owned physical channel and validated endpoint settings.
    pub(crate) fn new(
        channel: tonic::transport::Channel,
        metadata: std::vec::Vec<crate::YellowstoneGrpcMetadataEntry>,
        settings: crate::YellowstoneGrpcSessionSettings,
    ) -> Self {
        return Self {
            channel,
            metadata,
            unary_timeout: settings.unary_timeout(),
            max_inbound_message_size_bytes: settings.max_inbound_message_size_bytes(),
            max_outbound_message_size_bytes: settings.max_outbound_message_size_bytes(),
        };
    }

    /// Returns the first slot retained by the endpoint for replay when advertised.
    pub async fn subscribe_replay_info(&self) -> ksp_core_lib::Result<crate::YellowstoneReplayInfo> {
        let response: yellowstone_grpc_proto::geyser::SubscribeReplayInfoResponse =
            match self.unary("SubscribeReplayInfo", PATH_SUBSCRIBE_REPLAY_INFO, yellowstone_grpc_proto::geyser::SubscribeReplayInfoRequest {}).await {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
        return std::result::Result::Ok(crate::YellowstoneReplayInfo::new(response.first_available));
    }

    /// Executes the standard Yellowstone unary `Ping` RPC and verifies the echoed count.
    pub async fn ping(&self, count: i32) -> ksp_core_lib::Result<crate::YellowstonePong> {
        let response: yellowstone_grpc_proto::geyser::PongResponse =
            match self.unary("Ping", PATH_PING, yellowstone_grpc_proto::geyser::PingRequest { count }).await {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
        if response.count != count {
            return invalid_unary_response("Ping", "Yellowstone Ping response count does not match the request");
        }
        return std::result::Result::Ok(crate::YellowstonePong::new(response.count));
    }

    /// Returns the latest blockhash advertised by the Yellowstone endpoint at an optional commitment.
    pub async fn get_latest_blockhash(
        &self,
        commitment: std::option::Option<crate::SolanaCommitment>,
    ) -> ksp_core_lib::Result<crate::YellowstoneLatestBlockhash> {
        let response: yellowstone_grpc_proto::geyser::GetLatestBlockhashResponse = match self
            .unary(
                "GetLatestBlockhash",
                PATH_GET_LATEST_BLOCKHASH,
                yellowstone_grpc_proto::geyser::GetLatestBlockhashRequest { commitment: commitment_to_wire(commitment) },
            )
            .await
        {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        if response.blockhash.is_empty() || response.blockhash.len() > MAX_YELLOWSTONE_BLOCKHASH_LENGTH_BYTES {
            return invalid_unary_response("GetLatestBlockhash", "Yellowstone latest blockhash response violates the KSP text bound");
        }
        return std::result::Result::Ok(crate::YellowstoneLatestBlockhash::new(response.slot, response.blockhash, response.last_valid_block_height));
    }

    /// Returns the current block height at an optional commitment.
    pub async fn get_block_height(&self, commitment: std::option::Option<crate::SolanaCommitment>) -> ksp_core_lib::Result<crate::YellowstoneBlockHeight> {
        let response: yellowstone_grpc_proto::geyser::GetBlockHeightResponse = match self
            .unary(
                "GetBlockHeight",
                PATH_GET_BLOCK_HEIGHT,
                yellowstone_grpc_proto::geyser::GetBlockHeightRequest { commitment: commitment_to_wire(commitment) },
            )
            .await
        {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        return std::result::Result::Ok(crate::YellowstoneBlockHeight::new(response.block_height));
    }

    /// Returns the current slot at an optional commitment.
    pub async fn get_slot(&self, commitment: std::option::Option<crate::SolanaCommitment>) -> ksp_core_lib::Result<crate::YellowstoneSlot> {
        let response: yellowstone_grpc_proto::geyser::GetSlotResponse =
            match self.unary("GetSlot", PATH_GET_SLOT, yellowstone_grpc_proto::geyser::GetSlotRequest { commitment: commitment_to_wire(commitment) }).await {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
        return std::result::Result::Ok(crate::YellowstoneSlot::new(response.slot));
    }

    /// Checks one opaque Solana blockhash at an optional commitment.
    pub async fn is_blockhash_valid(
        &self,
        blockhash: impl std::convert::Into<std::string::String>,
        commitment: std::option::Option<crate::SolanaCommitment>,
    ) -> ksp_core_lib::Result<crate::YellowstoneBlockhashValidity> {
        let blockhash = blockhash.into();
        if blockhash.is_empty() || blockhash.len() > MAX_YELLOWSTONE_BLOCKHASH_LENGTH_BYTES || blockhash.trim() != blockhash {
            return std::result::Result::Err(
                ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RPC_PARAMETERS, "Yellowstone blockhash must be non-empty, bounded and untrimmed")
                    .with_context("grpc_operation", "IsBlockhashValid")
                    .with_context("field", "blockhash"),
            );
        }
        let response: yellowstone_grpc_proto::geyser::IsBlockhashValidResponse = match self
            .unary(
                "IsBlockhashValid",
                PATH_IS_BLOCKHASH_VALID,
                yellowstone_grpc_proto::geyser::IsBlockhashValidRequest { blockhash, commitment: commitment_to_wire(commitment) },
            )
            .await
        {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        return std::result::Result::Ok(crate::YellowstoneBlockhashValidity::new(response.slot, response.valid));
    }

    /// Returns the endpoint's Yellowstone/validator version string after applying a deterministic KSP text bound.
    pub async fn get_version(&self) -> ksp_core_lib::Result<crate::YellowstoneVersionInfo> {
        let response: yellowstone_grpc_proto::geyser::GetVersionResponse =
            match self.unary("GetVersion", PATH_GET_VERSION, yellowstone_grpc_proto::geyser::GetVersionRequest {}).await {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
        if response.version.is_empty() || response.version.len() > MAX_YELLOWSTONE_VERSION_LENGTH_BYTES {
            return invalid_unary_response("GetVersion", "Yellowstone version response violates the KSP text bound");
        }
        return std::result::Result::Ok(crate::YellowstoneVersionInfo::new(response.version));
    }

    async fn unary<Request, Response>(&self, operation: &'static str, path: &'static str, message: Request) -> ksp_core_lib::Result<Response>
    where
        Request: tonic_prost::prost::Message + Send + Sync + 'static,
        Response: tonic_prost::prost::Message + std::default::Default + Send + Sync + 'static,
    {
        let path = match path.parse::<http::uri::PathAndQuery>() {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => {
                return std::result::Result::Err(
                    ksp_core_lib::Error::new(crate::ERROR_CODE_GRPC_CHANNEL_FAILED, "internal Yellowstone gRPC method path is invalid")
                        .with_context("grpc_operation", operation),
                );
            },
        };
        let mut grpc = tonic::client::Grpc::new(self.channel.clone())
            .max_decoding_message_size(self.max_inbound_message_size_bytes)
            .max_encoding_message_size(self.max_outbound_message_size_bytes);
        let future = async {
            if grpc.ready().await.is_err() {
                return std::result::Result::Err(
                    ksp_core_lib::Error::new(crate::ERROR_CODE_GRPC_CHANNEL_FAILED, "Yellowstone gRPC channel is not ready for unary dispatch")
                        .with_context("grpc_operation", operation),
                );
            }
            let mut request = tonic::Request::new(message);
            for entry in &self.metadata {
                if let std::result::Result::Err(error) = entry.append_to(request.metadata_mut()) {
                    return std::result::Result::Err(error);
                }
            }
            let response = match grpc.unary(request, path, tonic_prost::ProstCodec::<Request, Response>::default()).await {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(status) => return std::result::Result::Err(grpc_status_error(operation, status)),
            };
            return std::result::Result::Ok(response.into_inner());
        };
        return match tokio::time::timeout(self.unary_timeout, future).await {
            std::result::Result::Ok(result) => result,
            std::result::Result::Err(_) => std::result::Result::Err(
                ksp_core_lib::Error::new(crate::ERROR_CODE_TIMEOUT, "Yellowstone gRPC unary operation exceeded the KSP deadline")
                    .with_context("grpc_operation", operation),
            ),
        };
    }
}

impl std::fmt::Debug for SolanaYellowstoneGrpcUnaryClient {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        return formatter
            .debug_struct("SolanaYellowstoneGrpcUnaryClient")
            .field("metadata_count", &self.metadata.len())
            .field("unary_timeout", &self.unary_timeout)
            .field("max_inbound_message_size_bytes", &self.max_inbound_message_size_bytes)
            .field("max_outbound_message_size_bytes", &self.max_outbound_message_size_bytes)
            .field("channel", &"<private>")
            .finish();
    }
}

/// Replay availability advertised by `SubscribeReplayInfo`.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct YellowstoneReplayInfo {
    first_available: std::option::Option<u64>,
}

impl YellowstoneReplayInfo {
    /// Creates a replay-info projection.
    #[must_use]
    pub const fn new(first_available: std::option::Option<u64>) -> Self {
        return Self { first_available };
    }

    /// Returns the first replayable slot when the endpoint advertises one.
    #[must_use]
    pub const fn first_available(&self) -> std::option::Option<u64> {
        return self.first_available;
    }
}

/// Echo returned by the Yellowstone unary `Ping` RPC.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct YellowstonePong {
    count: i32,
}

impl YellowstonePong {
    /// Creates a ping response projection.
    #[must_use]
    pub const fn new(count: i32) -> Self {
        return Self { count };
    }

    /// Returns the echoed ping count.
    #[must_use]
    pub const fn count(&self) -> i32 {
        return self.count;
    }
}

/// Latest blockhash response returned by Yellowstone.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct YellowstoneLatestBlockhash {
    slot: u64,
    blockhash: std::string::String,
    last_valid_block_height: u64,
}

impl YellowstoneLatestBlockhash {
    /// Creates a latest-blockhash projection.
    #[must_use]
    pub fn new(slot: u64, blockhash: std::string::String, last_valid_block_height: u64) -> Self {
        return Self { slot, blockhash, last_valid_block_height };
    }

    /// Returns the response slot.
    #[must_use]
    pub const fn slot(&self) -> u64 {
        return self.slot;
    }

    /// Returns the opaque Solana blockhash text.
    #[must_use]
    pub fn blockhash(&self) -> &str {
        return self.blockhash.as_str();
    }

    /// Returns the last valid block height associated with the blockhash.
    #[must_use]
    pub const fn last_valid_block_height(&self) -> u64 {
        return self.last_valid_block_height;
    }
}

/// Current block height returned by Yellowstone.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct YellowstoneBlockHeight {
    block_height: u64,
}

impl YellowstoneBlockHeight {
    /// Creates a block-height projection.
    #[must_use]
    pub const fn new(block_height: u64) -> Self {
        return Self { block_height };
    }

    /// Returns the block height.
    #[must_use]
    pub const fn block_height(&self) -> u64 {
        return self.block_height;
    }
}

/// Current slot returned by Yellowstone.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct YellowstoneSlot {
    slot: u64,
}

impl YellowstoneSlot {
    /// Creates a slot projection.
    #[must_use]
    pub const fn new(slot: u64) -> Self {
        return Self { slot };
    }

    /// Returns the slot.
    #[must_use]
    pub const fn slot(&self) -> u64 {
        return self.slot;
    }
}

/// Result of checking one blockhash with Yellowstone.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct YellowstoneBlockhashValidity {
    slot: u64,
    valid: bool,
}

impl YellowstoneBlockhashValidity {
    /// Creates a blockhash-validity projection.
    #[must_use]
    pub const fn new(slot: u64, valid: bool) -> Self {
        return Self { slot, valid };
    }

    /// Returns the response slot.
    #[must_use]
    pub const fn slot(&self) -> u64 {
        return self.slot;
    }

    /// Returns whether the blockhash is valid at the requested commitment.
    #[must_use]
    pub const fn valid(&self) -> bool {
        return self.valid;
    }
}

/// Bounded Yellowstone endpoint version information.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct YellowstoneVersionInfo {
    version: std::string::String,
}

impl YellowstoneVersionInfo {
    /// Creates a version projection.
    #[must_use]
    pub fn new(version: std::string::String) -> Self {
        return Self { version };
    }

    /// Returns the bounded endpoint version string.
    #[must_use]
    pub fn version(&self) -> &str {
        return self.version.as_str();
    }
}

fn commitment_to_wire(commitment: std::option::Option<crate::SolanaCommitment>) -> std::option::Option<i32> {
    return commitment.map(|value| {
        return match value {
            crate::SolanaCommitment::Processed => yellowstone_grpc_proto::geyser::CommitmentLevel::Processed as i32,
            crate::SolanaCommitment::Confirmed => yellowstone_grpc_proto::geyser::CommitmentLevel::Confirmed as i32,
            crate::SolanaCommitment::Finalized => yellowstone_grpc_proto::geyser::CommitmentLevel::Finalized as i32,
        };
    });
}

fn grpc_status_error(operation: &'static str, status: tonic::Status) -> ksp_core_lib::Error {
    let code = status.code().to_string();
    ksp_logging_lib::warn!(target: crate::TRACING_TARGET, grpc_operation = operation, grpc_code = code.as_str(), "Yellowstone gRPC unary endpoint returned a status");
    return ksp_core_lib::Error::new(crate::ERROR_CODE_GRPC_STATUS, "Yellowstone gRPC endpoint returned a gRPC status")
        .with_context("grpc_operation", operation)
        .with_context("grpc_code", code);
}

fn invalid_unary_response<T>(operation: &'static str, message: &str) -> ksp_core_lib::Result<T> {
    return std::result::Result::Err(ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, message).with_context("grpc_operation", operation));
}

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