ark_grpc/

client.rs

use crate::generated;
use crate::generated::ark::v1::ark_service_client::ArkServiceClient;
use crate::generated::ark::v1::get_subscription_response;
use crate::generated::ark::v1::indexer_service_client::IndexerServiceClient;
use crate::generated::ark::v1::indexer_tx_history_record::Key;
use crate::generated::ark::v1::ConfirmRegistrationRequest;
use crate::generated::ark::v1::EstimateIntentFeeRequest;
use crate::generated::ark::v1::GetEventStreamRequest;
use crate::generated::ark::v1::GetInfoRequest;
use crate::generated::ark::v1::GetSubscriptionRequest;
use crate::generated::ark::v1::GetTransactionsStreamRequest;
use crate::generated::ark::v1::IndexerChainedTxType;
use crate::generated::ark::v1::Intent;
use crate::generated::ark::v1::Outpoint;
use crate::generated::ark::v1::RegisterIntentRequest;
use crate::generated::ark::v1::SubmitSignedForfeitTxsRequest;
use crate::generated::ark::v1::SubmitTreeNoncesRequest;
use crate::generated::ark::v1::SubmitTreeSignaturesRequest;
use crate::generated::ark::v1::SubscribeForScriptsRequest;
use crate::generated::ark::v1::UnsubscribeForScriptsRequest;
use crate::Error;
use ark_core::asset::AssetId;
use ark_core::history;
use ark_core::server::parse_sequence_number;
use ark_core::server::ArkTransaction;
use ark_core::server::AssetInfo;
use ark_core::server::BatchFailed;
use ark_core::server::BatchFinalizationEvent;
use ark_core::server::BatchFinalizedEvent;
use ark_core::server::BatchStartedEvent;
use ark_core::server::BatchTreeEventType;
use ark_core::server::ChainedTxType;
use ark_core::server::CommitmentTransaction;
use ark_core::server::FinalizeOffchainTxResponse;
use ark_core::server::GetVtxosRequest;
use ark_core::server::GetVtxosRequestFilter;
use ark_core::server::GetVtxosRequestReference;
use ark_core::server::IndexerPage;
use ark_core::server::Info;
use ark_core::server::NoncePks;
use ark_core::server::PartialSigTree;
use ark_core::server::StreamEvent;
use ark_core::server::StreamStartedEvent;
use ark_core::server::StreamTransactionData;
use ark_core::server::SubmitOffchainTxResponse;
use ark_core::server::SubscriptionEvent;
use ark_core::server::SubscriptionResponse;
use ark_core::server::TreeNoncesAggregatedEvent;
use ark_core::server::TreeNoncesEvent;
use ark_core::server::TreeSignatureEvent;
use ark_core::server::TreeSigningStartedEvent;
use ark_core::server::TreeTxEvent;
use ark_core::server::TreeTxNoncePks;
use ark_core::server::VirtualTxOutPoint;
use ark_core::server::VirtualTxsResponse;
use ark_core::server::VtxoChain;
use ark_core::server::VtxoChains;
use ark_core::server::SDK_VERSION;
use ark_core::server::TARGET_ARKD_VERSION;
use ark_core::ArkAddress;
use ark_core::TxGraphChunk;
use async_stream::stream;
use base64::Engine;
use bitcoin::hex::FromHex;
use bitcoin::secp256k1::PublicKey;
use bitcoin::taproot::Signature;
use bitcoin::OutPoint;
use bitcoin::Psbt;
use bitcoin::ScriptBuf;
use bitcoin::SignedAmount;
use bitcoin::Transaction;
use bitcoin::Txid;
use futures::Future;
use futures::Stream;
use futures::StreamExt;
use futures::TryStreamExt;
use std::collections::HashMap;
use std::error::Error as StdError;
use std::fmt;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::RwLock;

#[derive(Clone, Default)]
struct HeaderState {
    digest: Arc<RwLock<Option<String>>>,
}

impl HeaderState {
    fn set_digest(&self, digest: String) {
        let digest = (!digest.is_empty()).then_some(digest);
        match self.digest.write() {
            Ok(mut guard) => *guard = digest,
            Err(poisoned) => {
                log::warn!("digest header state lock poisoned while updating; recovering");
                *poisoned.into_inner() = digest;
            }
        }
    }

    fn digest(&self) -> Option<String> {
        match self.digest.read() {
            Ok(guard) => guard.clone(),
            Err(poisoned) => {
                log::warn!("digest header state lock poisoned while reading; recovering");
                poisoned.into_inner().clone()
            }
        }
    }
}

#[derive(Clone, Default)]
struct HeaderInterceptor {
    state: HeaderState,
}

impl tonic::service::Interceptor for HeaderInterceptor {
    fn call(&mut self, mut req: tonic::Request<()>) -> Result<tonic::Request<()>, tonic::Status> {
        let metadata = req.metadata_mut();
        metadata.insert(
            "x-build-version",
            tonic::metadata::MetadataValue::from_static(TARGET_ARKD_VERSION),
        );
        metadata.insert(
            "x-sdk-version",
            tonic::metadata::MetadataValue::from_static(SDK_VERSION),
        );

        if let Some(digest) = self.state.digest() {
            if let Ok(value) = tonic::metadata::MetadataValue::try_from(digest.as_str()) {
                metadata.insert("x-digest", value);
            }
        }

        Ok(req)
    }
}

type InterceptedChannel =
    tonic::codegen::InterceptedService<tonic::transport::Channel, HeaderInterceptor>;

type InfoRefreshHook =
    Arc<dyn Fn(Info) -> Result<(), Box<dyn StdError + Send + Sync + 'static>> + Send + Sync>;

#[derive(Clone, Default)]
struct SharedState {
    headers: HeaderState,
    info_refresh_hook: Arc<RwLock<Option<InfoRefreshHook>>>,
}

impl SharedState {
    fn set_info_refresh_hook(&self, hook: InfoRefreshHook) {
        match self.info_refresh_hook.write() {
            Ok(mut guard) => *guard = Some(hook),
            Err(poisoned) => {
                log::warn!("info refresh hook lock poisoned while updating; recovering");
                *poisoned.into_inner() = Some(hook);
            }
        }
    }

    /// Runs one RPC operation with digest-mismatch handling.
    ///
    /// If the server rejects the operation because our cached `/info` digest is stale,
    /// this fetches fresh `/info` through the supplied Ark client, runs the refresh hook,
    /// commits the new digest header, and returns `ServerInfoChanged` for the original
    /// failure. The original operation is never retried automatically.
    async fn guarded<T>(
        &self,
        info_client: ArkServiceClient<InterceptedChannel>,
        op: impl Future<Output = Result<T, Error>>,
    ) -> Result<T, Error> {
        match op.await {
            Ok(value) => Ok(value),
            Err(err) if err.is_digest_mismatch() => {
                let original = err;
                let info = self.fetch_info_unguarded(info_client).await?;
                let digest = info.digest.clone();

                if let Some(hook) = self.info_refresh_hook() {
                    hook(info).map_err(Error::conversion)?;
                }

                self.headers.set_digest(digest);
                Err(Error::server_info_changed(original))
            }
            Err(err) => Err(err),
        }
    }

    fn info_refresh_hook(&self) -> Option<InfoRefreshHook> {
        match self.info_refresh_hook.read() {
            Ok(hook) => hook.clone(),
            Err(poisoned) => {
                log::warn!("info refresh hook lock poisoned while reading; recovering");
                poisoned.into_inner().clone()
            }
        }
    }

    async fn get_info_unguarded(
        &self,
        client: ArkServiceClient<InterceptedChannel>,
    ) -> Result<Info, Error> {
        let info = self.fetch_info_unguarded(client).await?;
        self.headers.set_digest(info.digest.clone());
        Ok(info)
    }

    async fn fetch_info_unguarded(
        &self,
        mut client: ArkServiceClient<InterceptedChannel>,
    ) -> Result<Info, Error> {
        let response = client
            .get_info(GetInfoRequest {})
            .await
            .map_err(Error::request)?;

        response.into_inner().try_into()
    }
}

#[derive(Clone)]
pub struct Client {
    url: String,
    ark: Option<guarded::Ark>,
    indexer: Option<guarded::Indexer>,
    shared: SharedState,
}

impl fmt::Debug for Client {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Client")
            .field("url", &self.url)
            .field("connected", &self.ark.is_some())
            .finish()
    }
}

impl Client {
    pub fn new(url: String) -> Self {
        Self {
            url,
            ark: None,
            indexer: None,
            shared: SharedState::default(),
        }
    }

    pub fn set_info_refresh_hook(
        &mut self,
        hook: impl Fn(Info) -> Result<(), Box<dyn StdError + Send + Sync + 'static>>
            + Send
            + Sync
            + 'static,
    ) {
        self.shared.set_info_refresh_hook(Arc::new(hook));
    }

    pub async fn connect(&mut self) -> Result<(), Error> {
        let endpoint =
            tonic::transport::Endpoint::from_shared(self.url.clone()).map_err(Error::connect)?;

        #[cfg(any(feature = "tls-webpki-roots", feature = "tls-native-roots"))]
        let endpoint = {
            let tls = tonic::transport::ClientTlsConfig::new();
            #[cfg(feature = "tls-webpki-roots")]
            let tls = tls.with_webpki_roots();
            #[cfg(feature = "tls-native-roots")]
            let tls = tls.with_native_roots();
            endpoint.tls_config(tls).map_err(Error::connect)?
        };

        let channel = endpoint.connect().await.map_err(Error::connect)?;

        let interceptor = HeaderInterceptor {
            state: self.shared.headers.clone(),
        };
        let ark_service_client =
            ArkServiceClient::with_interceptor(channel.clone(), interceptor.clone());
        let indexer_client = IndexerServiceClient::with_interceptor(channel, interceptor);

        self.ark = Some(guarded::Ark::new(
            ark_service_client.clone(),
            self.shared.clone(),
        ));
        self.indexer = Some(guarded::Indexer::new(
            indexer_client,
            ark_service_client,
            self.shared.clone(),
        ));
        Ok(())
    }

    pub async fn get_info(&self) -> Result<Info, Error> {
        self.ark()?.get_info().await
    }

    /// List VTXOs with pagination support.
    /// Returns a single page of results along with pagination info.
    pub async fn list_vtxos(&self, request: GetVtxosRequest) -> Result<ListVtxosResponse, Error> {
        if request.reference().is_empty() {
            return Ok(ListVtxosResponse {
                vtxos: Vec::new(),
                page: None,
            });
        }

        let response = self
            .indexer()?
            .request(move |mut client| async move {
                client
                    .get_vtxos(generated::ark::v1::GetVtxosRequest::from(request))
                    .await
            })
            .await?;

        let inner = response.into_inner();

        let vtxos = inner
            .vtxos
            .iter()
            .map(VirtualTxOutPoint::try_from)
            .collect::<Result<Vec<_>, _>>()?;

        let page = inner
            .page
            .map(IndexerPage::try_from)
            .transpose()
            .map_err(Error::conversion)?;

        Ok(ListVtxosResponse { vtxos, page })
    }

    pub async fn register_intent(&self, intent: ark_core::intent::Intent) -> Result<String, Error> {
        let intent = intent.try_into()?;
        let request = RegisterIntentRequest {
            intent: Some(intent),
        };

        let response = self
            .ark()?
            .request(move |mut client| async move { client.register_intent(request).await })
            .await?;

        let intent_id = response.into_inner().intent_id;

        Ok(intent_id)
    }

    pub async fn submit_offchain_transaction_request(
        &self,
        ark_tx: Psbt,
        checkpoint_txs: Vec<Psbt>,
    ) -> Result<SubmitOffchainTxResponse, Error> {
        let base64 = base64::engine::GeneralPurpose::new(
            &base64::alphabet::STANDARD,
            base64::engine::GeneralPurposeConfig::new(),
        );

        let ark_tx = base64.encode(ark_tx.serialize());

        let checkpoint_txs = checkpoint_txs
            .into_iter()
            .map(|tx| base64.encode(tx.serialize()))
            .collect();

        let res = self
            .ark()?
            .request(move |mut client| async move {
                client
                    .submit_tx(generated::ark::v1::SubmitTxRequest {
                        signed_ark_tx: ark_tx,
                        checkpoint_txs,
                    })
                    .await
            })
            .await?;

        let res = res.into_inner();

        let signed_ark_tx = res.final_ark_tx;
        let signed_ark_tx = base64.decode(signed_ark_tx).map_err(Error::conversion)?;
        let signed_ark_tx = Psbt::deserialize(&signed_ark_tx).map_err(Error::conversion)?;

        let signed_checkpoint_txs = res
            .signed_checkpoint_txs
            .into_iter()
            .map(|tx| {
                let tx = base64.decode(tx).map_err(Error::conversion)?;
                let tx = Psbt::deserialize(&tx).map_err(Error::conversion)?;

                Ok(tx)
            })
            .collect::<Result<Vec<_>, Error>>()?;

        Ok(SubmitOffchainTxResponse {
            signed_ark_tx,
            signed_checkpoint_txs,
        })
    }

    pub async fn finalize_offchain_transaction(
        &self,
        txid: Txid,
        checkpoint_txs: Vec<Psbt>,
    ) -> Result<FinalizeOffchainTxResponse, Error> {
        let base64 = base64::engine::GeneralPurpose::new(
            &base64::alphabet::STANDARD,
            base64::engine::GeneralPurposeConfig::new(),
        );

        let checkpoint_txs = checkpoint_txs
            .into_iter()
            .map(|tx| base64.encode(tx.serialize()))
            .collect();

        self.ark()?
            .request(move |mut client| async move {
                client
                    .finalize_tx(generated::ark::v1::FinalizeTxRequest {
                        ark_txid: txid.to_string(),
                        final_checkpoint_txs: checkpoint_txs,
                    })
                    .await
            })
            .await?;

        Ok(FinalizeOffchainTxResponse {})
    }

    pub async fn get_pending_tx(
        &self,
        intent: ark_core::intent::Intent,
    ) -> Result<Vec<ark_core::server::PendingTx>, Error> {
        let intent: Intent = intent.try_into()?;

        let res = self
            .ark()?
            .request(move |mut client| async move {
                client
                    .get_pending_tx(generated::ark::v1::GetPendingTxRequest {
                        identifier: Some(
                            generated::ark::v1::get_pending_tx_request::Identifier::Intent(intent),
                        ),
                    })
                    .await
            })
            .await?;

        let inner = res.into_inner();
        let base64 = base64::engine::GeneralPurpose::new(
            &base64::alphabet::STANDARD,
            base64::engine::GeneralPurposeConfig::new(),
        );

        inner
            .pending_txs
            .into_iter()
            .map(|tx| {
                let ark_txid = tx.ark_txid.parse().map_err(Error::conversion)?;

                let signed_ark_tx = base64.decode(&tx.final_ark_tx).map_err(Error::conversion)?;
                let signed_ark_tx = Psbt::deserialize(&signed_ark_tx).map_err(Error::conversion)?;

                let signed_checkpoint_txs = tx
                    .signed_checkpoint_txs
                    .into_iter()
                    .map(|cp| {
                        let bytes = base64.decode(cp).map_err(Error::conversion)?;
                        Psbt::deserialize(&bytes).map_err(Error::conversion)
                    })
                    .collect::<Result<Vec<_>, Error>>()?;

                Ok(ark_core::server::PendingTx {
                    ark_txid,
                    signed_ark_tx,
                    signed_checkpoint_txs,
                })
            })
            .collect()
    }

    pub async fn confirm_registration(&self, intent_id: String) -> Result<(), Error> {
        self.ark()?
            .request(move |mut client| async move {
                client
                    .confirm_registration(ConfirmRegistrationRequest { intent_id })
                    .await
            })
            .await?;

        Ok(())
    }

    pub async fn submit_tree_nonces(
        &self,
        batch_id: &str,
        cosigner_pubkey: PublicKey,
        pub_nonce_tree: NoncePks,
    ) -> Result<(), Error> {
        self.ark()?
            .request(move |mut client| async move {
                client
                    .submit_tree_nonces(SubmitTreeNoncesRequest {
                        batch_id: batch_id.to_string(),
                        pubkey: cosigner_pubkey.to_string(),
                        tree_nonces: pub_nonce_tree.encode(),
                    })
                    .await
            })
            .await?;

        Ok(())
    }

    pub async fn submit_tree_signatures(
        &self,
        batch_id: &str,
        cosigner_pk: PublicKey,
        partial_sig_tree: PartialSigTree,
    ) -> Result<(), Error> {
        self.ark()?
            .request(move |mut client| async move {
                client
                    .submit_tree_signatures(SubmitTreeSignaturesRequest {
                        batch_id: batch_id.to_string(),
                        pubkey: cosigner_pk.to_string(),
                        tree_signatures: partial_sig_tree.encode(),
                    })
                    .await
            })
            .await?;

        Ok(())
    }

    pub async fn submit_signed_forfeit_txs(
        &self,
        signed_forfeit_txs: Vec<Psbt>,
        signed_commitment_tx: Option<Psbt>,
    ) -> Result<(), Error> {
        let base64 = base64::engine::GeneralPurpose::new(
            &base64::alphabet::STANDARD,
            base64::engine::GeneralPurposeConfig::new(),
        );

        let signed_commitment_tx = signed_commitment_tx
            .map(|tx| base64.encode(tx.serialize()))
            .unwrap_or_default();

        self.ark()?
            .request(move |mut client| async move {
                client
                    .submit_signed_forfeit_txs(SubmitSignedForfeitTxsRequest {
                        signed_forfeit_txs: signed_forfeit_txs
                            .iter()
                            .map(|psbt| base64.encode(psbt.serialize()))
                            .collect(),
                        signed_commitment_tx,
                    })
                    .await
            })
            .await?;

        Ok(())
    }

    pub async fn get_event_stream(
        &self,
        topics: Vec<String>,
    ) -> Result<impl Stream<Item = Result<StreamEvent, Error>> + Unpin, Error> {
        let response = self
            .ark()?
            .request(move |mut client| async move {
                client
                    .get_event_stream(GetEventStreamRequest { topics })
                    .await
            })
            .await?;
        let mut stream = response.into_inner();

        let stream = stream! {
            loop {
                match stream.try_next().await {
                    Ok(Some(event)) => match event.event {
                        None => {
                            log::debug!("Got empty message");
                        }
                        Some(event) => {
                            yield Ok(StreamEvent::try_from(event)?);
                        }
                    },
                    Ok(None) => {
                        yield Err(Error::event_stream_disconnect());
                    }
                    Err(e) => {
                        yield Err(Error::event_stream(e));
                    }
                }
            }
        };

        Ok(stream.boxed())
    }

    pub async fn get_tx_stream(
        &self,
    ) -> Result<impl Stream<Item = Result<StreamTransactionData, Error>> + Unpin, Error> {
        let response = self
            .ark()?
            .request(|mut client| async move {
                client
                    .get_transactions_stream(GetTransactionsStreamRequest {})
                    .await
            })
            .await?;

        let mut stream = response.into_inner();

        let stream = stream! {
            loop {
                match stream.try_next().await {
                    Ok(Some(event)) => match event.data {
                        None => {
                            log::debug!("Got empty message");
                        }
                        Some(event) => {
                            yield Ok(StreamTransactionData::try_from(event)?);
                        }
                    },
                    Ok(None) => {
                        yield Err(Error::event_stream_disconnect());
                    }
                    Err(e) => {
                        yield Err(Error::event_stream(e));
                    }
                }
            }
        };

        Ok(stream.boxed())
    }

    pub async fn get_vtxo_chain(
        &self,
        outpoint: Option<OutPoint>,
        size_and_index: Option<(i32, i32)>,
    ) -> Result<VtxoChainResponse, Error> {
        let response = self
            .indexer()?
            .request(move |mut client| async move {
                client
                    .get_vtxo_chain(generated::ark::v1::GetVtxoChainRequest {
                        outpoint: outpoint.map(|o| generated::ark::v1::IndexerOutpoint {
                            txid: o.txid.to_string(),
                            vout: o.vout,
                        }),
                        page: size_and_index.map(|(size, index)| {
                            generated::ark::v1::IndexerPageRequest { size, index }
                        }),
                    })
                    .await
            })
            .await?;
        let response = response.into_inner();
        let result = response.try_into()?;
        Ok(result)
    }

    pub async fn get_virtual_txs(
        &self,
        txids: Vec<String>,
        size_and_index: Option<(i32, i32)>,
    ) -> Result<VirtualTxsResponse, Error> {
        let response = self
            .indexer()?
            .request(move |mut client| async move {
                client
                    .get_virtual_txs(generated::ark::v1::GetVirtualTxsRequest {
                        txids,
                        page: size_and_index.map(|(size, index)| {
                            generated::ark::v1::IndexerPageRequest { size, index }
                        }),
                    })
                    .await
            })
            .await?;
        let response = response.into_inner();
        let result = response.try_into()?;
        Ok(result)
    }

    /// Allows to subscribe for tx notifications related to the provided
    /// vtxo scripts.
    ///
    /// It can also be used to update an existing subscriptions by adding
    /// new scripts to it.
    ///
    /// Note: for new subscriptions, don't provide a `subscription_id`
    ///
    /// Returns the subscription id if successful
    pub async fn subscribe_to_scripts(
        &self,
        scripts: Vec<ArkAddress>,
        subscription_id: Option<String>,
    ) -> Result<String, Error> {
        let scripts = scripts
            .iter()
            .map(|address| address.to_p2tr_script_pubkey().to_hex_string())
            .collect::<Vec<_>>();

        // For new subscription we expect empty string ("") here
        let subscription_id = subscription_id.unwrap_or_default();

        let response = self
            .indexer()?
            .request(move |mut client| async move {
                client
                    .subscribe_for_scripts(SubscribeForScriptsRequest {
                        scripts,
                        subscription_id,
                    })
                    .await
            })
            .await?;

        let response = response.into_inner();

        Ok(response.subscription_id)
    }

    /// Allows to remove scripts from an existing subscription.
    pub async fn unsubscribe_from_scripts(
        &self,
        scripts: Vec<ArkAddress>,
        subscription_id: String,
    ) -> Result<(), Error> {
        let scripts = scripts
            .iter()
            .map(|address| address.to_p2tr_script_pubkey().to_hex_string())
            .collect::<Vec<_>>();

        self.indexer()?
            .request(move |mut client| async move {
                client
                    .unsubscribe_for_scripts(UnsubscribeForScriptsRequest {
                        subscription_id,
                        scripts,
                    })
                    .await
            })
            .await?;

        Ok(())
    }

    /// Gets a subscription stream that returns subscription responses.
    pub async fn get_subscription(
        &self,
        subscription_id: String,
    ) -> Result<impl Stream<Item = Result<SubscriptionResponse, Error>> + Unpin, Error> {
        let response = self
            .indexer()?
            .request(move |mut client| async move {
                client
                    .get_subscription(GetSubscriptionRequest { subscription_id })
                    .await
            })
            .await?;

        let mut stream = response.into_inner();

        let stream = stream! {
            loop {
                match stream.try_next().await {
                    Ok(Some(response)) => {
                        match SubscriptionResponse::try_from(response) {
                            Ok(subscription_response) => {
                                yield Ok(subscription_response);
                            }
                            Err(e) => {
                                yield Err(e);
                            }
                        }
                    }
                    Ok(None) => {
                        break;
                    }
                    Err(e) => {
                        yield Err(Error::event_stream(e));
                    }
                }
            }
        };

        Ok(stream.boxed())
    }

    pub async fn estimate_fees(
        &self,
        intent: ark_core::intent::Intent,
    ) -> Result<SignedAmount, Error> {
        let intent = intent.try_into()?;
        let response = self
            .ark()?
            .request(move |mut client| async move {
                client
                    .estimate_intent_fee(EstimateIntentFeeRequest {
                        intent: Some(intent),
                    })
                    .await
            })
            .await?;
        let response = response.into_inner();

        Ok(SignedAmount::from_sat(response.fee))
    }

    pub async fn get_asset(&self, asset_id: AssetId) -> Result<AssetInfo, Error> {
        let response = self
            .indexer()?
            .request(move |mut client| async move {
                client
                    .get_asset(generated::ark::v1::GetAssetRequest {
                        asset_id: asset_id.to_string(),
                    })
                    .await
            })
            .await?;

        let inner = response.into_inner();

        let supply = inner.supply.parse::<u64>().map_err(Error::conversion)?;

        let asset_id = inner.asset_id.parse().map_err(Error::conversion)?;
        let control_asset_id = if inner.control_asset.is_empty() {
            None
        } else {
            Some(inner.control_asset.parse().map_err(Error::conversion)?)
        };

        Ok(AssetInfo {
            asset_id,
            control_asset_id,
            supply,
            metadata: inner.metadata,
        })
    }

    fn ark(&self) -> Result<guarded::Ark, Error> {
        self.ark.clone().ok_or(Error::not_connected())
    }

    fn indexer(&self) -> Result<guarded::Indexer, Error> {
        self.indexer.clone().ok_or(Error::not_connected())
    }
}

impl TryFrom<ark_core::intent::Intent> for Intent {
    type Error = Error;

    fn try_from(value: ark_core::intent::Intent) -> Result<Self, Self::Error> {
        Ok(Self {
            proof: value.serialize_proof(),
            message: value.serialize_message().map_err(Error::conversion)?,
        })
    }
}

impl TryFrom<generated::ark::v1::BatchStartedEvent> for BatchStartedEvent {
    type Error = Error;

    fn try_from(value: generated::ark::v1::BatchStartedEvent) -> Result<Self, Self::Error> {
        let batch_expiry = parse_sequence_number(value.batch_expiry).map_err(Error::conversion)?;

        Ok(BatchStartedEvent {
            id: value.id,
            intent_id_hashes: value.intent_id_hashes,
            batch_expiry,
        })
    }
}

impl TryFrom<generated::ark::v1::StreamStartedEvent> for StreamStartedEvent {
    type Error = Error;

    fn try_from(value: generated::ark::v1::StreamStartedEvent) -> Result<Self, Self::Error> {
        Ok(StreamStartedEvent { id: value.id })
    }
}

impl TryFrom<generated::ark::v1::BatchFinalizationEvent> for BatchFinalizationEvent {
    type Error = Error;

    fn try_from(value: generated::ark::v1::BatchFinalizationEvent) -> Result<Self, Self::Error> {
        let base64 = &base64::engine::GeneralPurpose::new(
            &base64::alphabet::STANDARD,
            base64::engine::GeneralPurposeConfig::new(),
        );

        let commitment_tx = base64
            .decode(&value.commitment_tx)
            .map_err(Error::conversion)?;
        let commitment_tx = Psbt::deserialize(&commitment_tx).map_err(Error::conversion)?;

        Ok(BatchFinalizationEvent {
            id: value.id,
            commitment_tx,
        })
    }
}

impl TryFrom<generated::ark::v1::BatchFinalizedEvent> for BatchFinalizedEvent {
    type Error = Error;

    fn try_from(value: generated::ark::v1::BatchFinalizedEvent) -> Result<Self, Self::Error> {
        let commitment_txid = value.commitment_txid.parse().map_err(Error::conversion)?;

        Ok(BatchFinalizedEvent {
            id: value.id,
            commitment_txid,
        })
    }
}

impl From<generated::ark::v1::BatchFailedEvent> for BatchFailed {
    fn from(value: generated::ark::v1::BatchFailedEvent) -> Self {
        BatchFailed {
            id: value.id,
            reason: value.reason,
        }
    }
}

impl TryFrom<generated::ark::v1::TreeSigningStartedEvent> for TreeSigningStartedEvent {
    type Error = Error;

    fn try_from(value: generated::ark::v1::TreeSigningStartedEvent) -> Result<Self, Self::Error> {
        let unsigned_commitment_tx = base64::engine::GeneralPurpose::new(
            &base64::alphabet::STANDARD,
            base64::engine::GeneralPurposeConfig::new(),
        )
        .decode(&value.unsigned_commitment_tx)
        .map_err(Error::conversion)?;

        let unsigned_commitment_tx =
            Psbt::deserialize(&unsigned_commitment_tx).map_err(Error::conversion)?;

        Ok(TreeSigningStartedEvent {
            id: value.id,
            cosigners_pubkeys: value
                .cosigners_pubkeys
                .into_iter()
                .map(|pk| pk.parse().map_err(Error::conversion))
                .collect::<Result<Vec<_>, Error>>()?,
            unsigned_commitment_tx,
        })
    }
}

impl TryFrom<generated::ark::v1::TreeNoncesAggregatedEvent> for TreeNoncesAggregatedEvent {
    type Error = Error;

    fn try_from(value: generated::ark::v1::TreeNoncesAggregatedEvent) -> Result<Self, Self::Error> {
        let tree_nonces = NoncePks::decode(value.tree_nonces).map_err(Error::conversion)?;

        Ok(TreeNoncesAggregatedEvent {
            id: value.id,
            tree_nonces,
        })
    }
}

impl TryFrom<generated::ark::v1::TreeTxEvent> for TreeTxEvent {
    type Error = Error;

    fn try_from(value: generated::ark::v1::TreeTxEvent) -> Result<Self, Self::Error> {
        let batch_tree_event_type = match value.batch_index {
            0 => BatchTreeEventType::Vtxo,
            1 => BatchTreeEventType::Connector,
            n => return Err(Error::conversion(format!("unsupported batch index: {n}"))),
        };

        let txid = if value.txid.is_empty() {
            None
        } else {
            Some(value.txid.parse().map_err(Error::conversion)?)
        };

        let base64 = &base64::engine::GeneralPurpose::new(
            &base64::alphabet::STANDARD,
            base64::engine::GeneralPurposeConfig::new(),
        );

        let bytes = base64.decode(&value.tx).map_err(Error::conversion)?;
        let tx = Psbt::deserialize(&bytes).map_err(Error::conversion)?;

        let children = value
            .children
            .iter()
            .map(|(index, txid)| Ok((*index, txid.parse().map_err(Error::conversion)?)))
            .collect::<Result<HashMap<_, _>, Error>>()?;

        Ok(Self {
            id: value.id,
            topic: value.topic,
            batch_tree_event_type,
            tx_graph_chunk: TxGraphChunk { txid, tx, children },
        })
    }
}

impl TryFrom<generated::ark::v1::TreeSignatureEvent> for TreeSignatureEvent {
    type Error = Error;

    fn try_from(value: generated::ark::v1::TreeSignatureEvent) -> Result<Self, Self::Error> {
        let batch_tree_event_type = match value.batch_index {
            0 => BatchTreeEventType::Vtxo,
            1 => BatchTreeEventType::Connector,
            n => return Err(Error::conversion(format!("unsupported batch index: {n}"))),
        };

        let txid = value.txid.parse().map_err(Error::conversion)?;

        let signature = Vec::from_hex(&value.signature).map_err(Error::conversion)?;
        let signature = Signature::from_slice(&signature).map_err(Error::conversion)?;

        Ok(Self {
            id: value.id,
            topic: value.topic,
            batch_tree_event_type,
            txid,
            signature,
        })
    }
}

impl TryFrom<generated::ark::v1::TreeNoncesEvent> for TreeNoncesEvent {
    type Error = Error;

    fn try_from(value: generated::ark::v1::TreeNoncesEvent) -> Result<Self, Self::Error> {
        let txid = value.txid.parse().map_err(Error::conversion)?;

        let nonces = TreeTxNoncePks::decode(value.nonces).map_err(Error::conversion)?;

        Ok(Self {
            id: value.id,
            topic: value.topic,
            txid,
            nonces,
        })
    }
}

impl TryFrom<generated::ark::v1::get_event_stream_response::Event> for StreamEvent {
    type Error = Error;

    fn try_from(
        value: generated::ark::v1::get_event_stream_response::Event,
    ) -> Result<Self, Self::Error> {
        Ok(match value {
            generated::ark::v1::get_event_stream_response::Event::StreamStarted(e) => {
                StreamEvent::StreamStarted(e.try_into()?)
            }
            generated::ark::v1::get_event_stream_response::Event::BatchStarted(e) => {
                StreamEvent::BatchStarted(e.try_into()?)
            }
            generated::ark::v1::get_event_stream_response::Event::BatchFinalization(e) => {
                StreamEvent::BatchFinalization(e.try_into()?)
            }
            generated::ark::v1::get_event_stream_response::Event::BatchFinalized(e) => {
                StreamEvent::BatchFinalized(e.try_into()?)
            }
            generated::ark::v1::get_event_stream_response::Event::BatchFailed(e) => {
                StreamEvent::BatchFailed(e.into())
            }
            generated::ark::v1::get_event_stream_response::Event::TreeSigningStarted(e) => {
                StreamEvent::TreeSigningStarted(e.try_into()?)
            }
            generated::ark::v1::get_event_stream_response::Event::TreeNoncesAggregated(e) => {
                StreamEvent::TreeNoncesAggregated(e.try_into()?)
            }
            generated::ark::v1::get_event_stream_response::Event::TreeTx(e) => {
                StreamEvent::TreeTx(e.try_into()?)
            }
            generated::ark::v1::get_event_stream_response::Event::TreeSignature(e) => {
                StreamEvent::TreeSignature(e.try_into()?)
            }
            generated::ark::v1::get_event_stream_response::Event::TreeNonces(e) => {
                StreamEvent::TreeNonces(e.try_into()?)
            }
            generated::ark::v1::get_event_stream_response::Event::Heartbeat(_) => {
                StreamEvent::Heartbeat
            }
        })
    }
}

impl TryFrom<generated::ark::v1::get_transactions_stream_response::Data> for StreamTransactionData {
    type Error = Error;

    fn try_from(
        value: generated::ark::v1::get_transactions_stream_response::Data,
    ) -> Result<Self, Self::Error> {
        match value {
            generated::ark::v1::get_transactions_stream_response::Data::CommitmentTx(
                commitment_tx,
            ) => Ok(StreamTransactionData::Commitment(
                CommitmentTransaction::try_from(commitment_tx)?,
            )),
            generated::ark::v1::get_transactions_stream_response::Data::ArkTx(redeem) => Ok(
                StreamTransactionData::Ark(ArkTransaction::try_from(redeem)?),
            ),
            generated::ark::v1::get_transactions_stream_response::Data::Heartbeat(_) => {
                Ok(StreamTransactionData::Heartbeat)
            }
            generated::ark::v1::get_transactions_stream_response::Data::SweepTx(tx) => {
                Ok(StreamTransactionData::Ark(ArkTransaction::try_from(tx)?))
            }
        }
    }
}

impl TryFrom<generated::ark::v1::TxNotification> for CommitmentTransaction {
    type Error = Error;

    fn try_from(value: generated::ark::v1::TxNotification) -> Result<Self, Self::Error> {
        let spent_vtxos = value
            .spent_vtxos
            .iter()
            .map(VirtualTxOutPoint::try_from)
            .collect::<Result<Vec<_>, _>>()?;

        let spendable_vtxos = value
            .spendable_vtxos
            .iter()
            .map(VirtualTxOutPoint::try_from)
            .collect::<Result<Vec<_>, _>>()?;

        Ok(CommitmentTransaction {
            txid: Txid::from_str(value.txid.as_str()).map_err(Error::conversion)?,
            spent_vtxos,
            unspent_vtxos: spendable_vtxos,
        })
    }
}

impl TryFrom<generated::ark::v1::TxNotification> for ArkTransaction {
    type Error = Error;

    fn try_from(value: generated::ark::v1::TxNotification) -> Result<Self, Self::Error> {
        let spent_vtxos = value
            .spent_vtxos
            .iter()
            .map(VirtualTxOutPoint::try_from)
            .collect::<Result<Vec<_>, _>>()?;

        let spendable_vtxos = value
            .spendable_vtxos
            .iter()
            .map(VirtualTxOutPoint::try_from)
            .collect::<Result<Vec<_>, _>>()?;

        let tx = if value.tx.is_empty() {
            None
        } else {
            let base64 = base64::engine::GeneralPurpose::new(
                &base64::alphabet::STANDARD,
                base64::engine::GeneralPurposeConfig::new(),
            );
            let bytes = base64.decode(&value.tx).map_err(Error::conversion)?;
            Some(Psbt::deserialize(&bytes).map_err(Error::conversion)?)
        };

        let checkpoint_txs = value
            .checkpoint_txs
            .into_iter()
            .map(|(k, v)| {
                let out_point = OutPoint::from_str(k.as_str()).map_err(Error::conversion)?;
                let txid = v.txid.parse().map_err(Error::conversion)?;
                Ok((out_point, txid))
            })
            .collect::<Result<HashMap<_, _>, Error>>()?;

        let swept_vtxos = value
            .swept_vtxos
            .into_iter()
            .map(OutPoint::try_from)
            .collect::<Result<Vec<_>, _>>()?;

        Ok(ArkTransaction {
            txid: Txid::from_str(value.txid.as_str()).map_err(Error::conversion)?,
            tx,
            spent_vtxos,
            unspent_vtxos: spendable_vtxos,
            checkpoint_txs,
            swept_vtxos,
        })
    }
}

impl TryFrom<Outpoint> for OutPoint {
    type Error = Error;

    fn try_from(value: Outpoint) -> Result<Self, Self::Error> {
        let point = OutPoint {
            txid: Txid::from_str(value.txid.as_str()).map_err(Error::conversion)?,
            vout: value.vout,
        };
        Ok(point)
    }
}

pub struct VtxoChainResponse {
    pub chains: VtxoChains,
    pub page: Option<IndexerPage>,
}

pub struct ListVtxosResponse {
    pub vtxos: Vec<VirtualTxOutPoint>,
    pub page: Option<IndexerPage>,
}

impl TryFrom<generated::ark::v1::GetVtxoChainResponse> for VtxoChainResponse {
    type Error = Error;

    fn try_from(value: generated::ark::v1::GetVtxoChainResponse) -> Result<Self, Self::Error> {
        let chains = value
            .chain
            .iter()
            .map(VtxoChain::try_from)
            .collect::<Result<Vec<_>, Error>>()?;

        Ok(VtxoChainResponse {
            chains: VtxoChains { inner: chains },
            page: value
                .page
                .map(IndexerPage::try_from)
                .transpose()
                .map_err(Error::conversion)?,
        })
    }
}

impl TryFrom<generated::ark::v1::GetVirtualTxsResponse> for VirtualTxsResponse {
    type Error = Error;

    fn try_from(value: generated::ark::v1::GetVirtualTxsResponse) -> Result<Self, Self::Error> {
        let base64 = &base64::engine::GeneralPurpose::new(
            &base64::alphabet::STANDARD,
            base64::engine::GeneralPurposeConfig::new(),
        );

        let txs = value
            .txs
            .into_iter()
            .map(|tx| {
                let bytes = base64.decode(&tx).map_err(Error::conversion)?;
                let psbt = Psbt::deserialize(&bytes).map_err(Error::conversion)?;

                Ok(psbt)
            })
            .collect::<Result<Vec<_>, _>>()?;

        Ok(VirtualTxsResponse {
            txs,
            page: value
                .page
                .map(IndexerPage::try_from)
                .transpose()
                .map_err(Error::conversion)?,
        })
    }
}

impl TryFrom<&generated::ark::v1::IndexerChain> for VtxoChain {
    type Error = Error;

    fn try_from(value: &generated::ark::v1::IndexerChain) -> Result<Self, Self::Error> {
        let spends = value
            .spends
            .iter()
            .map(|txid| {
                // Handle the case where txid might be 66 bytes long by trimming the last 2 bytes.
                let txid_str = if txid.len() == 66 { &txid[..64] } else { txid };
                txid_str.parse().map_err(Error::conversion)
            })
            .collect::<Result<Vec<_>, Error>>()?;

        let tx_type = match value.r#type() {
            IndexerChainedTxType::Unspecified => ChainedTxType::Unspecified,
            IndexerChainedTxType::Commitment => ChainedTxType::Commitment,
            IndexerChainedTxType::Ark => ChainedTxType::Ark,
            IndexerChainedTxType::Tree => ChainedTxType::Tree,
            IndexerChainedTxType::Checkpoint => ChainedTxType::Checkpoint,
        };

        Ok(VtxoChain {
            txid: value.txid.parse().map_err(Error::conversion)?,
            tx_type,
            spends,
            expires_at: value.expires_at,
        })
    }
}

impl From<generated::ark::v1::IndexerPageResponse> for IndexerPage {
    fn from(value: generated::ark::v1::IndexerPageResponse) -> Self {
        IndexerPage {
            current: value.current,
            next: value.next,
            total: value.total,
        }
    }
}

impl TryFrom<&generated::ark::v1::IndexerTxHistoryRecord> for history::Transaction {
    type Error = Error;

    fn try_from(value: &generated::ark::v1::IndexerTxHistoryRecord) -> Result<Self, Self::Error> {
        let sign = match value.r#type() {
            generated::ark::v1::IndexerTxType::Received => 1,
            // Default to sent if unspecified.
            generated::ark::v1::IndexerTxType::Sent
            | generated::ark::v1::IndexerTxType::Unspecified => -1,
        };

        let amount = SignedAmount::from_sat(value.amount as i64 * sign);

        let tx = match &value.key {
            Some(Key::CommitmentTxid(txid)) => history::Transaction::Commitment {
                txid: txid.parse().map_err(Error::conversion)?,
                amount,
                created_at: value.created_at,
            },
            Some(Key::VirtualTxid(txid)) => history::Transaction::Ark {
                txid: txid.parse().map_err(Error::conversion)?,
                amount,
                is_settled: value.is_settled,
                created_at: value.created_at,
            },
            None => return Err(Error::conversion("invalid transaction without key")),
        };

        Ok(tx)
    }
}

impl TryFrom<generated::ark::v1::GetSubscriptionResponse> for SubscriptionResponse {
    type Error = Error;

    fn try_from(value: generated::ark::v1::GetSubscriptionResponse) -> Result<Self, Self::Error> {
        let value = match value.data {
            Some(get_subscription_response::Data::Heartbeat(_)) => return Ok(Self::Heartbeat),
            Some(get_subscription_response::Data::Event(event)) => event,
            None => return Err(Error::conversion("empty subscription response")),
        };

        let txid = value.txid.parse().map_err(Error::conversion)?;

        let new_vtxos = value
            .new_vtxos
            .iter()
            .map(VirtualTxOutPoint::try_from)
            .collect::<Result<Vec<_>, _>>()?;

        let spent_vtxos = value
            .spent_vtxos
            .iter()
            .map(VirtualTxOutPoint::try_from)
            .collect::<Result<Vec<_>, _>>()?;

        let tx = if value.tx.is_empty() {
            None
        } else {
            match Vec::from_hex(&value.tx)
                .ok()
                .and_then(|bytes| bitcoin::consensus::deserialize::<Transaction>(&bytes).ok())
            {
                Some(raw_tx) => Some(raw_tx),
                None => {
                    let base64 = base64::engine::GeneralPurpose::new(
                        &base64::alphabet::STANDARD,
                        base64::engine::GeneralPurposeConfig::new(),
                    );
                    let bytes = base64.decode(&value.tx).map_err(|e| {
                        let tx_prefix = value.tx.chars().take(24).collect::<String>();
                        Error::conversion(format!(
                            "invalid subscription tx payload (not hex tx or base64 psbt) (len={}, tx_prefix='{}'): {e}",
                            value.tx.len(),
                            tx_prefix
                        ))
                    })?;
                    let psbt = Psbt::deserialize(&bytes).map_err(|e| {
                        let tx_prefix = value.tx.chars().take(24).collect::<String>();
                        Error::conversion(format!(
                            "invalid subscription tx payload (base64 but not psbt) (len={}, tx_prefix='{}'): {e}",
                            value.tx.len(),
                            tx_prefix
                        ))
                    })?;
                    Some(psbt.unsigned_tx)
                }
            }
        };

        let checkpoint_txs = value
            .checkpoint_txs
            .into_iter()
            .map(|(k, v)| {
                let out_point = OutPoint::from_str(k.as_str()).map_err(Error::conversion)?;
                let txid = v.txid.parse().map_err(Error::conversion)?;
                Ok((out_point, txid))
            })
            .collect::<Result<HashMap<_, _>, Error>>()?;

        let scripts = value
            .scripts
            .iter()
            .map(|h| ScriptBuf::from_hex(h).map_err(Error::conversion))
            .collect::<Result<Vec<_>, _>>()?;

        Ok(Self::Event(Box::new(SubscriptionEvent {
            txid,
            scripts,
            new_vtxos,
            spent_vtxos,
            tx,
            checkpoint_txs,
        })))
    }
}

impl From<GetVtxosRequest> for generated::ark::v1::GetVtxosRequest {
    fn from(value: GetVtxosRequest) -> Self {
        let (spendable_only, spent_only, recoverable_only, pending_only) = match value.filter() {
            Some(GetVtxosRequestFilter::Spendable) => (true, false, false, false),
            Some(GetVtxosRequestFilter::Spent) => (false, true, false, false),
            Some(GetVtxosRequestFilter::Recoverable) => (false, false, true, false),
            Some(GetVtxosRequestFilter::PendingOnly) => (false, false, false, true),
            None => (false, false, false, false),
        };

        let page = value
            .page()
            .map(|p| generated::ark::v1::IndexerPageRequest {
                size: p.size,
                index: p.index,
            });

        match value.reference() {
            GetVtxosRequestReference::Scripts(script_bufs) => Self {
                scripts: script_bufs.iter().map(|s| s.to_hex_string()).collect(),
                outpoints: Vec::new(),
                spendable_only,
                spent_only,
                recoverable_only,
                page,
                pending_only,
                after: value.after().unwrap_or(0) as i64,
                before: value.before().unwrap_or(0) as i64,
            },
            GetVtxosRequestReference::OutPoints(outpoints) => Self {
                scripts: Vec::new(),
                outpoints: outpoints.iter().map(|o| o.to_string()).collect(),
                spendable_only,
                spent_only,
                recoverable_only,
                page,
                pending_only,
                after: value.after().unwrap_or(0) as i64,
                before: value.before().unwrap_or(0) as i64,
            },
        }
    }
}

mod guarded {
    //! Guarded wrappers around the generated tonic clients.
    //!
    //! `Client` methods use these wrappers instead of storing or cloning generated clients
    //! directly. `request` is the only normal RPC escape hatch: it clones the generated
    //! client, runs the closure's future, and routes the result through `SharedState` so a
    //! digest mismatch refreshes `/info` and updates the shared digest/header state.
    //!
    //! Each `request` closure must perform exactly one non-`GetInfo` gRPC call and return
    //! that call's `tonic::Status` unchanged. Do not batch multiple RPCs, call `GetInfo`,
    //! swallow errors, or synthesize fallback successes inside the closure. Guarding is
    //! scoped to a single failed operation, and digest-mismatch refresh intentionally does
    //! not retry that operation.

    use super::InterceptedChannel;
    use super::SharedState;
    use crate::generated::ark::v1::ark_service_client::ArkServiceClient;
    use crate::generated::ark::v1::indexer_service_client::IndexerServiceClient;
    use crate::Error;
    use ark_core::server::Info;
    use futures::Future;

    #[derive(Clone)]
    pub(super) struct Ark {
        raw: ArkServiceClient<InterceptedChannel>,
        shared: SharedState,
    }

    impl Ark {
        pub(super) fn new(raw: ArkServiceClient<InterceptedChannel>, shared: SharedState) -> Self {
            Self { raw, shared }
        }

        /// Runs one Ark service RPC through the digest guard.
        ///
        /// The closure must perform exactly one non-`GetInfo` gRPC call and return that
        /// call's `tonic::Status` unchanged. Do not batch multiple RPCs, call `GetInfo`,
        /// swallow errors, or synthesize fallback successes inside the closure. A digest
        /// mismatch refreshes `/info` and returns `ServerInfoChanged`; the failed RPC is
        /// not retried automatically.
        pub(super) async fn request<T, F, Fut>(&self, f: F) -> Result<T, Error>
        where
            F: FnOnce(ArkServiceClient<InterceptedChannel>) -> Fut,
            Fut: Future<Output = Result<T, tonic::Status>>,
        {
            let client = self.raw.clone();
            let info_client = self.raw.clone();

            self.shared
                .guarded(info_client, async move {
                    f(client).await.map_err(Error::request)
                })
                .await
        }

        pub(super) async fn get_info(&self) -> Result<Info, Error> {
            self.shared.get_info_unguarded(self.raw.clone()).await
        }
    }

    #[derive(Clone)]
    pub(super) struct Indexer {
        raw: IndexerServiceClient<InterceptedChannel>,
        info_client: ArkServiceClient<InterceptedChannel>,
        shared: SharedState,
    }

    impl Indexer {
        pub(super) fn new(
            raw: IndexerServiceClient<InterceptedChannel>,
            info_client: ArkServiceClient<InterceptedChannel>,
            shared: SharedState,
        ) -> Self {
            Self {
                raw,
                info_client,
                shared,
            }
        }

        /// Runs one Indexer service RPC through the digest guard.
        ///
        /// The closure must perform exactly one non-`GetInfo` gRPC call and return that
        /// call's `tonic::Status` unchanged. Do not batch multiple RPCs, call `GetInfo`,
        /// swallow errors, or synthesize fallback successes inside the closure. A digest
        /// mismatch refreshes `/info` and returns `ServerInfoChanged`; the failed RPC is
        /// not retried automatically.
        pub(super) async fn request<T, F, Fut>(&self, f: F) -> Result<T, Error>
        where
            F: FnOnce(IndexerServiceClient<InterceptedChannel>) -> Fut,
            Fut: Future<Output = Result<T, tonic::Status>>,
        {
            let client = self.raw.clone();
            let info_client = self.info_client.clone();

            self.shared
                .guarded(info_client, async move {
                    f(client).await.map_err(Error::request)
                })
                .await
        }
    }
}