fedimint_client/

client.rs

use std::collections::{BTreeMap, HashSet};
use std::fmt::{self, Formatter};
use std::future::{Future, pending};
use std::ops::Range;
use std::pin::Pin;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use anyhow::{Context as _, anyhow, bail, format_err};
use async_stream::try_stream;
use bitcoin::key::Secp256k1;
use bitcoin::key::rand::thread_rng;
use bitcoin::secp256k1::{self, PublicKey};
use fedimint_api_client::api::global_api::with_request_hook::ApiRequestHook;
use fedimint_api_client::api::{
    ApiVersionSet, DynGlobalApi, FederationApiExt as _, FederationResult, IGlobalFederationApi,
};
use fedimint_bitcoind::DynBitcoindRpc;
use fedimint_client_module::module::recovery::RecoveryProgress;
use fedimint_client_module::module::{
    ClientContextIface, ClientModule, ClientModuleRegistry, DynClientModule, FinalClientIface,
    IClientModule, IdxRange, OutPointRange, PrimaryModulePriority,
};
use fedimint_client_module::oplog::IOperationLog;
use fedimint_client_module::secret::{PlainRootSecretStrategy, RootSecretStrategy as _};
use fedimint_client_module::sm::executor::{ActiveStateKey, IExecutor, InactiveStateKey};
use fedimint_client_module::sm::{ActiveStateMeta, DynState, InactiveStateMeta};
use fedimint_client_module::transaction::{
    FeeQuote, FeeQuoteRequest, TRANSACTION_SUBMISSION_MODULE_INSTANCE, TransactionBuilder,
    TxSubmissionStates, TxSubmissionStatesSM,
};
use fedimint_client_module::{
    AddStateMachinesResult, ClientModuleInstance, GetInviteCodeRequest, ModuleGlobalContextGen,
    ModuleRecoveryCompleted, TransactionUpdates, TxCreatedEvent,
};
use fedimint_connectors::{ConnectorRegistry, PeerStatus};
use fedimint_core::config::{
    ClientConfig, FederationId, GlobalClientConfig, JsonClientConfig, ModuleInitRegistry,
};
use fedimint_core::core::{DynInput, DynOutput, ModuleInstanceId, ModuleKind, OperationId};
use fedimint_core::db::{
    AutocommitError, Database, DatabaseRecord, DatabaseTransaction,
    IDatabaseTransactionOpsCore as _, IDatabaseTransactionOpsCoreTyped as _, NonCommittable,
};
use fedimint_core::encoding::{Decodable, Encodable};
use fedimint_core::endpoint_constants::{CLIENT_CONFIG_ENDPOINT, VERSION_ENDPOINT};
use fedimint_core::envs::is_running_in_test_env;
use fedimint_core::invite_code::InviteCode;
use fedimint_core::module::registry::{ModuleDecoderRegistry, ModuleRegistry};
use fedimint_core::module::{
    AmountUnit, Amounts, ApiRequestErased, ApiVersion, MultiApiVersion,
    SupportedApiVersionsSummary, SupportedCoreApiVersions, SupportedModuleApiVersions,
};
use fedimint_core::net::api_announcement::SignedApiAnnouncement;
use fedimint_core::runtime::sleep;
use fedimint_core::task::{
    Elapsed, MaybeSend, MaybeSync, ShuttingDownError, TaskGroup, TaskHandle,
};
use fedimint_core::transaction::Transaction;
use fedimint_core::util::backoff_util::custom_backoff;
use fedimint_core::util::{
    BoxStream, FmtCompact as _, FmtCompactAnyhow as _, SafeUrl, backoff_util, retry,
};
use fedimint_core::{
    Amount, ChainId, NumPeers, OutPoint, PeerId, apply, async_trait_maybe_send, maybe_add_send,
    maybe_add_send_sync, runtime,
};
use fedimint_derive_secret::DerivableSecret;
use fedimint_eventlog::{
    DBTransactionEventLogExt as _, DynEventLogTrimableTracker, Event, EventKind, EventLogEntry,
    EventLogId, EventLogTrimableId, EventLogTrimableTracker, EventPersistence, PersistedLogEntry,
};
use fedimint_logging::{LOG_CLIENT, LOG_CLIENT_NET_API, LOG_CLIENT_RECOVERY};
use futures::stream::FuturesUnordered;
use futures::{Stream, StreamExt as _};
use global_ctx::ModuleGlobalClientContext;
use serde::{Deserialize, Serialize};
use tokio::sync::{broadcast, oneshot, watch};
use tokio_stream::wrappers::WatchStream;
use tracing::{Span, debug, info, warn};

use crate::ClientBuilder;
use crate::api_announcements::{ApiAnnouncementPrefix, get_api_urls};
use crate::backup::Metadata;
use crate::client::event_log::DefaultApplicationEventLogKey;
use crate::db::{
    ApiSecretKey, CachedApiVersionSet, CachedApiVersionSetKey, ChainIdKey,
    ChronologicalOperationLogKey, ClientConfigKey, ClientMetadataKey, ClientModuleRecovery,
    ClientModuleRecoveryState, EncodedClientSecretKey, OperationLogKey, PeerLastApiVersionsSummary,
    PeerLastApiVersionsSummaryKey, PendingClientConfigKey, TransactionFeesKey,
    apply_migrations_core_client_dbtx, get_decoded_client_secret, verify_client_db_integrity_dbtx,
};
use crate::meta::MetaService;
use crate::module_init::{ClientModuleInitRegistry, DynClientModuleInit, IClientModuleInit};
use crate::oplog::OperationLog;
use crate::sm::executor::{
    ActiveModuleOperationStateKeyPrefix, ActiveOperationStateKeyPrefix, Executor,
    InactiveModuleOperationStateKeyPrefix, InactiveOperationStateKeyPrefix,
};

pub(crate) mod builder;
pub(crate) mod event_log;
pub(crate) mod global_ctx;
pub(crate) mod handle;

#[cfg(test)]
mod tests;

/// List of core api versions supported by the implementation.
/// Notably `major` version is the one being supported, and corresponding
/// `minor` version is the one required (for given `major` version).
const SUPPORTED_CORE_API_VERSIONS: &[fedimint_core::module::ApiVersion] =
    &[ApiVersion { major: 0, minor: 0 }];

struct FinalizedTransaction {
    transaction: Transaction,
    states: Vec<DynState>,
    change_range: Range<u64>,
    fees: Amounts,
}

/// Primary module candidates at specific priority level
#[derive(Default)]
pub(crate) struct PrimaryModuleCandidates {
    /// Modules that listed specific units they handle
    specific: BTreeMap<AmountUnit, Vec<ModuleInstanceId>>,
    /// Modules handling any unit
    wildcard: Vec<ModuleInstanceId>,
}

/// An in-progress module recovery future, resolving to the amount recovered
/// from the module (if it tracks one) once recovery completes.
pub(crate) type ModuleRecoveryFuture =
    Pin<Box<maybe_add_send!(dyn Future<Output = anyhow::Result<Option<Amount>>>)>>;

/// The state of a single module's recovery, as tracked by the client.
///
/// [`RecoveryProgress`] can only ever express how far a recovery got, never
/// that it gave up, which makes a failed recovery indistinguishable from a
/// merely slow one and blocks every waiter on the outcome forever. Carrying the
/// failure as a state of the very thing the progress describes makes the
/// contradictory "done and failed" unrepresentable for a module, so a recovery
/// that terminally fails resolves the waiters it used to strand. Only a
/// terminal failure does: a recovery that keeps retrying instead of giving up,
/// as fetching the federation's history does, resolves nothing, and its waiters
/// keep waiting.
///
/// Only the APIs returning a result observe the failure. The ones exposing the
/// progress itself, like [`Client::has_pending_recoveries`] and
/// [`Client::subscribe_to_recovery_progress`], keep reporting a failed recovery
/// as not done, which it is.
///
/// Internal: the failure reaches callers as the error of the waiting API, not
/// as this type, so it stays crate-private rather than committing an
/// accessorless public type to the API surface.
///
/// The module this describes is the key of the map this is stored in, so it is
/// deliberately not repeated here: a copy could contradict the key it is
/// filed under.
#[derive(Clone, Debug)]
pub(crate) enum RecoveryStatus {
    /// The latest progress of a recovery that has not failed.
    ///
    /// A successfully completed recovery is represented by a done progress.
    InProgress(RecoveryProgress),
    /// The recovery terminally failed at `last_progress`, which is kept so the
    /// progress-reporting APIs can keep describing the module.
    Failed {
        last_progress: RecoveryProgress,
        error: String,
    },
}

impl RecoveryStatus {
    /// Whether the module's recovery completed successfully.
    ///
    /// A failure is terminal but never done: the module recovered nothing
    /// beyond the progress it got to.
    pub(crate) fn is_successfully_done(&self) -> bool {
        match self {
            Self::InProgress(progress) => progress.is_done(),
            Self::Failed { .. } => false,
        }
    }

    /// The progress view of the status, which is all a progress-reporting API
    /// can express.
    pub(crate) fn progress(&self) -> RecoveryProgress {
        match self {
            Self::InProgress(progress)
            | Self::Failed {
                last_progress: progress,
                ..
            } => *progress,
        }
    }
}

/// Main client type
///
/// A handle and API to interacting with a single federation. End user
/// applications that want to support interacting with multiple federations at
/// the same time, will need to instantiate and manage multiple instances of
/// this struct.
///
/// Under the hood it is starting and managing service tasks, state machines,
/// database and other resources required.
///
/// This type is shared externally and internally, and
/// [`crate::ClientHandle`] is responsible for external lifecycle management
/// and resource freeing of the [`Client`].
pub struct Client {
    final_client: FinalClientIface,
    config: tokio::sync::RwLock<ClientConfig>,
    api_secret: Option<String>,
    decoders: ModuleDecoderRegistry,
    connectors: ConnectorRegistry,
    db: Database,
    federation_id: FederationId,
    federation_config_meta: BTreeMap<String, String>,
    primary_modules: BTreeMap<PrimaryModulePriority, PrimaryModuleCandidates>,
    pub(crate) modules: ClientModuleRegistry,
    module_inits: ClientModuleInitRegistry,
    executor: Executor,
    pub(crate) api: DynGlobalApi,
    root_secret: DerivableSecret,
    operation_log: OperationLog,
    secp_ctx: Secp256k1<secp256k1::All>,
    meta_service: Arc<MetaService>,

    task_group: TaskGroup,

    /// Long-lived span attached to every task spawned via [`Client::spawn`] /
    /// [`Client::spawn_cancellable`], so logs from background tasks carry the
    /// federation prefix.
    client_span: Span,

    /// Updates about the recovery status of every recovering module, keyed by
    /// module instance
    ///
    /// Keyed rather than a single slot because a `watch` channel only keeps its
    /// latest value: if one module's terminal state overwrote another's before
    /// a waiter observed it, [`Self::wait_for_module_kind_recovery`] could miss
    /// a failure of its own kind and block forever. Keeping a status per module
    /// keeps the outcome determinate no matter how many modules fail.
    ///
    /// A [`RecoveryStatus::Failed`] is intentionally not persisted: on a
    /// restart the recovery is simply attempted again, which is the right thing
    /// to do for a transient cause.
    client_recovery_status_receiver: watch::Receiver<BTreeMap<ModuleInstanceId, RecoveryStatus>>,

    /// Internal client sender to wake up log ordering task every time a
    /// (unuordered) log event is added.
    log_ordering_wakeup_tx: watch::Sender<()>,
    /// Receiver for events fired every time (ordered) log event is added.
    log_event_added_rx: watch::Receiver<()>,
    log_event_added_transient_tx: broadcast::Sender<EventLogEntry>,
    request_hook: ApiRequestHook,
    iroh_enable_dht: bool,
    iroh_enable_next: bool,
    /// User-provided Bitcoin RPC client for modules to use
    ///
    /// Stored here for potential future access; currently passed to modules
    /// during initialization.
    #[allow(dead_code)]
    user_bitcoind_rpc: Option<DynBitcoindRpc>,
    /// User-provided Bitcoin RPC factory for when ChainId is not available
    ///
    /// This is used as a fallback when the federation doesn't support ChainId.
    /// Modules can call this with a URL from their config to get an RPC client.
    pub(crate) user_bitcoind_rpc_no_chain_id:
        Option<fedimint_client_module::module::init::BitcoindRpcNoChainIdFactory>,
}

#[derive(Debug, Serialize, Deserialize)]
struct ListOperationsParams {
    limit: Option<usize>,
    last_seen: Option<ChronologicalOperationLogKey>,
}

pub const DEFAULT_EVENT_LOG_PAGE_SIZE: u64 = 100;
pub const MAX_EVENT_LOG_PAGE_SIZE: u64 = 10_000;

#[derive(Debug, Clone, Serialize, Deserialize)]
struct GetEventLogRequest {
    pos: Option<EventLogId>,
    limit: Option<u64>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GetOperationIdRequest {
    operation_id: OperationId,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GetBalanceChangesRequest {
    #[serde(default = "AmountUnit::bitcoin")]
    unit: AmountUnit,
}

impl Client {
    /// Initialize a client builder that can be configured to create a new
    /// client.
    pub async fn builder() -> anyhow::Result<ClientBuilder> {
        Ok(ClientBuilder::new())
    }

    pub fn api(&self) -> &(dyn IGlobalFederationApi + 'static) {
        self.api.as_ref()
    }

    pub fn api_clone(&self) -> DynGlobalApi {
        self.api.clone()
    }

    /// Returns a stream that emits the current connection status of all peers
    /// whenever any peer's status changes. Emits initial state immediately.
    pub fn connection_status_stream(&self) -> impl Stream<Item = BTreeMap<PeerId, PeerStatus>> {
        self.api.connection_status_stream()
    }

    /// Establishes connections to all federation guardians once.
    ///
    /// Spawns tasks to connect to each guardian in the federation. Unlike
    /// [`Self::spawn_federation_reconnect`], this only attempts to establish
    /// connections once and completes - it does not maintain or reconnect.
    ///
    /// Useful for warming up connections before making API calls.
    pub fn federation_reconnect(&self) {
        let peers: Vec<PeerId> = self.api.all_peers().iter().copied().collect();

        for peer_id in peers {
            let api = self.api.clone();
            self.spawn_cancellable(format!("federation-reconnect-once-{peer_id}"), async move {
                if let Err(e) = api.get_peer_connection(peer_id).await {
                    debug!(
                        target: LOG_CLIENT_NET_API,
                        %peer_id,
                        err = %e.fmt_compact(),
                        "Failed to connect to peer"
                    );
                }
            });
        }
    }

    /// Spawns background tasks that proactively maintain connections to all
    /// federation guardians unconditionally.
    ///
    /// For each guardian, a task loops: establishes a connection, waits for it
    /// to disconnect, then reconnects.
    ///
    /// The tasks are cancellable and will be terminated when the client shuts
    /// down.
    ///
    /// By default [`Client`] creates connections on demand only, and share
    /// them as long as they are alive.
    ///
    /// Reconnecting continuously might increase data and battery usage,
    /// but potentially improve UX, depending on the time it takes to establish
    /// a new network connection in given network conditions.
    ///
    /// Downstream users are encouraged to implement their own version of
    /// this function, e.g. by reconnecting only when it is anticipated
    /// that connection might be needed, or alternatively pre-warm
    /// connections by calling [`Self::federation_reconnect`] when it seems
    /// worthwhile.
    pub fn spawn_federation_reconnect(&self) {
        let peers: Vec<PeerId> = self.api.all_peers().iter().copied().collect();

        for peer_id in peers {
            let api = self.api.clone();
            self.spawn_cancellable(format!("federation-reconnect-{peer_id}"), async move {
                loop {
                    match api.get_peer_connection(peer_id).await {
                        Ok(conn) => {
                            conn.await_disconnection().await;
                        }
                        Err(e) => {
                            // Connection failed, backoff is handled inside
                            // get_or_create_connection
                            debug!(
                                target: LOG_CLIENT_NET_API,
                                %peer_id,
                                err = %e.fmt_compact(),
                                "Failed to connect to peer, will retry"
                            );
                        }
                    }
                }
            });
        }
    }

    /// Get the [`TaskGroup`] that is tied to Client's lifetime.
    pub fn task_group(&self) -> &TaskGroup {
        &self.task_group
    }

    /// Construct the long-lived span attached to all tasks spawned by this
    /// client.
    ///
    /// `parent: None` keeps the span tree shallow; `runtime::spawn` already
    /// wraps each task in its own `spawn(task=…)` span, so log events from
    /// these tasks carry both `task` and `fed_id`.
    pub(crate) fn make_client_span(federation_id: FederationId) -> Span {
        tracing::info_span!(
            target: LOG_CLIENT,
            parent: None,
            "client",
            fed_id = %federation_id.to_prefix(),
        )
    }

    /// Spawn a cancellable task on the client's task group, instrumented with
    /// the client's [`Span`] so all events from the task carry `fed_id`.
    pub(crate) fn spawn_cancellable<R>(
        &self,
        name: impl Into<String>,
        future: impl Future<Output = R> + MaybeSend + 'static,
    ) -> oneshot::Receiver<Result<R, ShuttingDownError>>
    where
        R: MaybeSend + 'static,
    {
        self.task_group
            .spawn_cancellable_with_span(self.client_span.clone(), name, future)
    }

    /// Spawn a task on the client's task group, parented to the client's
    /// [`Span`] so all events from the task carry `fed_id` (including the
    /// task lifecycle events emitted by [`TaskGroup`] itself).
    pub(crate) fn spawn<Fut, R>(
        &self,
        name: impl Into<String>,
        f: impl FnOnce(TaskHandle) -> Fut + MaybeSend + 'static,
    ) -> oneshot::Receiver<R>
    where
        Fut: Future<Output = R> + MaybeSend + 'static,
        R: MaybeSend + 'static,
    {
        self.task_group
            .spawn_with_span(self.client_span.clone(), name, f)
    }

    /// Returns all registered Prometheus metrics encoded in text format.
    ///
    /// This can be used by downstream clients to expose metrics via their own
    /// HTTP server or print them for debugging purposes.
    pub fn get_metrics() -> anyhow::Result<String> {
        fedimint_metrics::get_metrics()
    }

    /// Useful for our CLI tooling, not meant for external use
    #[doc(hidden)]
    pub fn executor(&self) -> &Executor {
        &self.executor
    }

    pub async fn get_config_from_db(db: &Database) -> Option<ClientConfig> {
        let mut dbtx = db.begin_transaction_nc().await;
        dbtx.get_value(&ClientConfigKey).await
    }

    pub async fn get_pending_config_from_db(db: &Database) -> Option<ClientConfig> {
        let mut dbtx = db.begin_transaction_nc().await;
        dbtx.get_value(&PendingClientConfigKey).await
    }

    pub async fn get_api_secret_from_db(db: &Database) -> Option<String> {
        let mut dbtx = db.begin_transaction_nc().await;
        dbtx.get_value(&ApiSecretKey).await
    }

    pub async fn store_encodable_client_secret<T: Encodable>(
        db: &Database,
        secret: T,
    ) -> anyhow::Result<()> {
        let mut dbtx = db.begin_transaction().await;

        // Don't overwrite an existing secret
        if dbtx.get_value(&EncodedClientSecretKey).await.is_some() {
            bail!("Encoded client secret already exists, cannot overwrite")
        }

        let encoded_secret = T::consensus_encode_to_vec(&secret);
        dbtx.insert_entry(&EncodedClientSecretKey, &encoded_secret)
            .await;
        dbtx.commit_tx().await;
        Ok(())
    }

    pub async fn load_decodable_client_secret<T: Decodable>(db: &Database) -> anyhow::Result<T> {
        let Some(secret) = Self::load_decodable_client_secret_opt(db).await? else {
            bail!("Encoded client secret not present in DB")
        };

        Ok(secret)
    }
    pub async fn load_decodable_client_secret_opt<T: Decodable>(
        db: &Database,
    ) -> anyhow::Result<Option<T>> {
        let mut dbtx = db.begin_transaction_nc().await;

        let client_secret = dbtx.get_value(&EncodedClientSecretKey).await;

        Ok(match client_secret {
            Some(client_secret) => Some(
                T::consensus_decode_whole(&client_secret, &ModuleRegistry::default())
                    .context("Decoding failed")?,
            ),
            None => None,
        })
    }

    pub async fn load_or_generate_client_secret(db: &Database) -> anyhow::Result<[u8; 64]> {
        let client_secret = match Self::load_decodable_client_secret::<[u8; 64]>(db).await {
            Ok(secret) => secret,
            _ => {
                let secret = PlainRootSecretStrategy::random(&mut thread_rng());
                Self::store_encodable_client_secret(db, secret)
                    .await
                    .expect("Storing client secret must work");
                secret
            }
        };
        Ok(client_secret)
    }

    pub async fn is_initialized(db: &Database) -> bool {
        let mut dbtx = db.begin_transaction_nc().await;
        dbtx.raw_get_bytes(&[ClientConfigKey::DB_PREFIX])
            .await
            .expect("Unrecoverable error occurred while reading and entry from the database")
            .is_some()
    }

    pub fn start_executor(self: &Arc<Self>) {
        self.client_span.in_scope(|| {
            debug!(
                target: LOG_CLIENT,
                "Starting fedimint client executor",
            );
        });
        self.executor
            .start_executor(self.context_gen(), self.client_span.clone());
    }

    pub fn federation_id(&self) -> FederationId {
        self.federation_id
    }

    fn context_gen(self: &Arc<Self>) -> ModuleGlobalContextGen {
        let client_inner = Arc::downgrade(self);
        Arc::new(move |module_instance, operation| {
            ModuleGlobalClientContext {
                client: client_inner
                    .clone()
                    .upgrade()
                    .expect("ModuleGlobalContextGen called after client was dropped"),
                module_instance_id: module_instance,
                operation,
            }
            .into()
        })
    }

    pub async fn config(&self) -> ClientConfig {
        self.config.read().await.clone()
    }

    // TODO: change to `-> Option<&str>`
    pub fn api_secret(&self) -> &Option<String> {
        &self.api_secret
    }

    /// Returns the core API version that the federation supports
    ///
    /// This reads from the cached version stored during client initialization.
    /// If no cache is available (e.g., during initial setup), returns a default
    /// version (0, 0).
    pub async fn core_api_version(&self) -> ApiVersion {
        // Try to get from cache. If not available, return a conservative
        // default. The cache should always be populated after successful client init.
        self.db
            .begin_transaction_nc()
            .await
            .get_value(&CachedApiVersionSetKey)
            .await
            .map(|cached: CachedApiVersionSet| cached.0.core)
            .unwrap_or(ApiVersion { major: 0, minor: 0 })
    }

    /// Returns the chain ID (bitcoin block hash at height 1) from the
    /// federation
    ///
    /// This is cached in the database after the first successful fetch.
    /// The chain ID uniquely identifies which bitcoin network the federation
    /// operates on (mainnet, testnet, signet, regtest).
    pub async fn chain_id(&self) -> anyhow::Result<ChainId> {
        // Check cache first
        if let Some(chain_id) = self
            .db
            .begin_transaction_nc()
            .await
            .get_value(&ChainIdKey)
            .await
        {
            return Ok(chain_id);
        }

        // Fetch from federation with consensus
        let chain_id = self.api.chain_id().await?;

        // Cache the result
        let mut dbtx = self.db.begin_transaction().await;
        dbtx.insert_entry(&ChainIdKey, &chain_id).await;
        dbtx.commit_tx().await;

        Ok(chain_id)
    }

    pub fn decoders(&self) -> &ModuleDecoderRegistry {
        &self.decoders
    }

    /// Returns a reference to the module, panics if not found
    fn get_module(&self, instance: ModuleInstanceId) -> &maybe_add_send_sync!(dyn IClientModule) {
        self.try_get_module(instance)
            .expect("Module instance not found")
    }

    fn try_get_module(
        &self,
        instance: ModuleInstanceId,
    ) -> Option<&maybe_add_send_sync!(dyn IClientModule)> {
        Some(self.modules.get(instance)?.as_ref())
    }

    pub fn has_module(&self, instance: ModuleInstanceId) -> bool {
        self.modules.get(instance).is_some()
    }

    /// Returns the input amount and output amount of a transaction
    ///
    /// # Panics
    /// If any of the input or output versions in the transaction builder are
    /// unknown by the respective module.
    fn transaction_builder_get_balance(&self, builder: &TransactionBuilder) -> (Amounts, Amounts) {
        // FIXME: prevent overflows, currently not suitable for untrusted input
        let mut in_amounts = Amounts::ZERO;
        let mut out_amounts = Amounts::ZERO;
        let mut fee_amounts = Amounts::ZERO;

        for input in builder.inputs() {
            let module = self.get_module(input.input.module_instance_id());

            let item_fees = module.input_fee(&input.amounts, &input.input).expect(
                "We only build transactions with input versions that are supported by the module",
            );

            in_amounts.checked_add_mut(&input.amounts);
            fee_amounts.checked_add_mut(&item_fees);
        }

        for output in builder.outputs() {
            let module = self.get_module(output.output.module_instance_id());

            let item_fees = module.output_fee(&output.amounts, &output.output).expect(
                "We only build transactions with output versions that are supported by the module",
            );

            out_amounts.checked_add_mut(&output.amounts);
            fee_amounts.checked_add_mut(&item_fees);
        }

        out_amounts.checked_add_mut(&fee_amounts);
        (in_amounts, out_amounts)
    }

    pub fn get_internal_payment_markers(&self) -> anyhow::Result<(PublicKey, u64)> {
        Ok((self.federation_id().to_fake_ln_pub_key(&self.secp_ctx)?, 0))
    }

    /// Get metadata value from the federation config itself
    pub fn get_config_meta(&self, key: &str) -> Option<String> {
        self.federation_config_meta.get(key).cloned()
    }

    pub(crate) fn root_secret(&self) -> DerivableSecret {
        self.root_secret.clone()
    }

    pub async fn add_state_machines(
        &self,
        dbtx: &mut DatabaseTransaction<'_>,
        states: Vec<DynState>,
    ) -> AddStateMachinesResult {
        self.executor.add_state_machines_dbtx(dbtx, states).await
    }

    // TODO: implement as part of [`OperationLog`]
    pub async fn get_active_operations(&self) -> HashSet<OperationId> {
        let active_states = self.executor.get_active_states().await;
        let mut active_operations = HashSet::with_capacity(active_states.len());
        let mut dbtx = self.db().begin_transaction_nc().await;
        for (state, _) in active_states {
            let operation_id = state.operation_id();
            if dbtx
                .get_value(&OperationLogKey { operation_id })
                .await
                .is_some()
            {
                active_operations.insert(operation_id);
            }
        }
        active_operations
    }

    pub fn operation_log(&self) -> &OperationLog {
        &self.operation_log
    }

    /// Get the meta manager to read meta fields.
    pub fn meta_service(&self) -> &Arc<MetaService> {
        &self.meta_service
    }

    /// Get the meta manager to read meta fields.
    pub async fn get_meta_expiration_timestamp(&self) -> Option<SystemTime> {
        let meta_service = self.meta_service();
        let ts = meta_service
            .get_field::<u64>(self.db(), "federation_expiry_timestamp")
            .await
            .and_then(|v| v.value)?;
        Some(UNIX_EPOCH + Duration::from_secs(ts))
    }

    /// Adds funding to a transaction or removes over-funding via change.
    async fn finalize_transaction(
        &self,
        dbtx: &mut DatabaseTransaction<'_>,
        operation_id: OperationId,
        mut partial_transaction: TransactionBuilder,
    ) -> anyhow::Result<FinalizedTransaction> {
        let (in_amounts, out_amounts) = self.transaction_builder_get_balance(&partial_transaction);

        let mut added_inputs_bundles = vec![];
        let mut added_outputs_bundles = vec![];

        // The way currently things are implemented is OK for modules which can
        // collect a fee relative to one being used, but will break down in any
        // fancy scenarios. Future TODOs:
        //
        // * create_final_inputs_and_outputs needs to get broken down, so we can use
        //   primary modules using priorities (possibly separate prios for inputs and
        //   outputs to be able to drain modules, etc.); we need the split "check if
        //   possible" and "take" steps,
        // * extra inputs and outputs adding fees needs to be taken into account,
        //   possibly with some looping
        for unit in in_amounts.units().union(&out_amounts.units()) {
            let input_amount = in_amounts.get(unit).copied().unwrap_or_default();
            let output_amount = out_amounts.get(unit).copied().unwrap_or_default();
            if input_amount == output_amount {
                continue;
            }

            let Some((module_id, module)) = self.primary_module_for_unit(*unit) else {
                bail!("No module to balance a partial transaction (affected unit: {unit:?}");
            };

            let (added_input_bundle, added_output_bundle) = module
                .create_final_inputs_and_outputs(
                    module_id,
                    dbtx,
                    operation_id,
                    *unit,
                    input_amount,
                    output_amount,
                )
                .await?;

            added_inputs_bundles.push(added_input_bundle);
            added_outputs_bundles.push(added_output_bundle);
        }

        // This is the range of  outputs that will be added to the transaction
        // in order to balance it. Notice that it may stay empty in case the transaction
        // is already balanced.
        let change_range = Range {
            start: partial_transaction.outputs().count() as u64,
            end: (partial_transaction.outputs().count() as u64
                + added_outputs_bundles
                    .iter()
                    .map(|output| output.outputs().len() as u64)
                    .sum::<u64>()),
        };

        for added_inputs in added_inputs_bundles {
            partial_transaction = partial_transaction.with_inputs(added_inputs);
        }

        for added_outputs in added_outputs_bundles {
            partial_transaction = partial_transaction.with_outputs(added_outputs);
        }

        let (input_amounts, output_amounts) =
            self.transaction_builder_get_balance(&partial_transaction);

        for (unit, output_amount) in output_amounts {
            let input_amount = input_amounts.get(&unit).copied().unwrap_or_default();

            assert!(input_amount >= output_amount, "Transaction is underfunded");
        }

        // Compute fees as the difference between total input and output amounts.
        // This captures both explicit federation fees and any overpayment due to
        // denomination constraints.
        let fees = {
            let mut input_total = Amounts::ZERO;
            for input in partial_transaction.inputs() {
                input_total
                    .checked_add_mut(&input.amounts)
                    .expect("Own transaction amounts don't overflow");
            }
            let mut output_total = Amounts::ZERO;
            for output in partial_transaction.outputs() {
                output_total
                    .checked_add_mut(&output.amounts)
                    .expect("Own transaction amounts don't overflow");
            }
            input_total
                .checked_sub(&output_total)
                .expect("Inputs >= outputs for own transactions")
        };

        let (transaction, states) = partial_transaction.build(&self.secp_ctx, thread_rng());

        Ok(FinalizedTransaction {
            transaction,
            states,
            change_range,
            fees,
        })
    }

    /// Computes the fee that finalizing and submitting a transaction with the
    /// explicit items described by `request` would incur, without submitting
    /// anything.
    ///
    /// This is the read-only twin of `Self::finalize_transaction`: it runs
    /// the exact same primary-module balancing
    /// (`create_final_inputs_and_outputs`, including funding selection /
    /// consolidation / rebalancing) against a non-committable transaction
    /// that is dropped rather than committed, so the client's funds are
    /// read but left untouched. The quote is point-in-time: it depends on
    /// the current inventory and can move as funds change.
    ///
    /// The mechanism is module-agnostic — [`FeeQuoteRequest`] only summarizes
    /// the explicit side (gross amounts + federation fees) of whichever module
    /// is quoting (mint, lightning, wallet, …), and the change is generated by
    /// the primary module of each affected unit. The request and the resulting
    /// breakdown are multi-unit, so an operation spanning several units (e.g.
    /// Bitcoin plus a custom currency) is quoted in one call, mirroring how
    /// `finalize_transaction` balances each unit independently.
    pub async fn fee_quote(
        &self,
        operation_id: OperationId,
        request: FeeQuoteRequest,
    ) -> anyhow::Result<FeeQuote> {
        let FeeQuoteRequest {
            input_amount,
            output_amount,
            input_fee,
            output_fee,
        } = request;

        // Start the totals from the explicit side; the change the primary
        // modules add below is a disjoint set of items, accumulated into these
        // same totals.
        let mut gross_input = input_amount.clone();
        let mut gross_output = output_amount.clone();
        let mut input_fees = input_fee.clone();
        let mut output_fees = output_fee.clone();

        // The primary modules balance the transaction by minting change (and
        // possibly pulling in extra inputs). The amounts they must balance
        // mirror `finalize_transaction`: per unit the input side is the gross
        // input value, the output side carries the gross output value plus
        // every explicit fee (fees consume funding on the output side).
        let balance_input = input_amount;
        let balance_output = output_amount
            .checked_add(&input_fee)
            .and_then(|amounts| amounts.checked_add(&output_fee))
            .expect("explicit amounts and fees cannot overflow an Amounts");

        // Non-committable: the balancing writes (e.g. removing consolidated
        // notes) are discarded on drop, so this is a pure dry-run over the real
        // inventory.
        let mut dbtx = self.db.begin_transaction_nc().await;

        // Balance each affected unit independently via its own primary module,
        // exactly as `finalize_transaction` does over the real transaction.
        for unit in balance_input.units().union(&balance_output.units()) {
            let balance_input_amount = balance_input.get(unit).copied().unwrap_or_default();
            let balance_output_amount = balance_output.get(unit).copied().unwrap_or_default();
            if balance_input_amount == balance_output_amount {
                continue;
            }

            let Some((module_id, module)) = self.primary_module_for_unit(*unit) else {
                bail!("No module to balance a partial transaction (affected unit: {unit:?}");
            };

            let (change_input, change_output) = module
                .create_final_inputs_and_outputs(
                    module_id,
                    &mut dbtx.to_ref_nc(),
                    operation_id,
                    *unit,
                    balance_input_amount,
                    balance_output_amount,
                )
                .await?;

            // Fold the change into the totals. These are a disjoint set of items
            // from the explicit ones (the primary module only sees the scalar
            // amounts to balance, never the explicit items), so this is not
            // double-counting. Iterate the bundles the way `finalize_transaction`
            // would, looking each fee up via the module that owns the item.
            for input in change_input.inputs() {
                let module = self.get_module(input.input.module_instance_id());
                let fee = module
                    .input_fee(&input.amounts, &input.input)
                    .expect("Primary module must know its own change input fees");
                gross_input.checked_add_mut(&input.amounts);
                input_fees.checked_add_mut(&fee);
            }

            for output in change_output.outputs() {
                let module = self.get_module(output.output.module_instance_id());
                let fee = module
                    .output_fee(&output.amounts, &output.output)
                    .expect("Primary module must know its own change output fees");
                gross_output.checked_add_mut(&output.amounts);
                output_fees.checked_add_mut(&fee);
            }
        }

        // Mark the dropped dbtx as intentionally uncommitted so the commit
        // tracker doesn't warn.
        dbtx.ignore_uncommitted();

        // Per unit: net wallet gain = produced outputs − consumed inputs. The
        // total fee is everything the gross input value did not become a net
        // gain; whatever is left after the input/output fees is
        // sub-denomination dust.
        let mut dust = Amounts::ZERO;
        for unit in gross_input.units().union(&gross_output.units()) {
            let total = gross_input
                .get(unit)
                .copied()
                .unwrap_or_default()
                .saturating_sub(gross_output.get(unit).copied().unwrap_or_default());
            let fees = input_fees.get(unit).copied().unwrap_or_default()
                + output_fees.get(unit).copied().unwrap_or_default();
            dust = dust
                .checked_add_unit(total.saturating_sub(fees), *unit)
                .expect("dust cannot overflow an Amounts");
        }

        Ok(FeeQuote {
            input: input_fees,
            output: output_fees,
            dust,
        })
    }

    /// Add funding and/or change to the transaction builder as needed, finalize
    /// the transaction and submit it to the federation.
    ///
    /// ## Cancel safety
    /// This method is cancel safe. It performs all its work (funding the
    /// inputs, registering the state machines, and creating the operation log
    /// entry) inside a single `autocommit` transaction, so dropping the future
    /// either commits that unit in full or leaves the database untouched.
    /// Submission to the federation is not awaited here, it is carried out by
    /// the transaction-submission state machine in the executor, so cancelling
    /// does not abort an already-committed submission.
    ///
    /// ## Errors
    /// The function will return an error if the operation with given ID already
    /// exists.
    ///
    /// ## Panics
    /// The function will panic if the database transaction collides with
    /// other and fails with others too often, this should not happen except for
    /// excessively concurrent scenarios.
    pub async fn finalize_and_submit_transaction<F, M>(
        &self,
        operation_id: OperationId,
        operation_type: &str,
        operation_meta_gen: F,
        tx_builder: TransactionBuilder,
    ) -> anyhow::Result<OutPointRange>
    where
        F: Fn(OutPointRange) -> M + Clone + MaybeSend + MaybeSync,
        M: serde::Serialize + MaybeSend,
    {
        let operation_type = operation_type.to_owned();

        let autocommit_res = self
            .db
            .autocommit(
                |dbtx, _| {
                    let operation_type = operation_type.clone();
                    let tx_builder = tx_builder.clone();
                    let operation_meta_gen = operation_meta_gen.clone();
                    Box::pin(async move {
                        self.finalize_and_submit_transaction_dbtx(
                            dbtx,
                            operation_id,
                            &operation_type,
                            operation_meta_gen,
                            tx_builder,
                        )
                        .await
                    })
                },
                Some(100), // TODO: handle what happens after 100 retries
            )
            .await;

        match autocommit_res {
            Ok(txid) => Ok(txid),
            Err(AutocommitError::ClosureError { error, .. }) => Err(error),
            Err(AutocommitError::CommitFailed {
                attempts,
                last_error,
            }) => panic!(
                "Failed to commit tx submission dbtx after {attempts} attempts: {last_error}"
            ),
        }
    }

    /// See [`Self::finalize_and_submit_transaction`], just inside a database
    /// transaction.
    ///
    /// ## Cancel safety
    /// Unlike [`Self::finalize_and_submit_transaction`], this does not own the
    /// transaction: all writes go to the passed `dbtx`, so its cancel safety
    /// is that of the caller's transaction. The caller is responsible for
    /// committing (or rolling back) `dbtx` atomically.
    pub async fn finalize_and_submit_transaction_dbtx<F, M>(
        &self,
        dbtx: &mut DatabaseTransaction<'_>,
        operation_id: OperationId,
        operation_type: &str,
        operation_meta_gen: F,
        tx_builder: TransactionBuilder,
    ) -> anyhow::Result<OutPointRange>
    where
        F: FnOnce(OutPointRange) -> M + MaybeSend,
        M: serde::Serialize + MaybeSend,
    {
        if Client::operation_exists_dbtx(dbtx, operation_id).await {
            bail!("There already exists an operation with id {operation_id:?}")
        }

        let out_point_range = self
            .finalize_and_submit_transaction_inner(dbtx, operation_id, tx_builder)
            .await?;

        self.operation_log()
            .add_operation_log_entry_dbtx(
                dbtx,
                operation_id,
                operation_type,
                operation_meta_gen(out_point_range),
            )
            .await;

        Ok(out_point_range)
    }

    async fn finalize_and_submit_transaction_inner(
        &self,
        dbtx: &mut DatabaseTransaction<'_>,
        operation_id: OperationId,
        tx_builder: TransactionBuilder,
    ) -> anyhow::Result<OutPointRange> {
        let FinalizedTransaction {
            transaction,
            mut states,
            change_range,
            fees,
        } = self
            .finalize_transaction(&mut dbtx.to_ref_nc(), operation_id, tx_builder)
            .await?;

        if transaction.consensus_encode_to_vec().len() > Transaction::MAX_TX_SIZE {
            let inputs = transaction
                .inputs
                .iter()
                .map(DynInput::module_instance_id)
                .collect::<Vec<_>>();
            let outputs = transaction
                .outputs
                .iter()
                .map(DynOutput::module_instance_id)
                .collect::<Vec<_>>();
            warn!(
                target: LOG_CLIENT_NET_API,
                size=%transaction.consensus_encode_to_vec().len(),
                ?inputs,
                ?outputs,
                "Transaction too large",
            );
            debug!(target: LOG_CLIENT_NET_API, ?transaction, "transaction details");
            bail!(
                "The generated transaction would be rejected by the federation for being too large."
            );
        }

        let txid = transaction.tx_hash();

        debug!(
            target: LOG_CLIENT_NET_API,
            %txid,
            operation_id = %operation_id.fmt_short(),
            ?transaction,
            "Finalized and submitting transaction",
        );

        let tx_submission_sm = DynState::from_typed(
            TRANSACTION_SUBMISSION_MODULE_INSTANCE,
            TxSubmissionStatesSM {
                operation_id,
                state: TxSubmissionStates::Created(transaction),
            },
        );
        states.push(tx_submission_sm);

        self.executor.add_state_machines_dbtx(dbtx, states).await?;

        dbtx.insert_new_entry(&TransactionFeesKey(txid), &fees)
            .await;

        self.log_event_dbtx(dbtx, None, TxCreatedEvent { txid, operation_id })
            .await;

        Ok(OutPointRange::new(txid, IdxRange::from(change_range)))
    }

    async fn transaction_update_stream(
        &self,
        operation_id: OperationId,
    ) -> BoxStream<'static, TxSubmissionStatesSM> {
        self.executor
            .notifier()
            .module_notifier::<TxSubmissionStatesSM>(
                TRANSACTION_SUBMISSION_MODULE_INSTANCE,
                self.final_client.clone(),
            )
            .subscribe(operation_id)
            .await
    }

    pub async fn operation_exists(&self, operation_id: OperationId) -> bool {
        let mut dbtx = self.db().begin_transaction_nc().await;

        Client::operation_exists_dbtx(&mut dbtx, operation_id).await
    }

    pub async fn operation_exists_dbtx(
        dbtx: &mut DatabaseTransaction<'_>,
        operation_id: OperationId,
    ) -> bool {
        let active_state_exists = dbtx
            .find_by_prefix(&ActiveOperationStateKeyPrefix { operation_id })
            .await
            .next()
            .await
            .is_some();

        let inactive_state_exists = dbtx
            .find_by_prefix(&InactiveOperationStateKeyPrefix { operation_id })
            .await
            .next()
            .await
            .is_some();

        active_state_exists || inactive_state_exists
    }

    pub async fn has_active_states(&self, operation_id: OperationId) -> bool {
        self.db
            .begin_transaction_nc()
            .await
            .find_by_prefix(&ActiveOperationStateKeyPrefix { operation_id })
            .await
            .next()
            .await
            .is_some()
    }

    /// Calculates the federation fees paid in the course of the operation.
    ///
    /// Federation fees are fees paid to the federation (e.g. for ecash) and do
    /// not include fees paid to service providers like the Lightning gateway.
    /// These still have to be separately reported by the module.
    ///
    /// Fees are calculated by subtracting the total output amount from the
    /// total input amount, any difference is either due to direct fees or
    /// overpayment to avoid generating uneconomic outputs.
    ///
    /// The returned amount may still increase while the operation is active;
    /// use [`Client::has_active_states`] to check whether it is final.
    ///
    /// Returns `Ok(None)` if any transaction's fee data is missing (e.g. for
    /// operations created before this feature was enabled).
    ///
    /// # Errors
    /// Returns an error if the operation does not exist.
    pub async fn get_operation_fees(
        &self,
        operation_id: OperationId,
    ) -> anyhow::Result<Option<Amounts>> {
        if !self.operation_exists(operation_id).await {
            bail!("Operation does not exist");
        }

        let (active_states, inactive_states) =
            self.executor().get_operation_states(operation_id).await;

        let states = active_states
            .into_iter()
            .map(|(state, _)| state)
            .chain(inactive_states.into_iter().map(|(state, _)| state));

        let accepted_transactions = states
            .filter_map(|state| {
                let tx_state = state.as_any().downcast_ref::<TxSubmissionStatesSM>()?;

                match &tx_state.state {
                    TxSubmissionStates::Accepted(transaction_id) => Some(*transaction_id),
                    _ => None,
                }
            })
            .collect::<HashSet<_>>();

        // For each accepted transaction, look up its stored fees
        let mut dbtx = self.db.begin_transaction_nc().await;
        let mut total_fees = Amounts::ZERO;
        for txid in &accepted_transactions {
            let Some(fees) = dbtx.get_value(&TransactionFeesKey(*txid)).await else {
                return Ok(None);
            };
            total_fees = total_fees
                .checked_add(&fees)
                .expect("Fee amounts don't overflow in practice");
        }

        Ok(Some(total_fees))
    }

    /// Waits for an output from the primary module to reach its final
    /// state.
    pub async fn await_primary_bitcoin_module_output(
        &self,
        operation_id: OperationId,
        out_point: OutPoint,
    ) -> anyhow::Result<()> {
        self.primary_module_for_unit(AmountUnit::BITCOIN)
            .ok_or_else(|| anyhow!("No primary module available"))?
            .1
            .await_primary_module_output(operation_id, out_point)
            .await
    }

    /// Returns a reference to a typed module client instance by kind
    pub fn get_first_module<M: ClientModule>(
        &'_ self,
    ) -> anyhow::Result<ClientModuleInstance<'_, M>> {
        let module_kind = M::kind();
        let id = self
            .get_first_instance(&module_kind)
            .ok_or_else(|| format_err!("No modules found of kind {module_kind}"))?;
        let module: &M = self
            .try_get_module(id)
            .ok_or_else(|| format_err!("Unknown module instance {id}"))?
            .as_any()
            .downcast_ref::<M>()
            .ok_or_else(|| format_err!("Module is not of type {}", std::any::type_name::<M>()))?;
        let (db, _) = self.db().with_prefix_module_id(id);
        Ok(ClientModuleInstance {
            id,
            db,
            api: self.api().with_module(id),
            module,
        })
    }

    /// Returns an owned `Arc` to a typed module client instance by kind.
    ///
    /// Unlike [`Self::get_first_module`], this hands out a cloned `Arc` so the
    /// caller can hold the module independently of the `Client`'s lifetime.
    #[cfg(not(target_family = "wasm"))]
    pub fn get_first_module_arc<M: ClientModule>(&self) -> anyhow::Result<Arc<M>> {
        let module_kind = M::kind();
        let id = self
            .get_first_instance(&module_kind)
            .ok_or_else(|| format_err!("No modules found of kind {module_kind}"))?;
        let dyn_module = self
            .modules
            .get(id)
            .ok_or_else(|| format_err!("Unknown module instance {id}"))?;
        dyn_module
            .as_any_arc()
            .downcast::<M>()
            .map_err(|_| format_err!("Module is not of type {}", std::any::type_name::<M>()))
    }

    pub fn get_module_client_dyn(
        &self,
        instance_id: ModuleInstanceId,
    ) -> anyhow::Result<&maybe_add_send_sync!(dyn IClientModule)> {
        self.try_get_module(instance_id)
            .ok_or(anyhow!("Unknown module instance {}", instance_id))
    }

    pub fn db(&self) -> &Database {
        &self.db
    }

    pub fn endpoints(&self) -> &ConnectorRegistry {
        &self.connectors
    }

    /// Returns a stream of transaction updates for the given operation id that
    /// can later be used to watch for a specific transaction being accepted.
    pub async fn transaction_updates(&self, operation_id: OperationId) -> TransactionUpdates {
        TransactionUpdates {
            update_stream: self.transaction_update_stream(operation_id).await,
        }
    }

    /// Returns the instance id of the first module of the given kind.
    pub fn get_first_instance(&self, module_kind: &ModuleKind) -> Option<ModuleInstanceId> {
        self.modules
            .iter_modules()
            .find(|(_, kind, _module)| *kind == module_kind)
            .map(|(instance_id, _, _)| instance_id)
    }

    /// Returns the data from which the client's root secret is derived (e.g.
    /// BIP39 seed phrase struct).
    pub async fn root_secret_encoding<T: Decodable>(&self) -> anyhow::Result<T> {
        get_decoded_client_secret::<T>(self.db()).await
    }

    /// Waits for outputs from the primary module to reach its final
    /// state.
    pub async fn await_primary_bitcoin_module_outputs(
        &self,
        operation_id: OperationId,
        outputs: Vec<OutPoint>,
    ) -> anyhow::Result<()> {
        for out_point in outputs {
            self.await_primary_bitcoin_module_output(operation_id, out_point)
                .await?;
        }

        Ok(())
    }

    /// Returns the config of the client in JSON format.
    ///
    /// Compared to the consensus module format where module configs are binary
    /// encoded this format cannot be cryptographically verified but is easier
    /// to consume and to some degree human-readable.
    pub async fn get_config_json(&self) -> JsonClientConfig {
        self.config().await.to_json()
    }

    // Ideally this would not be in the API, but there's a lot of places where this
    // makes it easier.
    #[doc(hidden)]
    /// Like [`Self::get_balance`] but returns an error if primary module is not
    /// available
    pub async fn get_balance_for_btc(&self) -> anyhow::Result<Amount> {
        self.get_balance_for_unit(AmountUnit::BITCOIN).await
    }

    pub async fn get_balance_for_unit(&self, unit: AmountUnit) -> anyhow::Result<Amount> {
        let (id, module) = self
            .primary_module_for_unit(unit)
            .ok_or_else(|| anyhow!("Primary module not available"))?;
        Ok(module
            .get_balance(id, &mut self.db().begin_transaction_nc().await, unit)
            .await)
    }

    /// Returns a stream that yields the current client balance every time it
    /// changes.
    pub async fn subscribe_balance_changes(&self, unit: AmountUnit) -> BoxStream<'static, Amount> {
        let primary_module_things =
            if let Some((primary_module_id, primary_module)) = self.primary_module_for_unit(unit) {
                let balance_changes = primary_module.subscribe_balance_changes().await;
                let initial_balance = self
                    .get_balance_for_unit(unit)
                    .await
                    .expect("Primary is present");

                Some((
                    primary_module_id,
                    primary_module.clone(),
                    balance_changes,
                    initial_balance,
                ))
            } else {
                None
            };
        let db = self.db().clone();

        Box::pin(async_stream::stream! {
            let Some((primary_module_id, primary_module, mut balance_changes, initial_balance)) = primary_module_things else {
                // If there is no primary module, there will not be one until client is
                // restarted
                pending().await
            };


            yield initial_balance;
            let mut prev_balance = initial_balance;
            while let Some(()) = balance_changes.next().await {
                let mut dbtx = db.begin_transaction_nc().await;
                let balance = primary_module
                     .get_balance(primary_module_id, &mut dbtx, unit)
                    .await;

                // Deduplicate in case modules cannot always tell if the balance actually changed
                if balance != prev_balance {
                    prev_balance = balance;
                    yield balance;
                }
            }
        })
    }

    /// Make a single API version request to a peer after a delay.
    ///
    /// The delay is here to unify the type of a future both for initial request
    /// and possible retries.
    async fn make_api_version_request(
        delay: Duration,
        peer_id: PeerId,
        api: &DynGlobalApi,
    ) -> (
        PeerId,
        Result<SupportedApiVersionsSummary, fedimint_connectors::error::ServerError>,
    ) {
        runtime::sleep(delay).await;
        (
            peer_id,
            api.request_single_peer::<SupportedApiVersionsSummary>(
                VERSION_ENDPOINT.to_owned(),
                ApiRequestErased::default(),
                peer_id,
            )
            .await,
        )
    }

    /// Create a backoff strategy for API version requests.
    ///
    /// Keep trying, initially somewhat aggressively, but after a while retry
    /// very slowly, because chances for response are getting lower and
    /// lower.
    fn create_api_version_backoff() -> impl Iterator<Item = Duration> {
        custom_backoff(Duration::from_millis(200), Duration::from_secs(600), None)
    }

    /// Query the federation for API version support and then calculate
    /// the best API version to use (supported by most guardians).
    pub async fn fetch_common_api_versions_from_all_peers(
        num_peers: NumPeers,
        api: DynGlobalApi,
        db: Database,
        num_responses_sender: watch::Sender<usize>,
    ) {
        let mut backoff = Self::create_api_version_backoff();

        // NOTE: `FuturesUnordered` is a footgun, but since we only poll it for result
        // and make a single async db write operation, it should be OK.
        let mut requests = FuturesUnordered::new();

        for peer_id in num_peers.peer_ids() {
            requests.push(Self::make_api_version_request(
                Duration::ZERO,
                peer_id,
                &api,
            ));
        }

        let mut num_responses = 0;

        while let Some((peer_id, response)) = requests.next().await {
            let retry = match response {
                Err(err) => {
                    let has_previous_response = db
                        .begin_transaction_nc()
                        .await
                        .get_value(&PeerLastApiVersionsSummaryKey(peer_id))
                        .await
                        .is_some();
                    debug!(
                        target: LOG_CLIENT,
                        %peer_id,
                        err = %err.fmt_compact(),
                        %has_previous_response,
                        "Failed to refresh API versions of a peer"
                    );

                    !has_previous_response
                }
                Ok(o) => {
                    // Save the response to the database right away, just to
                    // not lose it
                    let mut dbtx = db.begin_transaction().await;
                    dbtx.insert_entry(
                        &PeerLastApiVersionsSummaryKey(peer_id),
                        &PeerLastApiVersionsSummary(o),
                    )
                    .await;
                    dbtx.commit_tx().await;
                    false
                }
            };

            if retry {
                requests.push(Self::make_api_version_request(
                    backoff.next().expect("Keeps retrying"),
                    peer_id,
                    &api,
                ));
            } else {
                num_responses += 1;
                num_responses_sender.send_replace(num_responses);
            }
        }
    }

    /// Fetch API versions from peers, retrying until we get threshold number of
    /// successful responses. Returns the successful responses collected
    /// from at least `num_peers.threshold()` peers.
    pub async fn fetch_peers_api_versions_from_threshold_of_peers(
        num_peers: NumPeers,
        api: DynGlobalApi,
    ) -> BTreeMap<PeerId, SupportedApiVersionsSummary> {
        let mut backoff = Self::create_api_version_backoff();

        // NOTE: `FuturesUnordered` is a footgun, but since we only poll it for result
        // and collect responses, it should be OK.
        let mut requests = FuturesUnordered::new();

        for peer_id in num_peers.peer_ids() {
            requests.push(Self::make_api_version_request(
                Duration::ZERO,
                peer_id,
                &api,
            ));
        }

        let mut successful_responses = BTreeMap::new();

        while successful_responses.len() < num_peers.threshold()
            && let Some((peer_id, response)) = requests.next().await
        {
            let retry = match response {
                Err(err) => {
                    debug!(
                        target: LOG_CLIENT,
                        %peer_id,
                        err = %err.fmt_compact(),
                        "Failed to fetch API versions from peer"
                    );
                    true
                }
                Ok(response) => {
                    successful_responses.insert(peer_id, response);
                    false
                }
            };

            if retry {
                requests.push(Self::make_api_version_request(
                    backoff.next().expect("Keeps retrying"),
                    peer_id,
                    &api,
                ));
            }
        }

        successful_responses
    }

    /// Fetch API versions from peers and discover common API versions to use.
    pub async fn fetch_common_api_versions(
        config: &ClientConfig,
        api: &DynGlobalApi,
    ) -> anyhow::Result<BTreeMap<PeerId, SupportedApiVersionsSummary>> {
        debug!(
            target: LOG_CLIENT,
            "Fetching common api versions"
        );

        let num_peers = NumPeers::from(config.global.api_endpoints.len());

        let peer_api_version_sets =
            Self::fetch_peers_api_versions_from_threshold_of_peers(num_peers, api.clone()).await;

        Ok(peer_api_version_sets)
    }

    /// Write API version set to database cache.
    /// Used when we have a pre-calculated API version set that should be stored
    /// for later use.
    pub async fn write_api_version_cache(
        dbtx: &mut DatabaseTransaction<'_>,
        api_version_set: ApiVersionSet,
    ) {
        debug!(
            target: LOG_CLIENT,
            value = ?api_version_set,
            "Writing API version set to cache"
        );

        dbtx.insert_entry(
            &CachedApiVersionSetKey,
            &CachedApiVersionSet(api_version_set),
        )
        .await;
    }

    /// Store prefetched peer API version responses and calculate/store common
    /// API version set. This processes the individual peer responses by
    /// storing them in the database and calculating the common API version
    /// set for caching.
    pub async fn store_prefetched_api_versions(
        db: &Database,
        config: &ClientConfig,
        client_module_init: &ClientModuleInitRegistry,
        peer_api_versions: &BTreeMap<PeerId, SupportedApiVersionsSummary>,
    ) {
        debug!(
            target: LOG_CLIENT,
            "Storing {} prefetched peer API version responses and calculating common version set",
            peer_api_versions.len()
        );

        let mut dbtx = db.begin_transaction().await;
        // Calculate common API version set from individual responses
        let client_supported_versions =
            Self::supported_api_versions_summary_static(config, client_module_init);
        match fedimint_client_module::api_version_discovery::discover_common_api_versions_set(
            &client_supported_versions,
            peer_api_versions,
        ) {
            Ok(common_api_versions) => {
                // Write the calculated common API version set to database cache
                Self::write_api_version_cache(&mut dbtx.to_ref_nc(), common_api_versions).await;
                debug!(target: LOG_CLIENT, "Calculated and stored common API version set");
            }
            Err(err) => {
                debug!(target: LOG_CLIENT, err = %err.fmt_compact_anyhow(), "Failed to calculate common API versions from prefetched data");
            }
        }

        // Store individual peer responses to database
        for (peer_id, peer_api_versions) in peer_api_versions {
            dbtx.insert_entry(
                &PeerLastApiVersionsSummaryKey(*peer_id),
                &PeerLastApiVersionsSummary(peer_api_versions.clone()),
            )
            .await;
        }
        dbtx.commit_tx().await;
        debug!(target: LOG_CLIENT, "Stored individual peer API version responses");
    }

    /// [`SupportedApiVersionsSummary`] that the client and its modules support
    pub fn supported_api_versions_summary_static(
        config: &ClientConfig,
        client_module_init: &ClientModuleInitRegistry,
    ) -> SupportedApiVersionsSummary {
        SupportedApiVersionsSummary {
            core: SupportedCoreApiVersions {
                core_consensus: config.global.consensus_version,
                api: MultiApiVersion::try_from_iter(SUPPORTED_CORE_API_VERSIONS.to_owned())
                    .expect("must not have conflicting versions"),
            },
            modules: config
                .modules
                .iter()
                .filter_map(|(&module_instance_id, module_config)| {
                    client_module_init
                        .get(module_config.kind())
                        .map(|module_init| {
                            (
                                module_instance_id,
                                SupportedModuleApiVersions {
                                    core_consensus: config.global.consensus_version,
                                    module_consensus: module_config.version,
                                    api: module_init.supported_api_versions(),
                                },
                            )
                        })
                })
                .collect(),
        }
    }

    pub async fn load_and_refresh_common_api_version(&self) -> anyhow::Result<ApiVersionSet> {
        Self::load_and_refresh_common_api_version_static(
            &self.config().await,
            &self.module_inits,
            self.connectors.clone(),
            &self.api,
            &self.db,
            &self.task_group,
            &self.client_span,
        )
        .await
    }

    /// Force refresh API versions from the federation, bypassing the cache.
    ///
    /// This queries all peers for their supported API versions and calculates
    /// the common API version set to use. The result is stored in the database
    /// cache for future use.
    pub async fn refresh_api_versions(&self) -> anyhow::Result<ApiVersionSet> {
        Self::refresh_common_api_version_static(
            &self.config().await,
            &self.module_inits,
            &self.api,
            &self.db,
            self.task_group.clone(),
            &self.client_span,
            true,
        )
        .await
    }

    /// Load the common api versions to use from cache and start a background
    /// process to refresh them.
    ///
    /// This is a compromise, so we not have to wait for version discovery to
    /// complete every time a [`Client`] is being built.
    pub(crate) async fn load_and_refresh_common_api_version_static(
        config: &ClientConfig,
        module_init: &ClientModuleInitRegistry,
        connectors: ConnectorRegistry,
        api: &DynGlobalApi,
        db: &Database,
        task_group: &TaskGroup,
        client_span: &Span,
    ) -> anyhow::Result<ApiVersionSet> {
        if let Some(v) = db
            .begin_transaction_nc()
            .await
            .get_value(&CachedApiVersionSetKey)
            .await
        {
            client_span.in_scope(|| {
                debug!(
                    target: LOG_CLIENT,
                    "Found existing cached common api versions"
                );
            });
            let config = config.clone();
            let client_module_init = module_init.clone();
            let api = api.clone();
            let db = db.clone();
            let task_group = task_group.clone();
            let client_span_owned = client_span.clone();
            // Separate task group, because we actually don't want to be waiting for this to
            // finish, and it's just best effort.
            task_group.clone().spawn_cancellable_with_span(
                client_span.clone(),
                "refresh_common_api_version_static",
                async move {
                    connectors.wait_for_initialized_connections().await;

                    if let Err(error) = Self::refresh_common_api_version_static(
                        &config,
                        &client_module_init,
                        &api,
                        &db,
                        task_group,
                        &client_span_owned,
                        false,
                    )
                    .await
                    {
                        warn!(
                            target: LOG_CLIENT,
                            err = %error.fmt_compact_anyhow(), "Failed to discover common api versions"
                        );
                    }
                },
            );

            return Ok(v.0);
        }

        info!(
            target: LOG_CLIENT,
            "Fetching initial API versions "
        );
        Self::refresh_common_api_version_static(
            config,
            module_init,
            api,
            db,
            task_group.clone(),
            client_span,
            true,
        )
        .await
    }

    async fn refresh_common_api_version_static(
        config: &ClientConfig,
        client_module_init: &ClientModuleInitRegistry,
        api: &DynGlobalApi,
        db: &Database,
        task_group: TaskGroup,
        client_span: &Span,
        block_until_ok: bool,
    ) -> anyhow::Result<ApiVersionSet> {
        debug!(
            target: LOG_CLIENT,
            "Refreshing common api versions"
        );

        let (num_responses_sender, mut num_responses_receiver) = tokio::sync::watch::channel(0);
        let num_peers = NumPeers::from(config.global.api_endpoints.len());

        task_group.spawn_cancellable_with_span(
            client_span.clone(),
            "refresh peers api versions",
            Client::fetch_common_api_versions_from_all_peers(
                num_peers,
                api.clone(),
                db.clone(),
                num_responses_sender,
            ),
        );

        let common_api_versions = loop {
            // Wait to collect enough answers before calculating a set of common api
            // versions to use. Note that all peers individual responses from
            // previous attempts are still being used, and requests, or even
            // retries for response of peers are not actually cancelled, as they
            // are happening on a separate task. This is all just to bound the
            // time user can be waiting for the join operation to finish, at the
            // risk of picking wrong version in very rare circumstances.
            let _: Result<_, Elapsed> = runtime::timeout(
                Duration::from_secs(30),
                num_responses_receiver.wait_for(|num| num_peers.threshold() <= *num),
            )
            .await;

            let peer_api_version_sets = Self::load_peers_last_api_versions(db, num_peers).await;

            match fedimint_client_module::api_version_discovery::discover_common_api_versions_set(
                &Self::supported_api_versions_summary_static(config, client_module_init),
                &peer_api_version_sets,
            ) {
                Ok(o) => break o,
                Err(err) if block_until_ok => {
                    warn!(
                        target: LOG_CLIENT,
                        err = %err.fmt_compact_anyhow(),
                        "Failed to discover API version to use. Retrying..."
                    );
                    continue;
                }
                Err(e) => return Err(e),
            }
        };

        debug!(
            target: LOG_CLIENT,
            value = ?common_api_versions,
            "Updating the cached common api versions"
        );
        let mut dbtx = db.begin_transaction().await;
        let _ = dbtx
            .insert_entry(
                &CachedApiVersionSetKey,
                &CachedApiVersionSet(common_api_versions.clone()),
            )
            .await;

        dbtx.commit_tx().await;

        Ok(common_api_versions)
    }

    /// Get the client [`Metadata`]
    pub async fn get_metadata(&self) -> Metadata {
        self.db
            .begin_transaction_nc()
            .await
            .get_value(&ClientMetadataKey)
            .await
            .unwrap_or_else(|| {
                warn!(
                    target: LOG_CLIENT,
                    "Missing existing metadata. This key should have been set on Client init"
                );
                Metadata::empty()
            })
    }

    /// Set the client [`Metadata`]
    pub async fn set_metadata(&self, metadata: &Metadata) {
        self.db
            .autocommit::<_, _, anyhow::Error>(
                |dbtx, _| {
                    Box::pin(async {
                        Self::set_metadata_dbtx(dbtx, metadata).await;
                        Ok(())
                    })
                },
                None,
            )
            .await
            .expect("Failed to autocommit metadata");
    }

    pub fn has_pending_recoveries(&self) -> bool {
        !self
            .client_recovery_status_receiver
            .borrow()
            .values()
            .all(RecoveryStatus::is_successfully_done)
    }

    /// Whether every module of this client can be used right now.
    ///
    /// A module that cannot be used while it recovers is left out of the module
    /// registry, and only becomes available once the client is reopened with
    /// its recovery complete. This reports whether any module is currently held
    /// back that way, so an application can tell whether it has to wait for
    /// [`Self::wait_for_all_recoveries`] and reopen before it can do anything,
    /// or can go ahead right away with a recovery still running.
    ///
    /// Always `true` for a client that is not recovering.
    pub fn all_modules_usable(&self) -> bool {
        self.client_recovery_status_receiver
            .borrow()
            .keys()
            .all(|module_instance_id| self.modules.get(*module_instance_id).is_some())
    }

    /// Wait for all module recoveries to finish
    ///
    /// Returns `Ok(())` once every module recovery completed, or an error as
    /// soon as any one of them fails terminally.
    ///
    /// An error does not mean the recovery task is done: the failed module's
    /// progress stays pending forever (so [`Self::has_pending_recoveries`]
    /// keeps returning `true`) and the recovery task stays parked. The
    /// failure is in-memory only and is not persisted, so reopening the
    /// client retries the recovery from its last persisted, non-terminal
    /// progress.
    ///
    /// A bit of a heavy approach.
    pub async fn wait_for_all_recoveries(&self) -> anyhow::Result<()> {
        Self::wait_for_recoveries(
            self.client_recovery_status_receiver.clone(),
            |_module_instance_id| true,
            "Recovery task completed and update receiver disconnected, but some modules failed to recover",
        )
        .await
    }

    /// Wait for the recovery of every module accepted by `module_filter` to
    /// either complete or fail.
    ///
    /// Since [`RecoveryProgress`] can never express a failure, waiting on
    /// progress alone would block forever on a module that gave up. A module
    /// that gave up is recorded as [`RecoveryStatus::Failed`] by
    /// [`Self::run_module_recoveries_task`] instead, so both outcomes are
    /// observable on the same per-module status.
    async fn wait_for_recoveries(
        mut status_receiver: watch::Receiver<BTreeMap<ModuleInstanceId, RecoveryStatus>>,
        module_filter: impl Fn(ModuleInstanceId) -> bool,
        disconnected_context: &'static str,
    ) -> anyhow::Result<()> {
        let failure = status_receiver
            .wait_for(|statuses| {
                let matching = || {
                    statuses
                        .iter()
                        .filter(|(module_instance_id, _status)| module_filter(**module_instance_id))
                        .map(|(_module_instance_id, status)| status)
                };

                // A failure is terminal, so waiting for the modules that are
                // still recovering would only delay an outcome that can no
                // longer change.
                matching().any(|status| matches!(status, RecoveryStatus::Failed { .. }))
                    || matching().all(RecoveryStatus::is_successfully_done)
            })
            .await
            .context(disconnected_context)?
            // Classified from the woken-up snapshot before anything else, so a
            // failure of one module still wins over another one completing.
            .iter()
            .find_map(|(module_instance_id, status)| match status {
                RecoveryStatus::Failed { error, .. } if module_filter(*module_instance_id) => {
                    Some((*module_instance_id, error.clone()))
                }
                _ => None,
            });

        match failure {
            Some((module_instance_id, error)) => Err(anyhow!(
                "Module recovery failed: module_instance_id={module_instance_id}, error={error}"
            )),
            None => Ok(()),
        }
    }

    /// Subscribe to recover progress for all the modules.
    ///
    /// This stream can contain duplicate progress for a module.
    /// Don't use this stream for detecting completion of recovery.
    ///
    /// It can't express a failure either: a module that failed to recover keeps
    /// being reported with the last progress it made, which is never done. Use
    /// [`Self::wait_for_all_recoveries`] or
    /// [`Self::wait_for_module_kind_recovery`] to learn the outcome.
    pub fn subscribe_to_recovery_progress(
        &self,
    ) -> impl Stream<Item = (ModuleInstanceId, RecoveryProgress)> + use<> {
        WatchStream::new(self.client_recovery_status_receiver.clone()).flat_map(|statuses| {
            futures::stream::iter(
                statuses
                    .into_iter()
                    .map(|(module_instance_id, status)| (module_instance_id, status.progress())),
            )
        })
    }

    /// Wait for the recoveries of all modules of `module_kind` to finish
    ///
    /// Returns `Ok(())` once every recovery of that kind completed, or an error
    /// as soon as one of them fails terminally. Failures of other module kinds
    /// are ignored.
    ///
    /// See [`Self::wait_for_all_recoveries`] for what an error does and does
    /// not say about the state of the recovery.
    pub async fn wait_for_module_kind_recovery(
        &self,
        module_kind: ModuleKind,
    ) -> anyhow::Result<()> {
        let config = self.config().await;
        Self::wait_for_recoveries(
            self.client_recovery_status_receiver.clone(),
            move |module_instance_id| {
                config
                    .modules
                    .get(&module_instance_id)
                    .is_some_and(|module| module.kind == module_kind)
            },
            "Recovery task completed and update receiver disconnected, but the desired modules are still unavailable or failed to recover",
        )
        .await
    }

    pub async fn wait_for_all_active_state_machines(&self) -> anyhow::Result<()> {
        loop {
            if self.executor.get_active_states().await.is_empty() {
                break;
            }
            sleep(Duration::from_millis(100)).await;
        }
        Ok(())
    }

    /// Set the client [`Metadata`]
    pub async fn set_metadata_dbtx(dbtx: &mut DatabaseTransaction<'_>, metadata: &Metadata) {
        dbtx.insert_new_entry(&ClientMetadataKey, metadata).await;
    }

    fn spawn_module_recoveries_task(
        &self,
        recovery_sender: watch::Sender<BTreeMap<ModuleInstanceId, RecoveryStatus>>,
        module_recoveries: BTreeMap<ModuleInstanceId, ModuleRecoveryFuture>,
        module_recovery_progress_receivers: BTreeMap<
            ModuleInstanceId,
            watch::Receiver<RecoveryProgress>,
        >,
        // Sourced from the config rather than `self.modules`, since modules
        // currently being recovered are not yet initialized in the registry.
        module_kinds: BTreeMap<ModuleInstanceId, ModuleKind>,
    ) {
        let db = self.db.clone();
        let log_ordering_wakeup_tx = self.log_ordering_wakeup_tx.clone();
        // A module can finish its own final database commit before the client
        // persists the corresponding completed progress below. Keep the
        // coordinator alive across graceful shutdown so that gap cannot cause
        // a non-idempotent module finalization to be replayed on reopen.
        self.spawn("module recoveries", |_task_handle| async {
            Self::run_module_recoveries_task(
                db,
                log_ordering_wakeup_tx,
                recovery_sender,
                module_recoveries,
                module_recovery_progress_receivers,
                module_kinds,
            )
            .await;
        });
    }

    async fn run_module_recoveries_task(
        db: Database,
        log_ordering_wakeup_tx: watch::Sender<()>,
        recovery_sender: watch::Sender<BTreeMap<ModuleInstanceId, RecoveryStatus>>,
        module_recoveries: BTreeMap<ModuleInstanceId, ModuleRecoveryFuture>,
        module_recovery_progress_receivers: BTreeMap<
            ModuleInstanceId,
            watch::Receiver<RecoveryProgress>,
        >,
        module_kinds: BTreeMap<ModuleInstanceId, ModuleKind>,
    ) {
        debug!(target: LOG_CLIENT_RECOVERY, num_modules=%module_recovery_progress_receivers.len(), "Staring module recoveries");

        // A recovery update for a single module: either an intermediate progress
        // report, or the final completion carrying the recovered amount (if the
        // module tracks it).
        enum RecoveryUpdate {
            Progress(RecoveryProgress),
            Completed(Option<Amount>),
        }

        let mut completed_stream = Vec::new();
        let progress_stream = futures::stream::FuturesUnordered::new();

        for (module_instance_id, f) in module_recoveries {
            let recovery_sender = recovery_sender.clone();
            completed_stream.push(futures::stream::once(Box::pin(async move {
                match f.await {
                    Ok(amount) => (module_instance_id, RecoveryUpdate::Completed(amount)),
                    Err(err) => {
                        let error = err.fmt_compact_anyhow().to_string();
                        warn!(
                            target: LOG_CLIENT,
                            err = %error.as_str(), module_instance_id, "Module recovery failed"
                        );
                        // since the progress a module reports can't express a
                        // failure, record it as the terminal state of this
                        // module's recovery for anyone waiting on the outcome.
                        // Keyed by module instance so a later failure of
                        // another module can't overwrite this one before a
                        // waiter observes it.
                        recovery_sender.send_modify(|statuses| {
                            let last_progress = statuses
                                .get(&module_instance_id)
                                .expect("existing status must be present")
                                .progress();
                            statuses.insert(
                                module_instance_id,
                                RecoveryStatus::Failed {
                                    last_progress,
                                    error,
                                },
                            );
                        });
                        // The terminal failure is now published to the waiters,
                        // and this branch stays pending on purpose so the
                        // recovery coordinator and its single status sender
                        // remain alive. The failed module's progress never
                        // becomes done either, so observers that only see the
                        // progress keep treating the recovery as pending, which
                        // it effectively is. The failure is deliberately not
                        // persisted, so reopening the client retries the
                        // recovery.
                        futures::future::pending::<()>().await;
                        unreachable!()
                    }
                }
            })));
        }

        for (module_instance_id, rx) in module_recovery_progress_receivers {
            progress_stream.push(
                tokio_stream::wrappers::WatchStream::new(rx)
                    .fuse()
                    .map(move |progress| (module_instance_id, RecoveryUpdate::Progress(progress))),
            );
        }

        let mut futures = futures::stream::select(
            futures::stream::select_all(progress_stream),
            futures::stream::select_all(completed_stream),
        );

        while let Some((module_instance_id, update)) = futures.next().await {
            // Cloned so the `watch::Ref` is released right here: holding it
            // across the awaits or the `send_modify` below would deadlock the
            // channel it borrows from.
            let prev_status = recovery_sender
                .borrow()
                .get(&module_instance_id)
                .expect("existing status must be present")
                .clone();

            // A failure is the terminal state of a module's recovery, but
            // progress and completion are merged without any ordering between
            // them, so a stale progress update of that module can still arrive
            // after it. Applying it would overwrite the failure with an
            // in-progress status, and a waiter subscribing afterwards would
            // block forever again. Ignoring it entirely, without touching the
            // channel, also keeps subscribers from seeing a snapshot that says
            // nothing new. Mirrors the sticky "once done, stick with it"
            // handling below.
            if matches!(prev_status, RecoveryStatus::Failed { .. }) {
                debug!(
                    target: LOG_CLIENT_RECOVERY,
                    module_instance_id,
                    "Ignoring a recovery update of a module whose recovery already failed"
                );
                continue;
            }

            let prev_progress = prev_status.progress();

            // A module reports its progress on a channel it owns, so the values
            // arriving here are untrusted: `update_recovery_progress` filters
            // them, but a module can send on `progress_tx` directly. Reject the
            // values that would break the invariants downstream depends on
            // before anything is persisted or broadcast.
            if let RecoveryUpdate::Progress(progress) = &update {
                if progress.is_done() {
                    warn!(
                        target: LOG_CLIENT_RECOVERY,
                        module_instance_id,
                        "Module bypassed the sanctioned recovery progress reporting API and reported a completed recovery progress. Ignoring"
                    );
                    continue;
                }

                // The module's channel starts at the progress the client seeded
                // it with, so a "none" that doesn't regress anything is the
                // normal start of a recovery, not a module misbehaving. Neither
                // is one seen after the module already completed: progress and
                // completion are merged without ordering between them, so a
                // seeded value can arrive after the completion it preceded.
                if progress.is_none() && !prev_progress.is_none() && !prev_progress.is_done() {
                    warn!(
                        target: LOG_CLIENT_RECOVERY,
                        module_instance_id,
                        "Module bypassed the sanctioned recovery progress reporting API and reported a none recovery progress, regressing its previous one. Ignoring"
                    );
                    continue;
                }
            }

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

            // The recovered amount is only known once the module's recovery
            // future resolves, which is also the only way progress transitions
            // to "done": the guard above rejects a completed progress reported
            // by a module, so `RecoveryUpdate::Completed` stays the only
            // producer of a done progress, and done therefore implies success.
            let (progress, recovered_amount) = if prev_progress.is_done() {
                // since updates might be out of order, once done, stick with it
                (prev_progress, None)
            } else {
                match update {
                    RecoveryUpdate::Progress(progress) => (progress, None),
                    RecoveryUpdate::Completed(amount) => (prev_progress.to_complete(), amount),
                }
            };

            if !prev_progress.is_done() && progress.is_done() {
                info!(
                    target: LOG_CLIENT,
                    module_instance_id,
                    progress = format!("{}/{}", progress.complete, progress.total),
                    amount = ?recovered_amount,
                    "Recovery complete"
                );
                dbtx.log_event(
                    log_ordering_wakeup_tx.clone(),
                    None,
                    ModuleRecoveryCompleted {
                        module_id: module_instance_id,
                        kind: module_kinds.get(&module_instance_id).cloned(),
                        amount: recovered_amount,
                    },
                )
                .await;
            } else {
                info!(
                    target: LOG_CLIENT,
                    module_instance_id,
                    kind = ?module_kinds.get(&module_instance_id),
                    progress = format!("{}/{}", progress.complete, progress.total),
                    "Recovery progress"
                );
            }

            dbtx.insert_entry(
                &ClientModuleRecovery { module_instance_id },
                &ClientModuleRecoveryState { progress },
            )
            .await;
            dbtx.commit_tx().await;

            recovery_sender.send_modify(|statuses| {
                statuses.insert(module_instance_id, RecoveryStatus::InProgress(progress));
            });
        }
        debug!(target: LOG_CLIENT_RECOVERY, "Recovery executor stopped");
    }

    async fn load_peers_last_api_versions(
        db: &Database,
        num_peers: NumPeers,
    ) -> BTreeMap<PeerId, SupportedApiVersionsSummary> {
        let mut peer_api_version_sets = BTreeMap::new();

        let mut dbtx = db.begin_transaction_nc().await;
        for peer_id in num_peers.peer_ids() {
            if let Some(v) = dbtx
                .get_value(&PeerLastApiVersionsSummaryKey(peer_id))
                .await
            {
                peer_api_version_sets.insert(peer_id, v.0);
            }
        }
        drop(dbtx);
        peer_api_version_sets
    }

    /// You likely want to use [`Client::get_peer_urls`]. This function returns
    /// only the announcements and doesn't use the config as fallback.
    pub async fn get_peer_url_announcements(&self) -> BTreeMap<PeerId, SignedApiAnnouncement> {
        self.db()
            .begin_transaction_nc()
            .await
            .find_by_prefix(&ApiAnnouncementPrefix)
            .await
            .map(|(announcement_key, announcement)| (announcement_key.0, announcement))
            .collect()
            .await
    }

    /// Returns guardian metadata stored in the client database
    pub async fn get_guardian_metadata(
        &self,
    ) -> BTreeMap<PeerId, fedimint_core::net::guardian_metadata::SignedGuardianMetadata> {
        self.db()
            .begin_transaction_nc()
            .await
            .find_by_prefix(&crate::guardian_metadata::GuardianMetadataPrefix)
            .await
            .map(|(key, metadata)| (key.0, metadata))
            .collect()
            .await
    }

    /// Returns a list of guardian API URLs
    pub async fn get_peer_urls(&self) -> BTreeMap<PeerId, SafeUrl> {
        get_api_urls(&self.db, &self.config().await, self.iroh_enable_next).await
    }

    /// Create an invite code with the api endpoint of the given peer which can
    /// be used to download this client config
    pub async fn invite_code(&self, peer: PeerId) -> Option<InviteCode> {
        self.get_peer_urls()
            .await
            .into_iter()
            .find_map(|(peer_id, url)| (peer == peer_id).then_some(url))
            .map(|peer_url| {
                InviteCode::new(
                    peer_url.clone(),
                    peer,
                    self.federation_id(),
                    self.api_secret.clone(),
                )
            })
    }

    /// Blocks till the client has synced the guardian public key set
    /// (introduced in version 0.4) and returns it. Once it has been fetched
    /// once this function is guaranteed to return immediately.
    pub async fn get_guardian_public_keys_blocking(
        &self,
    ) -> BTreeMap<PeerId, fedimint_core::secp256k1::PublicKey> {
        self.db
            .autocommit(
                |dbtx, _| {
                    Box::pin(async move {
                        let config = self.config().await;

                        let guardian_pub_keys = self
                            .get_or_backfill_broadcast_public_keys(dbtx, config)
                            .await;

                        Result::<_, ()>::Ok(guardian_pub_keys)
                    })
                },
                None,
            )
            .await
            .expect("Will retry forever")
    }

    async fn get_or_backfill_broadcast_public_keys(
        &self,
        dbtx: &mut DatabaseTransaction<'_>,
        config: ClientConfig,
    ) -> BTreeMap<PeerId, PublicKey> {
        match config.global.broadcast_public_keys {
            Some(guardian_pub_keys) => guardian_pub_keys,
            _ => {
                let (guardian_pub_keys, new_config) = self.fetch_and_update_config(config).await;

                dbtx.insert_entry(&ClientConfigKey, &new_config).await;
                *(self.config.write().await) = new_config;
                guardian_pub_keys
            }
        }
    }

    pub async fn fetch_session_count(&self) -> FederationResult<u64> {
        self.api.session_count().await
    }

    async fn fetch_and_update_config(
        &self,
        config: ClientConfig,
    ) -> (BTreeMap<PeerId, PublicKey>, ClientConfig) {
        let fetched_config = retry(
            "Fetching guardian public keys",
            backoff_util::background_backoff(),
            || async {
                anyhow::Ok(
                    self.api
                        .request_current_consensus::<ClientConfig>(
                            CLIENT_CONFIG_ENDPOINT.to_owned(),
                            ApiRequestErased::default(),
                        )
                        .await?,
                )
            },
        )
        .await
        .expect("Will never return on error");

        let Some(guardian_pub_keys) = fetched_config.global.broadcast_public_keys else {
            warn!(
                target: LOG_CLIENT,
                "Guardian public keys not found in fetched config, server not updated to 0.4 yet"
            );
            pending::<()>().await;
            unreachable!("Pending will never return");
        };

        let new_config = ClientConfig {
            global: GlobalClientConfig {
                broadcast_public_keys: Some(guardian_pub_keys.clone()),
                ..config.global
            },
            modules: config.modules,
        };
        (guardian_pub_keys, new_config)
    }

    pub fn handle_global_rpc(
        &self,
        method: String,
        params: serde_json::Value,
    ) -> BoxStream<'_, anyhow::Result<serde_json::Value>> {
        Box::pin(try_stream! {
            match method.as_str() {
                "get_balance" => {
                    let balance = self.get_balance_for_btc().await.unwrap_or_default();
                    yield serde_json::to_value(balance)?;
                }
                "subscribe_balance_changes" => {
                    let req: GetBalanceChangesRequest= serde_json::from_value(params)?;
                    let mut stream = self.subscribe_balance_changes(req.unit).await;
                    while let Some(balance) = stream.next().await {
                        yield serde_json::to_value(balance)?;
                    }
                }
                "get_config" => {
                    let config = self.config().await;
                    yield serde_json::to_value(config)?;
                }
                "get_federation_id" => {
                    let federation_id = self.federation_id();
                    yield serde_json::to_value(federation_id)?;
                }
                "get_invite_code" => {
                    let req: GetInviteCodeRequest = serde_json::from_value(params)?;
                    let invite_code = self.invite_code(req.peer).await;
                    yield serde_json::to_value(invite_code)?;
                }
                "get_operation" => {
                    let req: GetOperationIdRequest = serde_json::from_value(params)?;
                    let operation = self.operation_log().get_operation(req.operation_id).await;
                    yield serde_json::to_value(operation)?;
                }
                "list_operations" => {
                    let req: ListOperationsParams = serde_json::from_value(params)?;
                    let limit = if req.limit.is_none() && req.last_seen.is_none() {
                        usize::MAX
                    } else {
                        req.limit.unwrap_or(usize::MAX)
                    };
                    let operations = self.operation_log()
                        .paginate_operations_rev(limit, req.last_seen)
                        .await;
                    yield serde_json::to_value(operations)?;
                }
                "get_event_log" => {
                    let req: GetEventLogRequest = serde_json::from_value(params)?;
                    let limit = req
                        .limit
                        .unwrap_or(DEFAULT_EVENT_LOG_PAGE_SIZE)
                        .min(MAX_EVENT_LOG_PAGE_SIZE);
                    let events = self.get_event_log(req.pos, limit).await;
                    yield serde_json::to_value(events)?;
                }
                "session_count" => {
                    let count = self.fetch_session_count().await?;
                    yield serde_json::to_value(count)?;
                }
                "has_pending_recoveries" => {
                    let has_pending = self.has_pending_recoveries();
                    yield serde_json::to_value(has_pending)?;
                }
                "wait_for_all_recoveries" => {
                    self.wait_for_all_recoveries().await?;
                    yield serde_json::Value::Null;
                }
                "subscribe_to_recovery_progress" => {
                    let mut stream = self.subscribe_to_recovery_progress();
                    while let Some((module_id, progress)) = stream.next().await {
                        yield serde_json::json!({
                            "module_id": module_id,
                            "progress": progress
                        });
                    }
                }
                #[allow(deprecated)]
                "backup_to_federation" => {
                    let metadata = if params.is_null() {
                        Metadata::from_json_serialized(serde_json::json!({}))
                    } else {
                        Metadata::from_json_serialized(params)
                    };
                    self.backup_to_federation(metadata).await?;
                    yield serde_json::Value::Null;
                }
                _ => {
                    Err(anyhow::format_err!("Unknown method: {}", method))?;
                    unreachable!()
                },
            }
        })
    }

    pub async fn log_event<E>(&self, module_id: Option<ModuleInstanceId>, event: E)
    where
        E: Event + Send,
    {
        let mut dbtx = self.db.begin_transaction().await;
        self.log_event_dbtx(&mut dbtx, module_id, event).await;
        dbtx.commit_tx().await;
    }

    pub async fn log_event_dbtx<E, Cap>(
        &self,
        dbtx: &mut DatabaseTransaction<'_, Cap>,
        module_id: Option<ModuleInstanceId>,
        event: E,
    ) where
        E: Event + Send,
        Cap: Send,
    {
        dbtx.log_event(self.log_ordering_wakeup_tx.clone(), module_id, event)
            .await;
    }

    pub async fn log_event_raw_dbtx<Cap>(
        &self,
        dbtx: &mut DatabaseTransaction<'_, Cap>,
        kind: EventKind,
        module: Option<(ModuleKind, ModuleInstanceId)>,
        payload: Vec<u8>,
        persist: EventPersistence,
    ) where
        Cap: Send,
    {
        let module_id = module.as_ref().map(|m| m.1);
        let module_kind = module.map(|m| m.0);
        dbtx.log_event_raw(
            self.log_ordering_wakeup_tx.clone(),
            kind,
            module_kind,
            module_id,
            payload,
            persist,
        )
        .await;
    }

    /// Built in event log (trimmable) tracker
    ///
    /// For the convenience of downstream applications, [`Client`] can store
    /// internally event log position for the main application using/driving it.
    ///
    /// Note that this position is a singleton, so this tracker should not be
    /// used for multiple purposes or applications, etc. at the same time.
    ///
    /// If the application has a need to follow log using multiple trackers, it
    /// should implement own [`DynEventLogTrimableTracker`] and store its
    /// persient data by itself.
    pub fn built_in_application_event_log_tracker(&self) -> DynEventLogTrimableTracker {
        struct BuiltInApplicationEventLogTracker;

        #[apply(async_trait_maybe_send!)]
        impl EventLogTrimableTracker for BuiltInApplicationEventLogTracker {
            // Store position in the event log
            async fn store(
                &mut self,
                dbtx: &mut DatabaseTransaction<NonCommittable>,
                pos: EventLogTrimableId,
            ) -> anyhow::Result<()> {
                dbtx.insert_entry(&DefaultApplicationEventLogKey, &pos)
                    .await;
                Ok(())
            }

            /// Load the last previous stored position (or None if never stored)
            async fn load(
                &mut self,
                dbtx: &mut DatabaseTransaction<NonCommittable>,
            ) -> anyhow::Result<Option<EventLogTrimableId>> {
                Ok(dbtx.get_value(&DefaultApplicationEventLogKey).await)
            }
        }
        Box::new(BuiltInApplicationEventLogTracker)
    }

    /// Like [`Self::handle_events`] but for historical data.
    ///
    ///
    /// This function can be used to process subset of events
    /// that is infrequent and important enough to be persisted
    /// forever. Most applications should prefer to use [`Self::handle_events`]
    /// which emits *all* events.
    pub async fn handle_historical_events<F, R>(
        &self,
        tracker: fedimint_eventlog::DynEventLogTracker,
        handler_fn: F,
    ) -> anyhow::Result<()>
    where
        F: Fn(&mut DatabaseTransaction<NonCommittable>, EventLogEntry) -> R,
        R: Future<Output = anyhow::Result<()>>,
    {
        fedimint_eventlog::handle_events(
            self.db.clone(),
            tracker,
            self.log_event_added_rx.clone(),
            handler_fn,
        )
        .await
    }

    /// Handle events emitted by the client
    ///
    /// This is a preferred method for reactive & asynchronous
    /// processing of events emitted by the client.
    ///
    /// It needs a `tracker` that will persist the position in the log
    /// as it is being handled. You can use the
    /// [`Client::built_in_application_event_log_tracker`] if this call is
    /// used for the single main application handling this instance of the
    /// [`Client`]. Otherwise you should implement your own tracker.
    ///
    /// This handler will call `handle_fn` with ever event emitted by
    /// [`Client`], including transient ones. The caller should atomically
    /// handle each event it is interested in and ignore other ones.
    ///
    /// This method returns only when client is shutting down or on internal
    /// error, so typically should be called in a background task dedicated
    /// to handling events.
    pub async fn handle_events<F, R>(
        &self,
        tracker: fedimint_eventlog::DynEventLogTrimableTracker,
        handler_fn: F,
    ) -> anyhow::Result<()>
    where
        F: Fn(&mut DatabaseTransaction<NonCommittable>, EventLogEntry) -> R,
        R: Future<Output = anyhow::Result<()>>,
    {
        fedimint_eventlog::handle_trimable_events(
            self.db.clone(),
            tracker,
            self.log_event_added_rx.clone(),
            handler_fn,
        )
        .await
    }

    pub async fn get_event_log(
        &self,
        pos: Option<EventLogId>,
        limit: u64,
    ) -> Vec<PersistedLogEntry> {
        self.get_event_log_dbtx(&mut self.db.begin_transaction_nc().await, pos, limit)
            .await
    }

    /// Returns the id that the next entry appended to the event log will be
    /// assigned, i.e. the position just past the current end of the log.
    pub async fn get_next_event_log_id(&self) -> EventLogId {
        self.db
            .begin_transaction_nc()
            .await
            .get_next_event_log_id()
            .await
    }

    pub async fn get_event_log_trimable(
        &self,
        pos: Option<EventLogTrimableId>,
        limit: u64,
    ) -> Vec<PersistedLogEntry> {
        self.get_event_log_trimable_dbtx(&mut self.db.begin_transaction_nc().await, pos, limit)
            .await
    }

    pub async fn get_event_log_dbtx<Cap>(
        &self,
        dbtx: &mut DatabaseTransaction<'_, Cap>,
        pos: Option<EventLogId>,
        limit: u64,
    ) -> Vec<PersistedLogEntry>
    where
        Cap: Send,
    {
        dbtx.get_event_log(pos, limit).await
    }

    pub async fn get_event_log_trimable_dbtx<Cap>(
        &self,
        dbtx: &mut DatabaseTransaction<'_, Cap>,
        pos: Option<EventLogTrimableId>,
        limit: u64,
    ) -> Vec<PersistedLogEntry>
    where
        Cap: Send,
    {
        dbtx.get_event_log_trimable(pos, limit).await
    }

    /// Register to receiver all new transient (unpersisted) events
    pub fn get_event_log_transient_receiver(&self) -> broadcast::Receiver<EventLogEntry> {
        self.log_event_added_transient_tx.subscribe()
    }

    /// Get a receiver that signals when new events are added to the event log
    pub fn log_event_added_rx(&self) -> watch::Receiver<()> {
        self.log_event_added_rx.clone()
    }

    pub fn iroh_enable_dht(&self) -> bool {
        self.iroh_enable_dht
    }

    /// Whether compatible iroh-next endpoints from guardian metadata are
    /// preferred.
    pub fn iroh_enable_next(&self) -> bool {
        self.iroh_enable_next
    }

    pub(crate) async fn run_core_migrations(
        db_no_decoders: &Database,
    ) -> Result<(), anyhow::Error> {
        let mut dbtx = db_no_decoders.begin_transaction().await;
        apply_migrations_core_client_dbtx(&mut dbtx.to_ref_nc(), "fedimint-client".to_string())
            .await?;
        if is_running_in_test_env() {
            verify_client_db_integrity_dbtx(&mut dbtx.to_ref_nc()).await;
        }
        dbtx.commit_tx_result().await?;
        Ok(())
    }

    /// Iterator over primary modules for a given `unit`
    fn primary_modules_for_unit(
        &self,
        unit: AmountUnit,
    ) -> impl Iterator<Item = (ModuleInstanceId, &DynClientModule)> {
        self.primary_modules
            .iter()
            .flat_map(move |(_prio, candidates)| {
                candidates
                    .specific
                    .get(&unit)
                    .into_iter()
                    .flatten()
                    .copied()
                    // within same priority, wildcard matches come last
                    .chain(candidates.wildcard.iter().copied())
            })
            .map(|id| (id, self.modules.get_expect(id)))
    }

    /// Primary module to use for `unit`
    ///
    /// Currently, just pick the first (highest priority) match
    pub fn primary_module_for_unit(
        &self,
        unit: AmountUnit,
    ) -> Option<(ModuleInstanceId, &DynClientModule)> {
        self.primary_modules_for_unit(unit).next()
    }

    /// [`Self::primary_module_for_unit`] for Bitcoin
    pub fn primary_module_for_btc(&self) -> (ModuleInstanceId, &DynClientModule) {
        self.primary_module_for_unit(AmountUnit::BITCOIN)
            .expect("No primary module for Bitcoin")
    }
}

#[apply(async_trait_maybe_send!)]
impl ClientContextIface for Client {
    fn get_module(&self, instance: ModuleInstanceId) -> &maybe_add_send_sync!(dyn IClientModule) {
        Client::get_module(self, instance)
    }

    fn api_clone(&self) -> DynGlobalApi {
        Client::api_clone(self)
    }
    fn decoders(&self) -> &ModuleDecoderRegistry {
        Client::decoders(self)
    }

    async fn finalize_and_submit_transaction(
        &self,
        operation_id: OperationId,
        operation_type: &str,
        operation_meta_gen: Box<maybe_add_send_sync!(dyn Fn(OutPointRange) -> serde_json::Value)>,
        tx_builder: TransactionBuilder,
    ) -> anyhow::Result<OutPointRange> {
        Client::finalize_and_submit_transaction(
            self,
            operation_id,
            operation_type,
            // |out_point_range| operation_meta_gen(out_point_range),
            &operation_meta_gen,
            tx_builder,
        )
        .await
    }

    async fn finalize_and_submit_transaction_dbtx(
        &self,
        dbtx: &mut DatabaseTransaction<'_>,
        operation_id: OperationId,
        operation_type: &str,
        operation_meta_gen: Box<maybe_add_send_sync!(dyn Fn(OutPointRange) -> serde_json::Value)>,
        tx_builder: TransactionBuilder,
    ) -> anyhow::Result<OutPointRange> {
        Client::finalize_and_submit_transaction_dbtx(
            self,
            dbtx,
            operation_id,
            operation_type,
            &operation_meta_gen,
            tx_builder,
        )
        .await
    }

    async fn finalize_and_submit_transaction_inner(
        &self,
        dbtx: &mut DatabaseTransaction<'_>,
        operation_id: OperationId,
        tx_builder: TransactionBuilder,
    ) -> anyhow::Result<OutPointRange> {
        Client::finalize_and_submit_transaction_inner(self, dbtx, operation_id, tx_builder).await
    }

    async fn fee_quote(
        &self,
        operation_id: OperationId,
        request: FeeQuoteRequest,
    ) -> anyhow::Result<FeeQuote> {
        Client::fee_quote(self, operation_id, request).await
    }

    async fn get_balance_for_unit(&self, unit: AmountUnit) -> anyhow::Result<Amount> {
        Client::get_balance_for_unit(self, unit).await
    }

    async fn transaction_updates(&self, operation_id: OperationId) -> TransactionUpdates {
        Client::transaction_updates(self, operation_id).await
    }

    async fn await_primary_module_outputs(
        &self,
        operation_id: OperationId,
        // TODO: make `impl Iterator<Item = ...>`
        outputs: Vec<OutPoint>,
    ) -> anyhow::Result<()> {
        Client::await_primary_bitcoin_module_outputs(self, operation_id, outputs).await
    }

    fn operation_log(&self) -> &dyn IOperationLog {
        Client::operation_log(self)
    }

    async fn has_active_states(&self, operation_id: OperationId) -> bool {
        Client::has_active_states(self, operation_id).await
    }

    async fn operation_exists(&self, operation_id: OperationId) -> bool {
        Client::operation_exists(self, operation_id).await
    }

    async fn config(&self) -> ClientConfig {
        Client::config(self).await
    }

    fn db(&self) -> &Database {
        Client::db(self)
    }

    fn executor(&self) -> &(maybe_add_send_sync!(dyn IExecutor + 'static)) {
        Client::executor(self)
    }

    async fn invite_code(&self, peer: PeerId) -> Option<InviteCode> {
        Client::invite_code(self, peer).await
    }

    fn get_internal_payment_markers(&self) -> anyhow::Result<(PublicKey, u64)> {
        Client::get_internal_payment_markers(self)
    }

    async fn log_event_json(
        &self,
        dbtx: &mut DatabaseTransaction<'_, NonCommittable>,
        module_kind: Option<ModuleKind>,
        module_id: ModuleInstanceId,
        kind: EventKind,
        payload: serde_json::Value,
        persist: EventPersistence,
    ) {
        dbtx.ensure_global()
            .expect("Must be called with global dbtx");
        self.log_event_raw_dbtx(
            dbtx,
            kind,
            module_kind.map(|kind| (kind, module_id)),
            serde_json::to_vec(&payload).expect("Serialization can't fail"),
            persist,
        )
        .await;
    }

    async fn read_operation_active_states<'dbtx>(
        &self,
        operation_id: OperationId,
        module_id: ModuleInstanceId,
        dbtx: &'dbtx mut DatabaseTransaction<'_>,
    ) -> Pin<Box<maybe_add_send!(dyn Stream<Item = (ActiveStateKey, ActiveStateMeta)> + 'dbtx)>>
    {
        Box::pin(
            dbtx.find_by_prefix(&ActiveModuleOperationStateKeyPrefix {
                operation_id,
                module_instance: module_id,
            })
            .await
            .map(move |(k, v)| (k.0, v)),
        )
    }
    async fn read_operation_inactive_states<'dbtx>(
        &self,
        operation_id: OperationId,
        module_id: ModuleInstanceId,
        dbtx: &'dbtx mut DatabaseTransaction<'_>,
    ) -> Pin<Box<maybe_add_send!(dyn Stream<Item = (InactiveStateKey, InactiveStateMeta)> + 'dbtx)>>
    {
        Box::pin(
            dbtx.find_by_prefix(&InactiveModuleOperationStateKeyPrefix {
                operation_id,
                module_instance: module_id,
            })
            .await
            .map(move |(k, v)| (k.0, v)),
        )
    }
}

// TODO: impl `Debug` for `Client` and derive here
impl fmt::Debug for Client {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        write!(f, "Client")
    }
}

pub fn client_decoders<'a>(
    registry: &ModuleInitRegistry<DynClientModuleInit>,
    module_kinds: impl Iterator<Item = (ModuleInstanceId, &'a ModuleKind)>,
) -> ModuleDecoderRegistry {
    let mut modules = BTreeMap::new();
    for (id, kind) in module_kinds {
        let Some(init) = registry.get(kind) else {
            debug!("Detected configuration for unsupported module id: {id}, kind: {kind}");
            continue;
        };

        modules.insert(
            id,
            (
                kind.clone(),
                IClientModuleInit::decoder(AsRef::<dyn IClientModuleInit + 'static>::as_ref(init)),
            ),
        );
    }
    ModuleDecoderRegistry::from(modules)
}