fedimint-client-module 0.12.0-beta.2

use std::fmt;
use std::future::Future;
use std::ops::RangeInclusive;
use std::sync::Arc;

use bitcoin::key::Keypair;
use bitcoin::secp256k1;
use fedimint_core::Amount;
use fedimint_core::core::{
    DynInput, DynOutput, IInput, IOutput, IntoDynInstance, ModuleInstanceId,
};
use fedimint_core::encoding::{Decodable, Encodable};
use fedimint_core::module::Amounts;
use fedimint_core::task::{MaybeSend, MaybeSync};
use fedimint_core::transaction::{Transaction, TransactionSignature};
use fedimint_logging::LOG_CLIENT;
use itertools::multiunzip;
use rand::{CryptoRng, Rng, RngCore};
use secp256k1::Secp256k1;
use tracing::warn;

use crate::module::{IdxRange, OutPointRange, StateGenerator};
use crate::sm::{self, DynState};
use crate::{
    InstancelessDynClientInput, InstancelessDynClientInputBundle, InstancelessDynClientInputSM,
    InstancelessDynClientOutput, InstancelessDynClientOutputBundle, InstancelessDynClientOutputSM,
    states_add_instance, states_to_instanceless_dyn,
};

#[derive(Clone, Debug)]
pub struct ClientInput<I = DynInput> {
    pub input: I,
    pub keys: Vec<Keypair>,
    pub amounts: Amounts,
}

#[derive(Clone)]
pub struct ClientInputSM<S = DynState> {
    pub state_machines: StateGenerator<S>,
}

impl<S> fmt::Debug for ClientInputSM<S> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("ClientInputSM")
    }
}

/// A fake [`sm::Context`] for [`NeverClientStateMachine`]
#[derive(Debug, Clone, Eq, PartialEq, Hash, Decodable, Encodable)]
pub enum NeverClientContext {}

impl sm::Context for NeverClientContext {
    const KIND: Option<fedimint_core::core::ModuleKind> = None;
}

/// A fake [`sm::State`] that can actually never happen.
///
/// Useful as a default for type inference in cases where there are no
/// state machines involved in [`ClientInputBundle`].
#[derive(Debug, Clone, Eq, PartialEq, Hash, Decodable, Encodable)]
pub enum NeverClientStateMachine {}

impl IntoDynInstance for NeverClientStateMachine {
    type DynType = DynState;

    fn into_dyn(self, _instance_id: ModuleInstanceId) -> Self::DynType {
        unreachable!()
    }
}
impl sm::State for NeverClientStateMachine {
    type ModuleContext = NeverClientContext;

    fn transitions(
        &self,
        _context: &Self::ModuleContext,
        _global_context: &crate::DynGlobalClientContext,
    ) -> Vec<sm::StateTransition<Self>> {
        unreachable!()
    }

    fn operation_id(&self) -> fedimint_core::core::OperationId {
        unreachable!()
    }
}

/// A group of inputs and state machines responsible for driving their state
///
/// These must be kept together as a whole when including in a transaction.
#[derive(Clone, Debug)]
pub struct ClientInputBundle<I = DynInput, S = DynState> {
    pub(crate) inputs: Vec<ClientInput<I>>,
    pub(crate) sm_gens: Vec<ClientInputSM<S>>,
}

impl<I> ClientInputBundle<I, NeverClientStateMachine> {
    /// A version of [`Self::new`] for times where input does not require any
    /// state machines
    ///
    /// This avoids type inference issues of `S`, and saves some typing.
    pub fn new_no_sm(inputs: Vec<ClientInput<I>>) -> Self {
        if inputs.is_empty() {
            // TODO: Make it return Result or assert?
            warn!(target: LOG_CLIENT, "Empty input bundle will be illegal in the future");
        }
        Self {
            inputs,
            sm_gens: vec![],
        }
    }
}

impl<I, S> ClientInputBundle<I, S>
where
    I: IInput + MaybeSend + MaybeSync + 'static,
    S: sm::IState + MaybeSend + MaybeSync + 'static,
{
    pub fn new(inputs: Vec<ClientInput<I>>, sm_gens: Vec<ClientInputSM<S>>) -> Self {
        Self { inputs, sm_gens }
    }

    pub fn sms(&self) -> &[ClientInputSM<S>] {
        &self.sm_gens
    }

    pub fn into_instanceless(self) -> InstancelessDynClientInputBundle {
        InstancelessDynClientInputBundle {
            inputs: self
                .inputs
                .into_iter()
                .map(|input| InstancelessDynClientInput {
                    input: Box::new(input.input),
                    keys: input.keys,
                    amounts: input.amounts,
                })
                .collect(),
            sm_gens: self
                .sm_gens
                .into_iter()
                .map(|input_sm| InstancelessDynClientInputSM {
                    state_machines: states_to_instanceless_dyn(input_sm.state_machines),
                })
                .collect(),
        }
    }
}

impl<I, S> ClientInputBundle<I, S> {
    pub fn inputs(&self) -> &[ClientInput<I>] {
        &self.inputs
    }

    pub fn is_empty(&self) -> bool {
        // Notably, sm_gen will not be called when inputs are empty anyway
        self.inputs.is_empty()
    }
}

impl<I> IntoDynInstance for ClientInput<I>
where
    I: IntoDynInstance<DynType = DynInput> + 'static,
{
    type DynType = ClientInput;

    fn into_dyn(self, module_instance_id: ModuleInstanceId) -> ClientInput {
        ClientInput {
            input: self.input.into_dyn(module_instance_id),
            keys: self.keys,
            amounts: self.amounts,
        }
    }
}

impl<S> IntoDynInstance for ClientInputSM<S>
where
    S: IntoDynInstance<DynType = DynState> + 'static,
{
    type DynType = ClientInputSM;

    fn into_dyn(self, module_instance_id: ModuleInstanceId) -> ClientInputSM {
        ClientInputSM {
            state_machines: state_gen_to_dyn(self.state_machines, module_instance_id),
        }
    }
}

impl<I, S> IntoDynInstance for ClientInputBundle<I, S>
where
    I: IntoDynInstance<DynType = DynInput> + 'static,
    S: IntoDynInstance<DynType = DynState> + 'static,
{
    type DynType = ClientInputBundle;

    fn into_dyn(self, module_instance_id: ModuleInstanceId) -> ClientInputBundle {
        ClientInputBundle {
            inputs: self
                .inputs
                .into_iter()
                .map(|input| input.into_dyn(module_instance_id))
                .collect::<Vec<ClientInput>>(),

            sm_gens: self
                .sm_gens
                .into_iter()
                .map(|input_sm| input_sm.into_dyn(module_instance_id))
                .collect::<Vec<ClientInputSM>>(),
        }
    }
}

impl IntoDynInstance for InstancelessDynClientInputBundle {
    type DynType = ClientInputBundle;

    fn into_dyn(self, module_instance_id: ModuleInstanceId) -> ClientInputBundle {
        ClientInputBundle {
            inputs: self
                .inputs
                .into_iter()
                .map(|input| ClientInput {
                    input: DynInput::from_parts(module_instance_id, input.input),
                    keys: input.keys,
                    amounts: input.amounts,
                })
                .collect::<Vec<ClientInput>>(),

            sm_gens: self
                .sm_gens
                .into_iter()
                .map(|input_sm| ClientInputSM {
                    state_machines: states_add_instance(
                        module_instance_id,
                        input_sm.state_machines,
                    ),
                })
                .collect::<Vec<ClientInputSM>>(),
        }
    }
}

#[derive(Clone, Debug)]
pub struct ClientOutputBundle<O = DynOutput, S = DynState> {
    pub(crate) outputs: Vec<ClientOutput<O>>,
    pub(crate) sm_gens: Vec<ClientOutputSM<S>>,
}

#[derive(Clone, Debug)]
pub struct ClientOutput<O = DynOutput> {
    pub output: O,
    pub amounts: Amounts,
}

#[derive(Clone)]
pub struct ClientOutputSM<S = DynState> {
    pub state_machines: StateGenerator<S>,
}

impl<S> fmt::Debug for ClientOutputSM<S> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("ClientOutputSM")
    }
}
impl<O> ClientOutputBundle<O, NeverClientStateMachine> {
    /// A version of [`Self::new`] for times where output does not require any
    /// state machines
    ///
    /// This avoids type inference issues of `S`, and saves some typing.
    pub fn new_no_sm(outputs: Vec<ClientOutput<O>>) -> Self {
        if outputs.is_empty() {
            // TODO: Make it return Result or assert?
            warn!(target: LOG_CLIENT, "Empty output bundle will be illegal in the future");
        }
        Self {
            outputs,
            sm_gens: vec![],
        }
    }
}
impl<O, S> ClientOutputBundle<O, S> {
    pub fn outputs(&self) -> &[ClientOutput<O>] {
        &self.outputs
    }
}

impl<O, S> ClientOutputBundle<O, S>
where
    O: IOutput + MaybeSend + MaybeSync + 'static,
    S: sm::IState + MaybeSend + MaybeSync + 'static,
{
    pub fn new(outputs: Vec<ClientOutput<O>>, sm_gens: Vec<ClientOutputSM<S>>) -> Self {
        Self { outputs, sm_gens }
    }

    pub fn sms(&self) -> &[ClientOutputSM<S>] {
        &self.sm_gens
    }

    pub fn with(mut self, other: Self) -> Self {
        self.outputs.extend(other.outputs);
        self.sm_gens.extend(other.sm_gens);
        self
    }

    pub fn into_instanceless(self) -> InstancelessDynClientOutputBundle {
        InstancelessDynClientOutputBundle {
            outputs: self
                .outputs
                .into_iter()
                .map(|output| InstancelessDynClientOutput {
                    output: Box::new(output.output),
                    amounts: output.amounts,
                })
                .collect(),
            sm_gens: self
                .sm_gens
                .into_iter()
                .map(|output_sm| InstancelessDynClientOutputSM {
                    state_machines: states_to_instanceless_dyn(output_sm.state_machines),
                })
                .collect(),
        }
    }
}

impl<O, S> ClientOutputBundle<O, S> {
    pub fn is_empty(&self) -> bool {
        // Notably, sm_gen will not be called when outputs are empty anyway
        self.outputs.is_empty()
    }
}

impl<I, S> IntoDynInstance for ClientOutputBundle<I, S>
where
    I: IntoDynInstance<DynType = DynOutput> + 'static,
    S: IntoDynInstance<DynType = DynState> + 'static,
{
    type DynType = ClientOutputBundle;

    fn into_dyn(self, module_instance_id: ModuleInstanceId) -> ClientOutputBundle {
        ClientOutputBundle {
            outputs: self
                .outputs
                .into_iter()
                .map(|output| output.into_dyn(module_instance_id))
                .collect::<Vec<ClientOutput>>(),

            sm_gens: self
                .sm_gens
                .into_iter()
                .map(|output_sm| output_sm.into_dyn(module_instance_id))
                .collect::<Vec<ClientOutputSM>>(),
        }
    }
}

impl IntoDynInstance for InstancelessDynClientOutputBundle {
    type DynType = ClientOutputBundle;

    fn into_dyn(self, module_instance_id: ModuleInstanceId) -> ClientOutputBundle {
        ClientOutputBundle {
            outputs: self
                .outputs
                .into_iter()
                .map(|output| ClientOutput {
                    output: DynOutput::from_parts(module_instance_id, output.output),
                    amounts: output.amounts,
                })
                .collect::<Vec<ClientOutput>>(),

            sm_gens: self
                .sm_gens
                .into_iter()
                .map(|output_sm| ClientOutputSM {
                    state_machines: states_add_instance(
                        module_instance_id,
                        output_sm.state_machines,
                    ),
                })
                .collect::<Vec<ClientOutputSM>>(),
        }
    }
}

impl<I> IntoDynInstance for ClientOutput<I>
where
    I: IntoDynInstance<DynType = DynOutput> + 'static,
{
    type DynType = ClientOutput;

    fn into_dyn(self, module_instance_id: ModuleInstanceId) -> ClientOutput {
        ClientOutput {
            output: self.output.into_dyn(module_instance_id),
            amounts: self.amounts,
        }
    }
}

impl<S> IntoDynInstance for ClientOutputSM<S>
where
    S: IntoDynInstance<DynType = DynState> + 'static,
{
    type DynType = ClientOutputSM;

    fn into_dyn(self, module_instance_id: ModuleInstanceId) -> ClientOutputSM {
        ClientOutputSM {
            state_machines: state_gen_to_dyn(self.state_machines, module_instance_id),
        }
    }
}

/// The explicit (operation-supplied) side of a transaction, summarized for a
/// fee quote via `Client::fee_quote`.
///
/// A module describes the inputs and outputs its operation would contribute by
/// their gross value and federation fees (per unit), rather than building the
/// actual transaction. The primary module for each affected unit then balances
/// the resulting imbalance with change, and the full breakdown is returned as a
/// [`FeeQuote`]. This avoids fabricating real inputs/outputs (which may require
/// cryptographic material not available when quoting), since fees depend only
/// on amounts and module.
///
/// All four fields are multi-unit [`Amounts`], so an operation can describe
/// explicit items spanning several units (e.g. Bitcoin plus a custom currency);
/// each unit is then balanced independently by its own primary module, the same
/// way `Client::finalize_transaction` does.
///
/// For a typical receive there are no explicit outputs, so `output_amount` and
/// `output_fee` are [`Amounts::ZERO`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FeeQuoteRequest {
    /// Gross value of the operation's explicit inputs, per unit.
    pub input_amount: Amounts,
    /// Gross value of the operation's explicit outputs, per unit.
    pub output_amount: Amounts,
    /// Federation fees charged on the operation's explicit inputs, per unit.
    pub input_fee: Amounts,
    /// Federation fees charged on the operation's explicit outputs, per unit.
    pub output_fee: Amounts,
}

/// Breakdown of the fee finalizing a transaction would incur, as computed by
/// `Client::fee_quote` (a dry-run of the same balancing the real submission
/// performs).
///
/// This is module-agnostic: the explicit inputs/outputs are described by the
/// [`FeeQuoteRequest`] from whichever module is quoting (mint, lightning,
/// wallet, …) and the change is generated by the primary module. The quote is
/// point-in-time: it depends on the client's current inventory and can move as
/// funds change.
///
/// Each field is a multi-unit [`Amounts`], since the quoted operation may span
/// several units. The total fee is the sum of the breakdown fields, available
/// via [`FeeQuote::total`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FeeQuote {
    /// Federation fees charged on the spent (input) items — both the
    /// transaction's explicit inputs and any inputs the primary module pulls in
    /// to balance it.
    pub input: Amounts,
    /// Federation fees charged on the created (output) items — both the
    /// transaction's explicit outputs and the change minted to balance it.
    pub output: Amounts,
    /// Sub-denomination remainder that cannot form an output and is lost.
    pub dust: Amounts,
}

impl FeeQuote {
    /// A zero fee, for operations that incur no cost at all — e.g. an ecash
    /// send served entirely from existing exact-change notes, which submits
    /// no transaction.
    pub const ZERO: Self = Self {
        input: Amounts::ZERO,
        output: Amounts::ZERO,
        dust: Amounts::ZERO,
    };

    /// Total fee per unit: everything the gross input value does not become a
    /// net wallet gain. Equal to `input + output + dust`.
    pub fn total(&self) -> Amounts {
        self.input
            .clone()
            .checked_add(&self.output)
            .and_then(|sum| sum.checked_add(&self.dust))
            .expect("aggregate fee components cannot overflow an Amounts")
    }
}

/// Finds the largest send amount payable in full out of `balance` when the
/// payment carries both an *external* per-amount fee (e.g. a Lightning gateway
/// routing fee) and the on-federation transaction fee.
///
/// Sending an amount `x` requires funding a value `gross_up(x)` on the
/// federation — the external fee grosses the amount up, e.g. the outgoing
/// Lightning contract is `x + gateway_fee(x)` — and funding that value costs an
/// additional federation fee reported by `fee_quote` (the module's output fee
/// plus the primary module's funding input fees, change output fees and dust).
/// This returns the largest `x` in `min_amount..=max_amount` satisfying
///
/// ```text
/// gross_up(x) + fee_quote(gross_up(x)).total().get_bitcoin() <= balance
/// ```
///
/// or `None` if even `min_amount` is unaffordable.
///
/// The cost is not a closed form of `x`: the federation fee is charged per
/// note, so note selection, denomination rounding, change and dust move it in
/// steps as `x` crosses thresholds. Rather than invert an analytic formula this
/// evaluates the real quote via a monotonic binary search — each step a single,
/// non-committing `fee_quote` dry-run over the current inventory — and only
/// ever advances the lower bound to a verified-affordable amount, so the result
/// never overestimates. A `fee_quote` error (e.g. the balance cannot fund a
/// value that large) is treated as unaffordable.
///
/// To keep those dry-runs cheap the search is *seeded near the top*: a first
/// pass binary-searches `gross_up` alone (pure arithmetic, no quotes) for the
/// largest fundable amount, peeling off the gateway fee for free, so the real
/// fee quotes only probe the small window the federation fee leaves — a handful
/// of dry-runs rather than one per bit of the balance.
///
/// The LNv2 and LNv1 send-all flows share this solver; they differ only in
/// `gross_up` (the gateway fee model) and which `fee_quote` they pass.
pub async fn max_affordable_send_amount<GrossUp, Quote, Fut>(
    balance: Amount,
    min_amount: Amount,
    max_amount: Amount,
    gross_up: GrossUp,
    fee_quote: Quote,
) -> Option<Amount>
where
    GrossUp: Fn(Amount) -> Amount,
    Quote: Fn(Amount) -> Fut,
    Fut: Future<Output = anyhow::Result<FeeQuote>>,
{
    // Nothing above the balance can ever be funded, so cap the upper bound.
    let hi_bound = max_amount.msats.min(balance.msats);
    let lo_bound = min_amount.msats;

    if lo_bound > hi_bound {
        return None;
    }

    // The maximum is never near the bottom of `[lo_bound, hi_bound]`: it sits
    // just below the largest amount the balance can *fund*, short by only the
    // federation fee. `gross_up` is a pure, cheap function (no fee quote), so
    // binary-search it for free to find that fundable ceiling and seed the real
    // (fee-quoting) search there. This peels off the gateway fee — usually the
    // larger of the two — for free, leaving the expensive quotes to probe only
    // the small window the federation fee opens up, instead of the whole balance.
    if gross_up(Amount::from_msats(lo_bound)).msats > balance.msats {
        // The balance can't even fund the smallest amount's gross-up.
        return None;
    }
    let fundable_max = {
        let mut lo = lo_bound;
        let mut hi = hi_bound;
        while lo < hi {
            let mid = lo + (hi - lo).div_ceil(2);
            if gross_up(Amount::from_msats(mid)).msats <= balance.msats {
                lo = mid;
            } else {
                hi = mid - 1;
            }
        }
        lo
    };

    let mut lo = lo_bound;
    let mut hi = fundable_max;
    let mut lo_affordable = false;

    // Probe the fee once at `fundable_max`. The send overhead (the gross-up plus
    // the federation fee, less the amount) is monotone non-decreasing, so the
    // overhead here is an upper bound on the overhead at the true maximum, and
    // `balance - overhead` is therefore a proven-affordable, tight lower bound.
    // If the quote errors — real note selection can't fund a value this large —
    // the seed is skipped and the search falls back to the full bracket below.
    let funded = gross_up(Amount::from_msats(fundable_max));
    if let Ok(quote) = fee_quote(funded).await {
        let total = funded
            .msats
            .saturating_add(quote.total().get_bitcoin().msats);
        if total <= balance.msats {
            // Even the largest fundable amount fits once the fee is included.
            return Some(Amount::from_msats(fundable_max));
        }
        let overhead = total.saturating_sub(fundable_max);
        let seed = balance
            .msats
            .saturating_sub(overhead)
            .clamp(lo_bound, fundable_max);
        if send_amount_affordable(Amount::from_msats(seed), balance, &gross_up, &fee_quote).await {
            lo = seed;
            lo_affordable = true;
        }
    }

    // Without a usable seed the search must still start from a verified
    // affordable lower bound, or give up if even `lo_bound` is unaffordable.
    if !lo_affordable
        && !send_amount_affordable(Amount::from_msats(lo_bound), balance, &gross_up, &fee_quote)
            .await
    {
        return None;
    }

    // Exact maximum within the (now tight) `[lo, hi]` bracket.
    while lo < hi {
        // Bias the midpoint up so the search converges toward `hi`.
        let mid = lo + (hi - lo).div_ceil(2);

        if send_amount_affordable(Amount::from_msats(mid), balance, &gross_up, &fee_quote).await {
            lo = mid;
        } else {
            hi = mid - 1;
        }
    }

    Some(Amount::from_msats(lo))
}

/// Whether sending `amount` is payable in full out of `balance`: the funded
/// value `gross_up(amount)` plus its federation `fee_quote` must fit within the
/// balance. A quote error (the balance cannot fund a value this large) counts
/// as unaffordable, making this safe as the monotone predicate for
/// [`max_affordable_send_amount`].
async fn send_amount_affordable<GrossUp, Quote, Fut>(
    amount: Amount,
    balance: Amount,
    gross_up: &GrossUp,
    fee_quote: &Quote,
) -> bool
where
    GrossUp: Fn(Amount) -> Amount,
    Quote: Fn(Amount) -> Fut,
    Fut: Future<Output = anyhow::Result<FeeQuote>>,
{
    let funded_amount = gross_up(amount);

    if funded_amount > balance {
        return false;
    }

    match fee_quote(funded_amount).await {
        Ok(quote) => funded_amount + quote.total().get_bitcoin() <= balance,
        Err(_) => false,
    }
}

#[derive(Default, Clone, Debug)]
pub struct TransactionBuilder {
    inputs: Vec<ClientInputBundle>,
    outputs: Vec<ClientOutputBundle>,
}

impl TransactionBuilder {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn with_inputs(mut self, inputs: ClientInputBundle) -> Self {
        self.inputs.push(inputs);
        self
    }

    pub fn with_outputs(mut self, outputs: ClientOutputBundle) -> Self {
        self.outputs.push(outputs);
        self
    }

    pub fn build<C, R: RngCore + CryptoRng>(
        self,
        secp_ctx: &Secp256k1<C>,
        mut rng: R,
    ) -> (Transaction, Vec<DynState>)
    where
        C: secp256k1::Signing + secp256k1::Verification,
    {
        // `input_idx_to_bundle_idx[input_idx]` stores the index of a bundle the input
        // at `input_idx` comes from, so we can call state machines of the
        // corresponding bundle for every input bundle. It is always
        // monotonically increasing, e.g. `[0, 0, 1, 2, 2, 2, 4]`
        let (input_idx_to_bundle_idx, inputs, input_keys): (Vec<_>, Vec<_>, Vec<_>) = multiunzip(
            self.inputs
                .iter()
                .enumerate()
                .flat_map(|(bundle_idx, bundle)| {
                    bundle
                        .inputs
                        .iter()
                        .map(move |input| (bundle_idx, input.input.clone(), input.keys.clone()))
                }),
        );
        // `output_idx_to_bundle` works exactly like `input_idx_to_bundle_idx` above,
        // but for outputs.
        let (output_idx_to_bundle_idx, outputs): (Vec<_>, Vec<_>) = multiunzip(
            self.outputs
                .iter()
                .enumerate()
                .flat_map(|(bundle_idx, bundle)| {
                    bundle
                        .outputs
                        .iter()
                        .map(move |output| (bundle_idx, output.output.clone()))
                }),
        );
        let nonce: [u8; 8] = rng.r#gen();

        let txid = Transaction::tx_hash_from_parts(&inputs, &outputs, nonce);
        let msg = secp256k1::Message::from_digest_slice(&txid[..]).expect("txid has right length");

        let signatures = input_keys
            .iter()
            .flatten()
            .map(|keypair| secp_ctx.sign_schnorr(&msg, keypair))
            .collect();

        let transaction = Transaction {
            inputs,
            outputs,
            nonce,
            signatures: TransactionSignature::NaiveMultisig(signatures),
        };

        let input_states = self
            .inputs
            .into_iter()
            .enumerate()
            .filter(|(_, bundle)| !bundle.is_empty())
            .flat_map(|(bundle_idx, bundle)| {
                let input_idxs = find_range_of_matching_items(&input_idx_to_bundle_idx, bundle_idx)
                    .expect("Non empty bundles must always have a match");
                bundle.sm_gens.into_iter().flat_map(move |sm| {
                    (sm.state_machines)(OutPointRange::new(
                        txid,
                        IdxRange::from_inclusive(input_idxs.clone()).expect("can't overflow"),
                    ))
                })
            });

        let output_states = self
            .outputs
            .into_iter()
            .enumerate()
            .filter(|(_, bundle)| !bundle.is_empty())
            .flat_map(|(bundle_idx, bundle)| {
                let output_idxs =
                    find_range_of_matching_items(&output_idx_to_bundle_idx, bundle_idx)
                        .expect("Non empty bundles must always have a match");
                bundle.sm_gens.into_iter().flat_map(move |sm| {
                    (sm.state_machines)(OutPointRange::new(
                        txid,
                        IdxRange::from_inclusive(output_idxs.clone())
                            .expect("can't possibly overflow"),
                    ))
                })
            });
        (transaction, input_states.chain(output_states).collect())
    }

    pub fn inputs(&self) -> impl Iterator<Item = &ClientInput> {
        self.inputs.iter().flat_map(|i| i.inputs.iter())
    }

    pub fn outputs(&self) -> impl Iterator<Item = &ClientOutput> {
        self.outputs.iter().flat_map(|i| i.outputs.iter())
    }
}

/// Find the range of indexes in an monotonically increasing `arr`, that is
/// equal to `item`
fn find_range_of_matching_items(arr: &[usize], item: usize) -> Option<RangeInclusive<u64>> {
    // `arr` must be monotonically increasing
    debug_assert!(arr.windows(2).all(|w| w[0] <= w[1]));

    arr.iter()
        .enumerate()
        .filter_map(|(arr_idx, arr_item)| (*arr_item == item).then_some(arr_idx as u64))
        .fold(None, |cur: Option<(u64, u64)>, idx| {
            Some(cur.map_or((idx, idx), |cur| (cur.0.min(idx), cur.1.max(idx))))
        })
        .map(|(start, end)| start..=end)
}

#[test]
fn find_range_of_matching_items_sanity() {
    assert_eq!(find_range_of_matching_items(&[0, 0], 0), Some(0..=1));
    assert_eq!(find_range_of_matching_items(&[0, 0, 1], 0), Some(0..=1));
    assert_eq!(find_range_of_matching_items(&[0, 0, 1], 1), Some(2..=2));
    assert_eq!(find_range_of_matching_items(&[0, 0, 1], 2), None);
    assert_eq!(find_range_of_matching_items(&[], 0), None);
}

fn state_gen_to_dyn<S>(
    state_gen: StateGenerator<S>,
    module_instance: ModuleInstanceId,
) -> StateGenerator<DynState>
where
    S: IntoDynInstance<DynType = DynState> + 'static,
{
    Arc::new(move |out_point_range| {
        let states = state_gen(out_point_range);
        states
            .into_iter()
            .map(|state| state.into_dyn(module_instance))
            .collect()
    })
}

#[cfg(test)]
mod tests;