fedimint_walletv2_client/

lib.rs

#![deny(clippy::pedantic)]
#![allow(clippy::missing_errors_doc)]
#![allow(clippy::missing_panics_doc)]
#![allow(clippy::must_use_candidate)]
#![allow(clippy::module_name_repetitions)]

pub use fedimint_walletv2_common as common;
pub use fedimint_walletv2_common::MempoolVisibility;

mod api;
#[cfg(feature = "cli")]
mod cli;
pub mod db;
pub mod events;
mod receive_sm;
mod send_sm;

use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;
use std::time::Duration;

use api::WalletFederationApi;
use bitcoin::address::NetworkUnchecked;
use bitcoin::{Address, ScriptBuf};
use db::{NextOutputIndexKey, ValidAddressIndexKey, ValidAddressIndexPrefix};
use events::{ReceivePaymentEvent, SendPaymentEvent};
use fedimint_api_client::api::{DynModuleApi, FederationError, FederationResult};
use fedimint_client::DynGlobalClientContext;
use fedimint_client::transaction::{
    ClientInput, ClientInputBundle, ClientInputSM, ClientOutput, ClientOutputBundle,
    ClientOutputSM, FeeQuote, FeeQuoteRequest, TransactionBuilder, max_affordable_send_amount,
};
use fedimint_client_module::db::ClientModuleMigrationFn;
use fedimint_client_module::error::{
    ClientModuleError, OperationLookupError, TransactionSubmitError,
};
use fedimint_client_module::module::init::{ClientModuleInit, ClientModuleInitArgs};
use fedimint_client_module::module::recovery::NoModuleBackup;
use fedimint_client_module::module::{ClientContext, ClientModule, OutPointRange};
use fedimint_client_module::sm::{Context, DynState, ModuleNotifier, State, StateTransition};
use fedimint_client_module::sm_enum_variant_translation;
use fedimint_core::core::{IntoDynInstance, ModuleInstanceId, ModuleKind, OperationId};
use fedimint_core::db::{
    AutocommitError, Database, DatabaseError, DatabaseTransaction, DatabaseVersion,
    IDatabaseTransactionOpsCoreTyped,
};
use fedimint_core::encoding::{Decodable, Encodable};
use fedimint_core::module::{
    AmountUnit, Amounts, ApiVersion, CommonModuleInit, ModuleCommon, ModuleInit, MultiApiVersion,
};
use fedimint_core::task::{MaybeSend, MaybeSync, TaskGroup, TaskHandle, sleep};
use fedimint_core::{Amount, OutPoint, TransactionId, apply, async_trait_maybe_send};
use fedimint_derive_secret::{ChildId, DerivableSecret};
use fedimint_eventlog::{Event, EventLogId};
use fedimint_logging::LOG_CLIENT_MODULE_WALLETV2;
use fedimint_walletv2_common::config::WalletClientConfig;
use fedimint_walletv2_common::{
    BLOCK_COUNT_VOTE_DELAY, KIND, OutputInfo, PendingOutput, StandardScript, TxInfo,
    WalletCommonInit, WalletInput, WalletInputV0, WalletModuleTypes, WalletOutput, WalletOutputV0,
    descriptor, is_potential_receive,
};
use futures::StreamExt;
use receive_sm::{ReceiveSMCommon, ReceiveSMState, ReceiveStateMachine};
use secp256k1::Keypair;
use send_sm::{SendSMCommon, SendSMState, SendStateMachine};
use serde::{Deserialize, Serialize};
use strum::IntoEnumIterator as _;
use thiserror::Error;
use tracing::{debug, warn};

/// Number of output info entries to scan per batch.
const SLICE_SIZE: u64 = 1000;

/// Number of event log entries to read per batch.
const EVENT_LOG_PAGE_SIZE: u64 = 1000;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum WalletOperationMeta {
    Send(SendMeta),
    Receive(ReceiveMeta),
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SendMeta {
    pub change_outpoint_range: OutPointRange,
    pub address: Address<NetworkUnchecked>,
    pub value: bitcoin::Amount,
    pub fee: bitcoin::Amount,
    #[serde(default)]
    pub custom_meta: serde_json::Value,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReceiveMeta {
    pub change_outpoint_range: OutPointRange,
    pub value: bitcoin::Amount,
    pub fee: bitcoin::Amount,
    pub address: Option<Address<NetworkUnchecked>>,
    pub outpoint: Option<bitcoin::OutPoint>,
}

/// The final state of an operation sending bitcoin onchain.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum FinalSendOperationState {
    /// The transaction was successful.
    Success(bitcoin::Txid),
    /// The funding transaction was aborted.
    Aborted,
    /// A programming error has occurred or the federation is malicious.
    Failure,
}

/// The final state of an operation receiving bitcoin onchain.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum FinalReceiveOperationState {
    /// The federation accepted the claiming transaction.
    Success,
    /// The federation rejected the claiming transaction.
    Aborted,
}

/// Progress of a single incoming peg-in.
///
/// Each variant describes one deposit, identified by the outpoint it is looked
/// up or paired with. An address that has been paid more than once has one of
/// these per payment rather than a combined view, so nothing is aggregated
/// away.
///
/// There is no "nothing has arrived" variant: a deposit no guardian has seen
/// has no outpoint to describe it, so absence is expressed by
/// [`WalletClientModule::address_receive_progress`] reporting no deposits.
///
/// Every variant is derived from advisory, non-consensus data reported by
/// individual guardians, so none of them is authoritative and progress may move
/// backwards: a reorg or an eviction can take a deposit back to being unseen
/// entirely. No variant means money has been received - a deposit is only
/// actually claimed once [`WalletClientModule::await_receive`] returns, and an
/// output is deliberately still reported here for a short while after that.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum ReceiveProgress {
    /// The deposit is in a guardian's mempool but is not mined.
    ///
    /// The weakest signal there is: an unmined transaction can be replaced or
    /// evicted, and a deposit in this state can disappear again. Treat it as
    /// "we have seen your payment", never as money received.
    ///
    /// Only guardians whose bitcoin backend can enumerate a mempool report
    /// these, so on a federation without one a deposit stays unseen until it is
    /// mined.
    Mempool { value: bitcoin::Amount },
    /// The deposit has been mined but is not yet deep enough for the federation
    /// to act on it.
    Confirming {
        value: bitcoin::Amount,
        /// Height of the block that mined the deposit.
        ///
        /// Reported as the guardians saw it, so a reorg can move a deposit to a
        /// different height or back to being unseen entirely.
        height: u64,
        /// Confirmations counting the block that mined the transaction as the
        /// first, matching what a block explorer displays.
        confirmations: u64,
        /// Confirmations needed before the federation records the output.
        required: u64,
    },
}

/// What the federation currently reports about the deposits to one address.
///
/// Returned by [`WalletClientModule::address_receive_progress`]. The two fields
/// are only meaningful together: they come from a single query, so the
/// visibility flag describes exactly the view the deposits were read from.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AddressReceiveProgress {
    /// One entry per deposit the guardians can see, ordered by outpoint.
    pub deposits: Vec<(bitcoin::OutPoint, ReceiveProgress)>,
    /// The best mempool view any guardian had for this query.
    ///
    /// With [`MempoolVisibility::Complete`], empty `deposits` mean no
    /// matching deposit was observed in the reported snapshot. That is not
    /// proof that no payment was sent: a guardian only inspects its own
    /// mempool, which a freshly broadcast transaction may not have reached
    /// yet, and a deposit mined long enough ago has left the reported block
    /// window. With [`MempoolVisibility::Partial`] the unmined deposits
    /// reported are real, but one may also be missing because a guardian
    /// failed to fetch its transaction or has not reached it yet while still
    /// working through a large mempool, which is normal progress rather than
    /// a backend fault; with
    /// [`MempoolVisibility::Unavailable`] the federation is blind to a
    /// payment until it is mined. One guardian that inspected its whole
    /// mempool is enough to make this `Complete` for everyone.
    ///
    /// A wallet showing a receive screen wants this to choose between "we have
    /// not seen your payment yet" and "waiting for the first confirmation",
    /// which are very different things to tell a user who has just paid.
    pub mempool_visibility: MempoolVisibility,
}

#[derive(Debug, Clone)]
pub struct WalletClientModule {
    root_secret: DerivableSecret,
    cfg: WalletClientConfig,
    notifier: ModuleNotifier<WalletClientStateMachines>,
    client_ctx: ClientContext<Self>,
    db: Database,
    module_api: DynModuleApi,
}

#[derive(Debug, Clone)]
pub struct WalletClientContext {
    pub client_ctx: ClientContext<WalletClientModule>,
}

impl Context for WalletClientContext {
    const KIND: Option<ModuleKind> = Some(KIND);
}

#[apply(async_trait_maybe_send!)]
impl ClientModule for WalletClientModule {
    type Init = WalletClientInit;
    type Common = WalletModuleTypes;
    type Backup = NoModuleBackup;
    type ModuleStateMachineContext = WalletClientContext;
    type States = WalletClientStateMachines;

    fn context(&self) -> Self::ModuleStateMachineContext {
        WalletClientContext {
            client_ctx: self.client_ctx.clone(),
        }
    }

    fn input_fee(
        &self,
        amount: &Amounts,
        _input: &<Self::Common as ModuleCommon>::Input,
    ) -> Option<Amounts> {
        amount
            .get(&AmountUnit::BITCOIN)
            .map(|a| Amounts::new_bitcoin(self.cfg.fee_consensus.fee(*a)))
    }

    fn output_fee(
        &self,
        amount: &Amounts,
        _output: &<Self::Common as ModuleCommon>::Output,
    ) -> Option<Amounts> {
        amount
            .get(&AmountUnit::BITCOIN)
            .map(|a| Amounts::new_bitcoin(self.cfg.fee_consensus.fee(*a)))
    }

    #[cfg(feature = "cli")]
    async fn handle_cli_command(
        &self,
        args: &[std::ffi::OsString],
    ) -> Result<serde_json::Value, ClientModuleError> {
        cli::handle_cli_command(self, args)
            .await
            .map_err(ClientModuleError::other)
    }
}

#[derive(Debug, Clone, Default)]
pub struct WalletClientInit;

impl ModuleInit for WalletClientInit {
    type Common = WalletCommonInit;

    async fn dump_database(
        &self,
        _dbtx: &mut DatabaseTransaction<'_>,
        _prefix_names: Vec<String>,
    ) -> Box<dyn Iterator<Item = (String, Box<dyn erased_serde::Serialize + Send>)> + '_> {
        Box::new(BTreeMap::new().into_iter())
    }
}

#[apply(async_trait_maybe_send!)]
impl ClientModuleInit for WalletClientInit {
    type Module = WalletClientModule;

    fn supported_api_versions(&self) -> MultiApiVersion {
        MultiApiVersion::try_from_iter([ApiVersion { major: 0, minor: 0 }])
            .expect("no version conflicts")
    }

    async fn init(
        &self,
        args: &ClientModuleInitArgs<Self>,
    ) -> Result<Self::Module, ClientModuleError> {
        let module = WalletClientModule {
            root_secret: args.module_root_secret().clone(),
            cfg: args.cfg().clone(),
            notifier: args.notifier().clone(),
            client_ctx: args.context(),
            db: args.db().clone(),
            module_api: args.module_api().clone(),
        };

        module.spawn_output_scanner(args.task_group(), args.client_span());

        Ok(module)
    }

    fn get_database_migrations(&self) -> BTreeMap<DatabaseVersion, ClientModuleMigrationFn> {
        BTreeMap::new()
    }

    fn used_db_prefixes(&self) -> Option<BTreeSet<u8>> {
        Some(db::DbKeyPrefix::iter().map(|p| p as u8).collect())
    }
}

impl WalletClientModule {
    /// Returns the Bitcoin network for this federation.
    pub fn get_network(&self) -> bitcoin::Network {
        self.cfg.network
    }

    /// Fetch the total value of bitcoin controlled by the federation.
    pub async fn total_value(&self) -> FederationResult<bitcoin::Amount> {
        self.module_api
            .federation_wallet()
            .await
            .map(|tx_out| tx_out.map_or(bitcoin::Amount::ZERO, |tx_out| tx_out.value))
    }

    /// Fetch the consensus block count of the federation.
    pub async fn block_count(&self) -> FederationResult<u64> {
        self.module_api.consensus_block_count().await
    }

    /// Fetch the current consensus feerate.
    pub async fn feerate(&self) -> FederationResult<Option<u64>> {
        self.module_api.consensus_feerate().await
    }

    /// Fetch information on the chain of pending bitcoin transactions.
    pub async fn pending_tx_chain(&self) -> FederationResult<Vec<TxInfo>> {
        self.module_api.pending_tx_chain().await
    }

    /// Display log of bitcoin transactions.
    pub async fn tx_chain(&self) -> FederationResult<Vec<TxInfo>> {
        self.module_api.tx_chain().await
    }

    /// Fetch the current fee required to send an onchain payment.
    pub async fn send_fee(&self) -> Result<bitcoin::Amount, SendError> {
        self.module_api
            .send_fee()
            .await
            .map_err(|e| SendError::Federation(Box::new(e)))?
            .ok_or(SendError::NoConsensusFeerateAvailable)
    }

    /// Computes the federation fee an onchain send of an output worth `amount`
    /// (the payment amount plus the on-chain miner fee) would incur, without
    /// submitting anything.
    ///
    /// A send submits a single wallet output worth `amount`; the primary module
    /// balances it by spending ecash to fund the output and minting any change.
    /// This quotes the fee of that transaction — the wallet output fee, the
    /// mint input fees on the funding notes, any mint change output fees,
    /// and sub-denomination dust — via the shared, module-agnostic fee
    /// quote.
    ///
    /// The on-chain Bitcoin miner fee is deliberately excluded: it is part of
    /// the output `amount` (see [`Self::send_fee`]), not the on-federation
    /// transaction fee.
    pub async fn send_fee_quote(
        &self,
        amount: bitcoin::Amount,
    ) -> Result<FeeQuote, TransactionSubmitError> {
        let amount = Amount::from_sats(amount.to_sat());
        self.client_ctx
            .fee_quote(
                OperationId::new_random(),
                FeeQuoteRequest {
                    input_amount: Amounts::ZERO,
                    output_amount: Amounts::new_bitcoin(amount),
                    input_fee: Amounts::ZERO,
                    output_fee: Amounts::new_bitcoin(self.cfg.fee_consensus.fee(amount)),
                },
            )
            .await
    }

    /// Finds the largest value that can be sent on chain in full out of
    /// `balance` — the amount a "send everything" sweep should use.
    ///
    /// Sending `value` costs `value + fee` (the on-chain miner fee is carried
    /// inside the wallet output, see [`Self::send`]) *plus* the federation fee
    /// of funding that output — the wallet output fee, the mint input fees on
    /// the funding notes, any mint change output fees and sub-denomination
    /// dust — as quoted by [`Self::send_fee_quote`]. This returns the largest
    /// `value` satisfying
    ///
    /// ```text
    /// value + fee + send_fee_quote(value + fee).total() <= balance
    /// ```
    ///
    /// `balance` is the client's current Bitcoin balance (e.g. from
    /// `Client::get_balance_for_btc`). `fee` is the on-chain fee from
    /// [`Self::send_fee`]; pass the *same* value on to [`Self::send`], since
    /// the required feerate rises with each pending federation transaction and
    /// a value computed against a stale fee would be rejected.
    ///
    /// The maximum is found by binary search over the real fee quote (see
    /// [`max_affordable_send_amount`]) rather than by subtracting a single
    /// quote: the federation fee is charged per note, so note selection,
    /// denomination rounding, change and dust move it in steps as the value
    /// crosses thresholds, and a quote taken at the full balance would fail
    /// outright — funding it is the very thing that is unaffordable.
    ///
    /// The quote is point-in-time and moves with the balance; [`Self::send`]
    /// remains the source of truth. Note that it cannot account for the
    /// federation's own on-chain constraints — a send whose change UTXO would
    /// fall below the dust limit is still rejected by the guardians.
    ///
    /// Returns [`SendError::InsufficientFunds`] if the balance cannot cover
    /// the dust limit plus fees, or [`SendError::Failed`] if the fee probe
    /// itself failed.
    pub async fn max_sendable_amount(
        &self,
        balance: Amount,
        fee: bitcoin::Amount,
    ) -> Result<bitcoin::Amount, SendError> {
        let fee_msats = Amount::from_sats(fee.to_sat());

        let max = max_affordable_send_amount(
            balance,
            Amount::from_sats(self.cfg.dust_limit.to_sat()),
            balance,
            // The solver searches millisatoshis, but a send funds a whole
            // number of satoshis. Rounding the probe up to the next satoshi
            // keeps the predicate conservative and makes the value handed to
            // the quote below an exact satoshi multiple.
            |value: Amount| Amount::from_sats(value.msats.div_ceil(1000)) + fee_msats,
            |funded: Amount| self.send_fee_quote(bitcoin::Amount::from_sat(funded.msats / 1000)),
        )
        .await
        .map_err(SendError::Failed)?
        .ok_or(SendError::InsufficientFunds)?;

        // `gross_up` rounded up to whole satoshis, so the largest affordable
        // amount already sits on a satoshi boundary; no value is lost here.
        Ok(bitcoin::Amount::from_sat(max.msats.div_ceil(1000)))
    }

    /// Fetch the current fee required to claim an onchain deposit (peg-in).
    pub async fn receive_fee(&self) -> Result<bitcoin::Amount, ReceiveError> {
        self.module_api
            .receive_fee()
            .await
            .map_err(|e| ReceiveError::Federation(Box::new(e)))?
            .ok_or(ReceiveError::NoConsensusFeerateAvailable)
    }

    /// Send an onchain payment with the given fee.
    pub async fn send(
        &self,
        address: Address<NetworkUnchecked>,
        value: bitcoin::Amount,
        fee: Option<bitcoin::Amount>,
        custom_meta: serde_json::Value,
    ) -> Result<OperationId, SendError> {
        if !address.is_valid_for_network(self.cfg.network) {
            return Err(SendError::WrongNetwork);
        }

        if value < self.cfg.dust_limit {
            return Err(SendError::DustValue);
        }

        let fee = match fee {
            Some(value) => value,
            None => self
                .module_api
                .send_fee()
                .await
                .map_err(|e| SendError::Federation(Box::new(e)))?
                .ok_or(SendError::NoConsensusFeerateAvailable)?,
        };

        let operation_id = OperationId::new_random();

        let destination = StandardScript::from_address(&address.clone().assume_checked())
            .ok_or(SendError::UnsupportedAddress)?;

        let client_output = ClientOutput::<WalletOutput> {
            output: WalletOutput::V0(WalletOutputV0 {
                destination,
                value,
                fee,
            }),
            amounts: Amounts::new_bitcoin(Amount::from_sats((value + fee).to_sat())),
        };

        let client_output_sm = ClientOutputSM::<WalletClientStateMachines> {
            state_machines: Arc::new(move |range: OutPointRange| {
                vec![WalletClientStateMachines::Send(SendStateMachine {
                    common: SendSMCommon {
                        operation_id,
                        outpoint: OutPoint {
                            txid: range.txid(),
                            out_idx: 0,
                        },
                        value,
                        fee,
                    },
                    state: SendSMState::Funding,
                })]
            }),
        };

        let client_output_bundle = self.client_ctx.make_client_outputs(ClientOutputBundle::new(
            vec![client_output],
            vec![client_output_sm],
        ));

        let address_clone = address.clone();

        self.submit_with_event(
            operation_id,
            move |change_outpoint_range| {
                WalletOperationMeta::Send(SendMeta {
                    change_outpoint_range,
                    address: address_clone.clone(),
                    value,
                    fee,
                    custom_meta: custom_meta.clone(),
                })
            },
            TransactionBuilder::new().with_outputs(client_output_bundle),
            SendPaymentEvent {
                operation_id,
                address,
                value,
                fee,
            },
        )
        .await
        .map_err(|error| match error {
            TransactionSubmitError::InsufficientFunds(_) => SendError::InsufficientFunds,
            error => SendError::Failed(error),
        })?;

        Ok(operation_id)
    }

    /// Await the final state of the send operation.
    pub async fn await_final_send_operation_state(
        &self,
        operation_id: OperationId,
    ) -> Result<FinalSendOperationState, OperationLookupError> {
        let operation = self.client_ctx.get_operation(operation_id).await?;
        let mut stream = self.notifier.subscribe(operation_id).await;

        let mut stream = self
            .client_ctx
            .outcome_or_updates(&operation, operation_id, |_| true, move || {
                async_stream::stream! {
                    loop {
                        if let Some(WalletClientStateMachines::Send(state)) = stream.next().await {
                            match state.state {
                                SendSMState::Funding => {}
                                SendSMState::Success(txid) => {
                                    yield FinalSendOperationState::Success(txid);
                                    return;
                                }
                                SendSMState::Aborted(..) => {
                                    yield FinalSendOperationState::Aborted;
                                    return;
                                }
                                SendSMState::Failure => {
                                    yield FinalSendOperationState::Failure;
                                    return;
                                }
                            }
                        }
                    }
                }
            })
            .into_stream();

        let mut final_state = None;

        while let Some(state) = stream.next().await {
            final_state = Some(state);
        }

        Ok(final_state.expect("Stream contains one final state"))
    }

    /// Await the final state of the receive operation.
    pub async fn await_final_receive_operation_state(
        &self,
        operation_id: OperationId,
    ) -> Result<FinalReceiveOperationState, OperationLookupError> {
        let operation = self.client_ctx.get_operation(operation_id).await?;
        let mut stream = self.notifier.subscribe(operation_id).await;

        let mut stream = self
            .client_ctx
            .outcome_or_updates(&operation, operation_id, |_| true, move || {
                async_stream::stream! {
                    loop {
                        if let Some(WalletClientStateMachines::Receive(state)) = stream.next().await {
                            match state.state {
                                ReceiveSMState::Funding => {}
                                ReceiveSMState::Success => {
                                    yield FinalReceiveOperationState::Success;
                                    return;
                                }
                                ReceiveSMState::Aborted(..) => {
                                    yield FinalReceiveOperationState::Aborted;
                                    return;
                                }
                            }
                        }
                    }
                }
            })
            .into_stream();

        let mut final_state = None;

        while let Some(state) = stream.next().await {
            final_state = Some(state);
        }

        Ok(final_state.expect("Stream contains one final state"))
    }

    /// Resolves one of our receive addresses back to the index it was derived
    /// at.
    ///
    /// Only addresses handed out by [`Self::receive`] resolve; the set of valid
    /// indices is small, so this is a short scan rather than a stored reverse
    /// index.
    async fn address_index(&self, address: &Address) -> Option<u64> {
        let script = address.script_pubkey();

        let indices: Vec<u64> = self
            .db
            .begin_transaction_nc()
            .await
            .find_by_prefix(&ValidAddressIndexPrefix)
            .await
            .map(|entry| entry.0.0)
            .collect()
            .await;

        indices
            .into_iter()
            .find(|index| self.derive_address(*index).script_pubkey() == script)
    }

    /// The current progress of the deposit at `outpoint`.
    ///
    /// Reads only the guardians' pending view, which is advisory and may move
    /// backwards, and which cannot tell a claimed deposit from a reorged one.
    /// Use [`Self::await_receive`] to establish that a deposit has actually
    /// been claimed.
    ///
    /// Each call queries the federation, so poll it sparingly.
    ///
    /// Errors if no deposit at `outpoint` is known to this client, which
    /// includes one that has been reorged out or evicted since it was last
    /// seen, and one paying an address this client did not derive.
    pub async fn receive_progress(
        &self,
        outpoint: bitcoin::OutPoint,
    ) -> Result<ReceiveProgress, ReceiveProgressError> {
        let pending = self.module_api.pending_outputs().await;

        // Guardians report every output passing the receive filter, most of
        // which belong to other clients, so ownership is established by matching
        // the script against our own addresses.
        let ours = self.address_map().await;

        pending
            .outputs
            .iter()
            .find(|output| output.outpoint == outpoint && ours.contains_key(&output.script))
            .map(|output| Self::pending_progress(output, pending.block_count))
            .ok_or(ReceiveProgressError::UnknownDeposit)
    }

    /// The progress of every deposit to `address` the federation has seen,
    /// ordered by outpoint.
    ///
    /// Empty deposits mean no guardian has reported a payment to the address
    /// yet. Even when [`AddressReceiveProgress::mempool_visibility`] is
    /// [`MempoolVisibility::Complete`], that only means no matching deposit
    /// was observed in the reported snapshot, never proof that no payment was
    /// sent: a broadcast transaction may not have reached the guardians'
    /// mempools yet. Without `Complete` an unmined deposit may be invisible
    /// to the federation altogether.
    ///
    /// Errors if `address` was not handed out by [`Self::receive`].
    pub async fn address_receive_progress(
        &self,
        address: &Address,
    ) -> Result<AddressReceiveProgress, ReceiveProgressError> {
        self.address_index(address)
            .await
            .ok_or(ReceiveProgressError::UnknownAddress)?;

        Ok(self.deposits_paying(&address.script_pubkey()).await)
    }

    /// Every deposit the federation currently reports against `script`.
    async fn deposits_paying(&self, script: &ScriptBuf) -> AddressReceiveProgress {
        let pending = self.module_api.pending_outputs().await;

        let mut deposits: Vec<(bitcoin::OutPoint, ReceiveProgress)> = pending
            .outputs
            .iter()
            .filter(|output| output.script == *script)
            .map(|output| {
                (
                    output.outpoint,
                    Self::pending_progress(output, pending.block_count),
                )
            })
            .collect();

        deposits.sort_by_key(|(outpoint, _)| *outpoint);

        AddressReceiveProgress {
            deposits,
            mempool_visibility: pending.mempool_visibility,
        }
    }

    /// The scripts of every receive address derived so far, by index.
    async fn address_map(&self) -> BTreeMap<ScriptBuf, u64> {
        self.db
            .begin_transaction_nc()
            .await
            .find_by_prefix(&ValidAddressIndexPrefix)
            .await
            .map(|entry| entry.0.0)
            .map(|index| (self.derive_address(index).script_pubkey(), index))
            .collect()
            .await
    }

    fn pending_progress(output: &PendingOutput, block_count: u64) -> ReceiveProgress {
        // Block count votes trail the chain tip by the vote delay, and an output
        // is only recorded once the consensus block count is past the block that
        // mined it, so a peg-in is claimable one standard confirmation later than
        // the vote delay itself.
        let required = BLOCK_COUNT_VOTE_DELAY + 1;

        match output.height {
            Some(height) => ReceiveProgress::Confirming {
                value: output.value,
                height,
                // Confirmations count the block that mined the deposit as the
                // first, matching what a block explorer displays.
                //
                // Guardians keep reporting an output for a while after it turns
                // final, so that progress does not blank out before the claim
                // begins. Clamp so a progress display cannot run past its own
                // target during that overlap.
                confirmations: block_count.saturating_sub(height).min(required),
                required,
            },
            None => ReceiveProgress::Mempool {
                value: output.value,
            },
        }
    }

    /// Returns the highest valid receive address index that the background
    /// scanner has derived so far, or `None` if it has not derived one yet.
    async fn valid_index(&self) -> Option<u64> {
        self.db
            .begin_transaction_nc()
            .await
            .find_by_prefix_sorted_descending(&ValidAddressIndexPrefix)
            .await
            .next()
            .await
            .map(|entry| entry.0.0)
    }

    /// Returns the next unused receive address.
    ///
    /// To wait for a payment to this address race-free, read the client's
    /// current event log position (via the global `get_next_event_log_id`)
    /// *before* calling this, then pass that position to
    /// [`Self::await_receive`]; it will only consider payments received
    /// after that position.
    ///
    /// If the background scanner has already derived a valid address index this
    /// returns immediately. Otherwise it blocks, letting the scanner grind
    /// until it finds the next valid index, and returns once one is
    /// available.
    pub async fn receive(&self) -> Address {
        loop {
            if let Some(index) = self.valid_index().await {
                return self.derive_address(index);
            }

            sleep(Duration::from_secs(1)).await;
        }
    }

    /// Block until the next on-chain payment recorded at or after `position` is
    /// received and successfully claimed by the federation.
    ///
    /// Returns the peg-in's final state together with the event log position
    /// just past it, so that a subsequent call can resume from there to wait
    /// for the following receive.
    ///
    /// A peg-in attempt may be aborted (rejected by the federation), in which
    /// case the still-unspent output is reprocessed into a new receive
    /// operation; this keeps waiting until one succeeds.
    pub async fn await_receive(
        &self,
        position: EventLogId,
    ) -> Result<(FinalReceiveOperationState, EventLogId), AwaitReceiveError> {
        let mut position = position;

        loop {
            let (operation_id, next_position) = self.next_receive_operation(position).await;

            position = next_position;

            let state = self
                .await_final_receive_operation_state(operation_id)
                .await?;

            // A successful peg-in is terminal; an aborted one is retried as a
            // new receive operation, so keep waiting.
            if state == FinalReceiveOperationState::Success {
                // Reaching `Success` only means the peg-in claim transaction was
                // accepted into consensus. The ecash it mints is issued
                // asynchronously by the primary module, so wait for those
                // outputs before returning; otherwise the freshly claimed funds
                // may not yet be reflected in the client's balance.
                let operation = self.client_ctx.get_operation(operation_id).await?;

                if let WalletOperationMeta::Receive(ReceiveMeta {
                    change_outpoint_range,
                    ..
                }) = operation.meta::<WalletOperationMeta>()
                {
                    self.client_ctx
                        .await_primary_module_outputs(
                            operation_id,
                            change_outpoint_range.into_iter().collect(),
                        )
                        .await?;
                }

                return Ok((state, position));
            }
        }
    }

    /// Scan the event log from `position` for the next [`ReceivePaymentEvent`],
    /// blocking until one is found, and return its operation id together with
    /// the event log position just past it.
    async fn next_receive_operation(&self, position: EventLogId) -> (OperationId, EventLogId) {
        let mut position = position;

        loop {
            let events = self
                .client_ctx
                .get_event_log(Some(position), EVENT_LOG_PAGE_SIZE)
                .await;

            for entry in &events {
                position = entry.id().saturating_add(1);

                if entry.module_kind() == Some(&KIND)
                    && entry.kind == ReceivePaymentEvent::KIND
                    && let Some(event) = entry.to_event::<ReceivePaymentEvent>()
                {
                    return (event.operation_id, position);
                }
            }

            if events.is_empty() {
                // Caught up with the log; wait for new events to be written.
                sleep(Duration::from_secs(1)).await;
            }
        }
    }

    fn derive_address(&self, index: u64) -> Address {
        descriptor(
            &self.cfg.bitcoin_pks,
            &self.derive_tweak(index).public_key().consensus_hash(),
        )
        .address(self.cfg.network)
    }

    fn derive_tweak(&self, index: u64) -> Keypair {
        self.root_secret
            .child_key(ChildId(index))
            .to_secp_key(secp256k1::SECP256K1)
    }

    /// Find the next valid index starting from (and including) `start_index`.
    ///
    /// Only ~1/65536 indices are valid, so the search is CPU-bound and may scan
    /// many indices before finding one. The scan runs in bounded batches and
    /// yields to the executor between them, so it does not stall the runtime —
    /// important on wasm, which is single-threaded. It stops and returns `None`
    /// once the task group begins shutting down.
    async fn next_valid_index(&self, start_index: u64, handle: &TaskHandle) -> Option<u64> {
        /// Indices to scan per batch before yielding to the executor.
        const SCAN_BATCH: u64 = 256;

        let pks_hash = self.cfg.bitcoin_pks.consensus_hash();

        let mut index = start_index;

        while !handle.is_shutting_down() {
            for _ in 0..SCAN_BATCH {
                if is_potential_receive(&self.derive_address(index).script_pubkey(), &pks_hash) {
                    return Some(index);
                }

                index += 1;
            }

            // Hand control back to the executor between batches.
            sleep(Duration::ZERO).await;
        }

        None
    }

    /// Issue ecash for an unspent output with a given fee.
    ///
    /// Returns `None` if the output value cannot cover the fee, or if the
    /// remainder is too small to fund the claim transaction's fees.
    async fn receive_output(
        &self,
        output_index: u64,
        value: bitcoin::Amount,
        address_index: u64,
        fee: bitcoin::Amount,
        outpoint: Option<bitcoin::OutPoint>,
    ) -> Option<(OperationId, TransactionId)> {
        let operation_id = OperationId::new_random();

        let client_input = ClientInput::<WalletInput> {
            input: WalletInput::V0(WalletInputV0 {
                output_index,
                fee,
                tweak: self.derive_tweak(address_index).public_key(),
            }),
            keys: vec![self.derive_tweak(address_index)],
            amounts: Amounts::new_bitcoin(Amount::from_sats(value.checked_sub(fee)?.to_sat())),
        };

        let client_input_sm = ClientInputSM::<WalletClientStateMachines> {
            state_machines: Arc::new(move |range: OutPointRange| {
                vec![WalletClientStateMachines::Receive(ReceiveStateMachine {
                    common: ReceiveSMCommon {
                        operation_id,
                        txid: range.txid(),
                        value,
                        fee,
                    },
                    state: ReceiveSMState::Funding,
                })]
            }),
        };

        let client_input_bundle = self.client_ctx.make_client_inputs(ClientInputBundle::new(
            vec![client_input],
            vec![client_input_sm],
        ));

        let address = self.derive_address(address_index).as_unchecked().clone();

        let meta_address = address.clone();
        let range = self
            .submit_with_event(
                operation_id,
                move |change_outpoint_range| {
                    WalletOperationMeta::Receive(ReceiveMeta {
                        change_outpoint_range,
                        value,
                        fee,
                        address: Some(meta_address.clone()),
                        outpoint,
                    })
                },
                TransactionBuilder::new().with_inputs(client_input_bundle),
                ReceivePaymentEvent {
                    operation_id,
                    value,
                    fee,
                    address,
                    outpoint,
                },
            )
            .await
            .ok()?;

        Some((operation_id, range.txid()))
    }

    /// Submits a transaction as a new operation and logs the event that
    /// announces the operation, in one database transaction.
    ///
    /// Committing the two together is what lets a reader of the event log rely
    /// on it: an operation that exists has its event, also after the client
    /// stopped at any point in between.
    async fn submit_with_event<F, E>(
        &self,
        operation_id: OperationId,
        operation_meta_gen: F,
        tx_builder: TransactionBuilder,
        event: E,
    ) -> Result<OutPointRange, TransactionSubmitError>
    where
        F: Fn(OutPointRange) -> WalletOperationMeta + Clone + MaybeSend + MaybeSync + 'static,
        E: Event + Clone + Send + MaybeSync,
    {
        self.client_ctx
            .module_db()
            .autocommit(
                |dbtx, _| {
                    let operation_meta_gen = operation_meta_gen.clone();
                    let tx_builder = tx_builder.clone();
                    let event = event.clone();
                    Box::pin(async move {
                        let range = self
                            .client_ctx
                            .finalize_and_submit_transaction_dbtx(
                                dbtx,
                                operation_id,
                                WalletCommonInit::KIND.as_str(),
                                operation_meta_gen,
                                tx_builder,
                            )
                            .await?;

                        self.client_ctx.log_event(dbtx, event).await;

                        Ok::<_, TransactionSubmitError>(range)
                    })
                },
                Some(100),
            )
            .await
            .map_err(|error| match error {
                AutocommitError::ClosureError { error, .. } => error,
                AutocommitError::CommitFailed { last_error, .. } => {
                    TransactionSubmitError::Database(last_error)
                }
            })
    }

    fn spawn_output_scanner(&self, task_group: &TaskGroup, client_span: &tracing::Span) {
        let module = self.clone();
        let handle = task_group.make_handle();

        task_group.spawn_cancellable_with_span(client_span.clone(), "output-scanner", async move {
            // Read the stored index with a short-lived transaction, runs the search
            // with no transaction open, and inserts the result in its own short
            // transaction. This avoids holding a transaction snapshot across a
            // long-running CPU-bound search that might outlive RocksDB's retained
            // write history, causing a SnapshotTooOld error and panic on commit.
            // See: https://github.com/fedimint/fedimint/issues/9202
            let address_index_exists = {
                let mut dbtx = module.db.begin_transaction().await;
                dbtx.find_by_prefix(&ValidAddressIndexPrefix)
                    .await
                    .next()
                    .await
                    .is_some()
            };

            if !address_index_exists {
                let Some(index) = module.next_valid_index(0, &handle).await else {
                    return;
                };

                let mut dbtx = module.db.begin_transaction().await;
                dbtx.insert_new_entry(&ValidAddressIndexKey(index), &())
                    .await;
                dbtx.commit_tx().await;
            }

            loop {
                match module.check_outputs(&handle).await {
                    Ok(skip_wait) => {
                        if skip_wait {
                            continue;
                        }
                    }
                    Err(e) => {
                        warn!(target: LOG_CLIENT_MODULE_WALLETV2, "Failed to fetch outputs: {e}");
                    }
                }

                sleep(fedimint_walletv2_common::sleep_duration()).await;
            }
        });
    }

    async fn check_outputs(&self, handle: &TaskHandle) -> Result<bool, CheckOutputsError> {
        let mut dbtx = self.db.begin_transaction_nc().await;

        let next_output_index = dbtx.get_value(&NextOutputIndexKey).await.unwrap_or(0);

        let mut valid_indices: Vec<u64> = dbtx
            .find_by_prefix(&ValidAddressIndexPrefix)
            .await
            .map(|entry| entry.0.0)
            .collect()
            .await;

        let mut address_map: BTreeMap<ScriptBuf, u64> = valid_indices
            .iter()
            .map(|&i| (self.derive_address(i).script_pubkey(), i))
            .collect();

        let outputs = self
            .module_api
            .output_info_slice(next_output_index, next_output_index + SLICE_SIZE)
            .await?;

        let returned_num = outputs.len();
        let mut matched_num: usize = 0;

        for output in &outputs {
            if let Some(&address_index) = address_map.get(&output.script) {
                matched_num += 1;

                // Claim before extending the valid index list: the index search
                // below is CPU-bound and can take longer than a short-lived
                // client process (e.g. a cli invocation) lives. The claim is
                // quick and the extension can be retried on the next scan.
                if !output.spent && !self.process_unspent_output(output, address_index).await? {
                    return Ok(false);
                }

                let next_address_index = valid_indices
                    .last()
                    .copied()
                    .expect("we have at least one address index");

                // If we used the highest valid index, add the next valid one
                if address_index == next_address_index {
                    let Some(index) = self.next_valid_index(next_address_index + 1, handle).await
                    else {
                        return Ok(false);
                    };

                    let mut dbtx = self.db.begin_transaction().await;

                    dbtx.insert_entry(&ValidAddressIndexKey(index), &()).await;

                    dbtx.commit_tx_result().await?;

                    valid_indices.push(index);

                    address_map.insert(self.derive_address(index).script_pubkey(), index);
                }
            }

            let mut dbtx = self.db.begin_transaction().await;

            dbtx.insert_entry(&NextOutputIndexKey, &(output.index + 1))
                .await;

            dbtx.commit_tx_result().await?;
        }

        debug!(
            target: LOG_CLIENT_MODULE_WALLETV2,
            next_output_index,
            returned_num,
            matched_num,
            valid_indices_num = valid_indices.len(),
            "Scanning for outputs"
        );

        Ok(!outputs.is_empty())
    }

    async fn process_unspent_output(
        &self,
        output: &OutputInfo,
        address_index: u64,
    ) -> Result<bool, ProcessOutputError> {
        debug!(
            target: LOG_CLIENT_MODULE_WALLETV2,
            output_index = output.index,
            value_sat = output.value.to_sat(),
            address_index,
            outpoint = ?output.outpoint,
            "Discovered unspent walletv2 receive output"
        );

        // In order to not overpay on fees we choose to wait,
        // the congestion will clear up within a few blocks.
        let pending_tx_chain_len = self.module_api.pending_tx_chain().await?.len();
        if 3 <= pending_tx_chain_len {
            debug!(
                target: LOG_CLIENT_MODULE_WALLETV2,
                output_index = output.index,
                pending_tx_chain_len,
                "Delaying walletv2 receive claim because pending transaction chain is full"
            );
            return Ok(false);
        }

        let receive_fee = self
            .module_api
            .receive_fee()
            .await?
            .ok_or(ProcessOutputError::NoFeerate)?;

        if let Some((operation_id, txid)) = self
            .receive_output(
                output.index,
                output.value,
                address_index,
                receive_fee,
                output.outpoint,
            )
            .await
        {
            debug!(
                target: LOG_CLIENT_MODULE_WALLETV2,
                output_index = output.index,
                ?operation_id,
                %txid,
                "Waiting for walletv2 receive claim acceptance"
            );
            self.client_ctx
                .transaction_updates(operation_id)
                .await
                .await_tx_accepted(txid)
                .await
                .map_err(ProcessOutputError::ClaimRejected)?;
            debug!(
                target: LOG_CLIENT_MODULE_WALLETV2,
                output_index = output.index,
                ?operation_id,
                %txid,
                "Walletv2 receive claim accepted"
            );
        } else {
            debug!(
                target: LOG_CLIENT_MODULE_WALLETV2,
                output_index = output.index,
                value_sat = output.value.to_sat(),
                fee_sat = receive_fee.to_sat(),
                "Skipping walletv2 receive claim; value cannot cover the claim fees"
            );
        }

        Ok(true)
    }
}

/// A failure to send an on-chain payment.
#[derive(Error, Debug)]
#[non_exhaustive]
pub enum SendError {
    /// The destination address is not valid on the federation's network.
    #[error("Address is from a different network than the federation")]
    WrongNetwork,

    /// The value to send is below the federation's dust limit.
    #[error("The value is too small")]
    DustValue,

    /// The federation could not be asked for the current on-chain fee.
    #[error("The federation returned an error")]
    Federation(#[source] Box<FederationError>),

    /// The guardians have not agreed a feerate yet, so no on-chain fee can be
    /// quoted.
    #[error("No consensus feerate is available at this time")]
    NoConsensusFeerateAvailable,

    /// The client cannot fund the wallet output this send needs.
    #[error("The client does not have sufficient funds to send the payment")]
    InsufficientFunds,

    /// The destination is not an address type the federation can pay.
    #[error("Unsupported address type")]
    UnsupportedAddress,

    /// The send transaction could not be submitted for a reason that is not
    /// about funding.
    #[error("The send transaction could not be submitted")]
    Failed(#[source] TransactionSubmitError),
}

/// A failure to quote the fee for claiming an on-chain deposit.
#[derive(Error, Debug)]
#[non_exhaustive]
pub enum ReceiveError {
    /// The federation could not be asked for the current claim fee.
    #[error("The federation returned an error")]
    Federation(#[source] Box<FederationError>),

    /// The guardians have not agreed a feerate yet, so no claim fee can be
    /// quoted.
    #[error("No consensus feerate is available at this time")]
    NoConsensusFeerateAvailable,
}

/// A failure to wait for the next on-chain payment to be received.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum AwaitReceiveError {
    /// The receive operation could not be looked up.
    #[error("The receive operation could not be looked up")]
    Operation(#[from] OperationLookupError),

    /// The deposit was claimed, but the ecash it mints was never issued.
    #[error("The ecash for the claimed deposit could not be issued")]
    Issuance(#[from] TransactionSubmitError),
}

/// A failure to look up the progress of an incoming peg-in.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum ReceiveProgressError {
    /// No guardian reports a deposit at this outpoint to one of our addresses.
    /// This includes a deposit that was reorged out or evicted since it was
    /// last seen.
    #[error("No deposit of ours at this outpoint")]
    UnknownDeposit,

    /// The address was not handed out by [`WalletClientModule::receive`].
    #[error("Address was not derived by this client")]
    UnknownAddress,
}

/// A failure while scanning the federation's outputs for payments to this
/// client.
#[derive(Debug, Error)]
enum CheckOutputsError {
    /// The federation did not serve the scan.
    #[error(transparent)]
    Federation(#[from] FederationError),

    /// The scan's progress could not be committed.
    #[error(transparent)]
    Database(#[from] DatabaseError),

    /// An unspent output paid to this client could not be claimed.
    #[error(transparent)]
    ProcessOutput(#[from] ProcessOutputError),
}

/// A failure to claim an unspent output paid to this client.
#[derive(Debug, Error)]
enum ProcessOutputError {
    /// The federation could not report its pending transactions or the
    /// receive fee.
    #[error(transparent)]
    Federation(#[from] FederationError),

    /// The federation has no consensus feerate to price the claim with.
    #[error("No consensus feerate is available")]
    NoFeerate,

    /// The claim transaction was rejected.
    #[error("Claim transaction was rejected: {0}")]
    ClaimRejected(String),
}

#[derive(Debug, Clone, Eq, PartialEq, Hash, Decodable, Encodable)]
pub enum WalletClientStateMachines {
    Send(send_sm::SendStateMachine),
    Receive(receive_sm::ReceiveStateMachine),
}

impl State for WalletClientStateMachines {
    type ModuleContext = WalletClientContext;

    fn transitions(
        &self,
        context: &Self::ModuleContext,
        global_context: &DynGlobalClientContext,
    ) -> Vec<StateTransition<Self>> {
        match self {
            WalletClientStateMachines::Send(sm) => sm_enum_variant_translation!(
                sm.transitions(context, global_context),
                WalletClientStateMachines::Send
            ),
            WalletClientStateMachines::Receive(sm) => sm_enum_variant_translation!(
                sm.transitions(context, global_context),
                WalletClientStateMachines::Receive
            ),
        }
    }

    fn operation_id(&self) -> OperationId {
        match self {
            WalletClientStateMachines::Send(sm) => sm.operation_id(),
            WalletClientStateMachines::Receive(sm) => sm.operation_id(),
        }
    }
}

impl IntoDynInstance for WalletClientStateMachines {
    type DynType = DynState;

    fn into_dyn(self, instance_id: ModuleInstanceId) -> Self::DynType {
        DynState::from_typed(instance_id, self)
    }
}