tape-sdk 0.4.4

//! High-level client for the Tapedrive storage network.

use std::sync::Arc;

use arc_swap::ArcSwap;
use tape_peer_http::{HttpApi, HttpApiBuilder};
use tape_peer_manager::PeerManager;
use tape_rpc::Rpc;
use tape_rpc_client::RpcClient;
use tape_api::program::tapedrive;
use tape_core::prelude::{CompressedTrack, StorageUnits};
use tape_core::types::coin::{SOL, TAPE};
use tape_core::types::ContentType;
use tape_crypto::prelude::{Address, Keypair};
use tape_protocol::{Api, ProtocolState};
use tokio::io::{AsyncRead, AsyncWrite};

use crate::balance::{sol_balance_of, tape_balance_of};
use crate::bootstrap::{BootstrapStore, Reputation};
use crate::error::TapedriveError;
use crate::keys::operator::TapeOperator;
use crate::keys::tape_key::TapeKey;
use crate::metrics::{Metrics, Noop, Operation, Phase, Timer};
use crate::read_options::ReadOptions;
use crate::write_options::WriteOptions;
use crate::stream::{
    read::{read_bytes, read_into},
    receipt::StreamReceipt,
    write::{write_bytes as write_stream_bytes, write_stream as write_reader_stream},
};
use crate::track::write::{write_or_resume, UNNAMED_TRACK, UNTYPED_TRACK};

/// High-level client for the Tapedrive storage network.
///
/// Generic over `Blockchain: Rpc` (on-chain) and `Cluster: Api` (storage nodes).
pub struct Tapedrive<Blockchain: Rpc, Cluster: Api> {
    pub state: ArcSwap<ProtocolState>,
    pub peer_manager: Arc<PeerManager>,
    pub api: Arc<Cluster>,
    pub rpc: Arc<RpcClient<Blockchain>>,
    pub payer: Option<Keypair>,
    pub metrics: Arc<dyn Metrics>,
    pub write_options: WriteOptions,
    pub read_options: ReadOptions,
    pub reputation: Arc<Reputation>,
}

/// Default constructor using `HttpApi`.
impl<Blockchain: Rpc> Tapedrive<Blockchain, HttpApi> {
    /// Create a new Tapedrive client.
    ///
    /// Takes an RPC backend and a payer keypair. Uses the default HTTP
    /// peer client for storage node communication.
    pub fn new(rpc: Blockchain, payer: Keypair) -> Self {
        Self::new_read_only(rpc).with_payer(payer)
    }

    /// Create a read-only client that presents a reader identity over mTLS.
    ///
    /// Nodes match a hidden track's share token against this identity's pubkey.
    pub fn new_with_identity(
        rpc: Blockchain,
        identity: Arc<Keypair>,
    ) -> Result<Self, tape_peer_tls::TlsError> {
        let rpc_client = Arc::new(RpcClient::from_rpc(rpc));
        let peer_manager = Arc::new(PeerManager::new());
        let api = Arc::new(
            HttpApiBuilder::new()
                .local_identity(identity)
                .build(peer_manager.clone())?,
        );

        Ok(Self::from_parts(
            ArcSwap::from_pointee(ProtocolState::default()),
            peer_manager,
            api,
            rpc_client,
            None,
        ))
    }

    /// Create a read-only Tapedrive client.
    pub fn new_read_only(rpc: Blockchain) -> Self {
        let rpc_client = Arc::new(RpcClient::from_rpc(rpc));
        let peer_manager = Arc::new(PeerManager::new());
        let api = Arc::new(HttpApi::with_default_timeouts(peer_manager.clone()));
        Self::from_parts(
            ArcSwap::from_pointee(ProtocolState::default()),
            peer_manager,
            api,
            rpc_client,
            None,
        )
    }
}

impl<Blockchain: Rpc, Cluster: Api> Tapedrive<Blockchain, Cluster> {
    /// Create a Tapedrive client from existing parts.
    pub fn from_parts(
        state: ArcSwap<ProtocolState>,
        peer_manager: Arc<PeerManager>,
        api: Arc<Cluster>,
        rpc: Arc<RpcClient<Blockchain>>,
        payer: Option<Keypair>,
    ) -> Self {
        Self {
            state,
            peer_manager,
            api,
            rpc,
            payer,
            metrics: Arc::new(Noop),
            write_options: WriteOptions::default(),
            read_options: ReadOptions::default(),
            reputation: Arc::new(Reputation::detached()),
        }
    }

    /// Replace the cached network state, freshly verified.
    ///
    /// The mark goes on before the store, so the state is never visible in the
    /// distrusted window between the two.
    pub fn store_state(&self, state: ProtocolState) {
        state.touch();
        self.state.store(Arc::new(state));
    }

    /// Share an existing reputation table rather than starting a fresh one.
    ///
    /// A long-lived caller that rebuilds a client per request must pass its own
    /// here, otherwise every request starts with an empty table and the peer
    /// ordering never learns anything.
    pub fn with_reputation(mut self, reputation: Arc<Reputation>) -> Self {
        self.reputation = reputation;
        self
    }

    /// Persist peer reputation and bootstrap hints to this store.
    pub fn with_bootstrap_cache(mut self, store: BootstrapStore) -> Self {
        self.reputation = Arc::new(Reputation::attach(store, tapedrive::id().into()));
        self
    }

    /// Attach or replace the payer used for mutating operations.
    pub fn with_payer(mut self, payer: Keypair) -> Self {
        self.payer = Some(payer);
        self
    }

    /// Replace the write concurrency knobs.
    pub fn with_write_options(mut self, options: WriteOptions) -> Self {
        self.write_options = options;
        self
    }

    /// Replace the read concurrency knobs.
    pub fn with_read_options(mut self, options: ReadOptions) -> Self {
        self.read_options = options;
        self
    }

    /// Attach or replace the metrics recorder.
    pub fn with_metrics(mut self, metrics: Arc<dyn Metrics>) -> Self {
        self.metrics = metrics;
        self
    }

    /// Access the underlying RPC client.
    pub fn rpc(&self) -> &RpcClient<Blockchain> {
        &self.rpc
    }

    /// Load the current protocol state (lock-free).
    pub fn state(&self) -> arc_swap::Guard<Arc<ProtocolState>> {
        self.state.load()
    }

    /// Return the payer keypair required for mutating operations.
    pub fn payer(&self) -> Result<&Keypair, TapedriveError> {
        self.payer.as_ref().ok_or(TapedriveError::MissingPayer)
    }

    /// The payer's SOL balance in lamports. A missing account reads as zero.
    pub async fn sol_balance(&self) -> Result<SOL, TapedriveError> {
        sol_balance_of(&self.rpc, &self.payer()?.address()).await
    }

    /// The payer's TAPE balance in flux. A missing token account reads as zero.
    pub async fn tape_balance(&self) -> Result<TAPE, TapedriveError> {
        tape_balance_of(&self.rpc, &self.payer()?.address()).await
    }

    pub(crate) fn timer(&self, operation: Operation, phase: Phase) -> Timer<'_> {
        Timer::start(self.metrics.as_ref(), operation, phase)
    }

    /// Write unnamed content-addressed data to the network in one call.
    ///
    /// Reserves the tape controlled by `tape_key` sized to fit `data`, registers
    /// a track, uploads erasure-coded slices to storage nodes, and certifies the
    /// track with BLS signatures. Unnamed tracks are excluded from object
    /// listings.
    ///
    /// The caller owns the tape key: generate and durably persist it before
    /// calling, so an interrupted write leaves the reserved tape recoverable.
    pub async fn write(
        &self,
        tape_key: &TapeKey,
        data: &[u8],
        epochs: u64,
    ) -> Result<CompressedTrack, TapedriveError> {
        self.write_named(
            tape_key,
            UNNAMED_TRACK,
            UNTYPED_TRACK,
            data,
            epochs,
        )
        .await
    }

    /// Write named data to the network in one call.
    ///
    /// Named tracks on non-system tapes are materialized into object listings.
    /// The caller owns the tape key, same as write.
    pub async fn write_named(
        &self,
        tape_key: &TapeKey,
        name: impl AsRef<[u8]>,
        content_type: ContentType,
        data: &[u8],
        epochs: u64,
    ) -> Result<CompressedTrack, TapedriveError> {
        let total = self
            .timer(Operation::Write, Phase::Total)
            .bytes(data.len() as u64);

        let result =
            write_or_resume(
                self,
                tape_key,
                name.as_ref(),
                content_type,
                data,
                epochs,
                self.write_options.visibility,
            )
            .await;
        total.finish_result(&result);

        result
    }

    /// Write unnamed in-memory bytes to an existing tape as a logical stream.
    ///
    /// Always writes a manifest track as the last track. For streams that fit
    /// in a single chunk, one data track and one manifest track are written.
    /// The manifest and chunks are excluded from object listings.
    pub async fn write_bytes(
        &self,
        tape_key: &TapeKey,
        data: &[u8],
    ) -> Result<StreamReceipt, TapedriveError> {
        self.write_named_bytes(
            tape_key,
            UNNAMED_TRACK,
            UNTYPED_TRACK,
            data,
        )
        .await
    }

    /// Write named in-memory bytes to an existing tape as a logical stream.
    ///
    /// The manifest track carries the object's name and content type; internal
    /// chunk tracks remain unnamed.
    pub async fn write_named_bytes(
        &self,
        tape_key: &TapeKey,
        name: impl AsRef<[u8]>,
        content_type: ContentType,
        data: &[u8],
    ) -> Result<StreamReceipt, TapedriveError> {
        self.write_named_bytes_as(tape_key, name, content_type, data)
            .await
    }

    /// Write named in-memory bytes as a stream.
    pub async fn write_named_bytes_as(
        &self,
        operator: &impl TapeOperator,
        name: impl AsRef<[u8]>,
        content_type: ContentType,
        data: &[u8],
    ) -> Result<StreamReceipt, TapedriveError> {
        let timer = self
            .timer(Operation::WriteStream, Phase::Total)
            .bytes(data.len() as u64);
        let result = write_stream_bytes(self, operator, name.as_ref(), content_type, data).await;
        timer.finish_result(&result);
        result
    }

    /// Write an unnamed byte stream from an async reader into an existing tape.
    ///
    /// The reader must yield exactly `size` bytes. The manifest and chunks are
    /// excluded from object listings.
    pub async fn write_stream<Reader: AsyncRead + Unpin>(
        &self,
        tape_key: &TapeKey,
        size: StorageUnits,
        reader: Reader,
    ) -> Result<StreamReceipt, TapedriveError> {
        self.write_named_stream(
            tape_key,
            UNNAMED_TRACK,
            UNTYPED_TRACK,
            size,
            reader,
        )
        .await
    }

    /// Write a named byte stream from an async reader into an existing tape.
    ///
    /// The manifest track carries the object's name and content type; internal
    /// chunk tracks remain unnamed.
    pub async fn write_named_stream<Reader: AsyncRead + Unpin>(
        &self,
        tape_key: &TapeKey,
        name: impl AsRef<[u8]>,
        content_type: ContentType,
        size: StorageUnits,
        reader: Reader,
    ) -> Result<StreamReceipt, TapedriveError> {
        self.write_named_stream_as(tape_key, name, content_type, size, reader)
            .await
    }

    /// Write a named byte stream from an async reader.
    pub async fn write_named_stream_as<Reader: AsyncRead + Unpin>(
        &self,
        operator: &impl TapeOperator,
        name: impl AsRef<[u8]>,
        content_type: ContentType,
        size: StorageUnits,
        reader: Reader,
    ) -> Result<StreamReceipt, TapedriveError> {
        let timer = self
            .timer(Operation::WriteStream, Phase::Total)
            .bytes(size.to_bytes());
        let result =
            write_reader_stream(self, operator, name.as_ref(), content_type, size, reader).await;
        timer.finish_result(&result);
        result
    }

    /// Read a stored stream by its manifest track address into memory.
    pub async fn read_bytes(
        &self,
        manifest: &Address,
    ) -> Result<Vec<u8>, TapedriveError> {
        let timer = self.timer(Operation::ReadStream, Phase::Total);
        let result = read_bytes(self, manifest).await;
        let timer = match &result {
            Ok(bytes) => timer.bytes(bytes.len() as u64),
            Err(_) => timer,
        };
        timer.finish_result(&result);
        result
    }

    /// Read a stored stream by its manifest track address into an async sink.
    pub async fn read_into<Writer: AsyncWrite + Unpin>(
        &self,
        manifest: &Address,
        writer: Writer,
    ) -> Result<(), TapedriveError> {
        let timer = self.timer(Operation::ReadStream, Phase::Total);
        let result = read_into(self, manifest, writer).await;
        timer.finish_result(&result);
        result
    }
}

#[cfg(test)]
mod tests {
    use rpc_litesvm::LiteSvmRpc;
    use tape_crypto::prelude::Keypair;

    use super::*;

    // fresh accounts read zero and an airdrop shows up in the sol balance
    #[tokio::test]
    async fn balances() {
        let rpc = LiteSvmRpc::new();
        let payer = Keypair::new(&mut rand::thread_rng());
        let address = payer.address();
        let client = Tapedrive::new(rpc.clone(), payer);

        assert_eq!(client.sol_balance().await.expect("sol balance"), SOL(0));
        assert_eq!(client.tape_balance().await.expect("tape balance"), TAPE(0));

        rpc.airdrop(&address.into(), 5_000_000_000).expect("airdrop");
        assert_eq!(
            client.sol_balance().await.expect("sol balance"),
            SOL(5_000_000_000)
        );
    }
}