Skip to main content

fedimint_wallet_client/
lib.rs

1#![deny(clippy::pedantic)]
2#![allow(clippy::cast_possible_truncation)]
3#![allow(clippy::missing_errors_doc)]
4#![allow(clippy::missing_panics_doc)]
5#![allow(clippy::module_name_repetitions)]
6#![allow(clippy::must_use_candidate)]
7
8pub mod api;
9#[cfg(feature = "cli")]
10mod cli;
11
12mod backup;
13
14pub mod client_db;
15/// Legacy, state-machine based peg-ins, replaced by `pegin_monitor`
16/// but retained for time being to ensure existing peg-ins complete.
17mod deposit;
18/// Error types of the wallet client.
19pub mod error;
20pub mod events;
21use events::SendPaymentEvent;
22/// Peg-in monitor: a task monitoring deposit addresses for peg-ins.
23mod pegin_monitor;
24mod withdraw;
25
26use std::collections::{BTreeMap, BTreeSet};
27use std::future;
28use std::sync::Arc;
29use std::time::{Duration, SystemTime};
30
31use async_stream::{stream, try_stream};
32use backup::WalletModuleBackup;
33use bitcoin::address::NetworkUnchecked;
34use bitcoin::secp256k1::{All, SECP256K1, Secp256k1};
35use bitcoin::{Address, Network, ScriptBuf};
36use client_db::{DbKeyPrefix, PegInTweakIndexKey, SupportsSafeDepositKey, TweakIdx};
37use fedimint_api_client::api::{DynModuleApi, FederationResult};
38use fedimint_bitcoind::{
39    BitcoinRpcError, BitcoindTracked, DynBitcoindRpc, IBitcoindRpc, create_esplora_rpc,
40};
41use fedimint_client_module::error::{ClientModuleError, TransactionSubmitError};
42use fedimint_client_module::module::init::{
43    ClientModuleInit, ClientModuleInitArgs, ClientModuleRecoverArgs, RecoveryMode,
44};
45use fedimint_client_module::module::recovery::RecoveryProgress;
46use fedimint_client_module::module::{ClientContext, ClientModule, IClientModule, OutPointRange};
47use fedimint_client_module::oplog::UpdateStreamOrOutcome;
48use fedimint_client_module::sm::{Context, DynState, ModuleNotifier, State, StateTransition};
49use fedimint_client_module::transaction::{
50    ClientOutput, ClientOutputBundle, ClientOutputSM, FeeQuote, FeeQuoteRequest,
51    TransactionBuilder, max_affordable_send_amount,
52};
53use fedimint_client_module::{DynGlobalClientContext, sm_enum_variant_translation};
54use fedimint_core::core::{Decoder, IntoDynInstance, ModuleInstanceId, ModuleKind, OperationId};
55use fedimint_core::db::{
56    Committable, Database, DatabaseError, DatabaseTransaction, IDatabaseTransactionOpsCoreTyped,
57};
58use fedimint_core::encoding::{Decodable, Encodable};
59use fedimint_core::envs::{BitcoinRpcConfig, is_running_in_test_env};
60use fedimint_core::module::{
61    Amounts, ApiAuth, ApiVersion, CommonModuleInit, ModuleCommon, ModuleConsensusVersion,
62    ModuleInit, MultiApiVersion,
63};
64use fedimint_core::task::{MaybeSend, MaybeSync, TaskGroup, sleep};
65use fedimint_core::util::backoff_util::background_backoff;
66use fedimint_core::util::{BoxStream, FmtCompact as _, backoff_util, retry};
67use fedimint_core::{
68    BitcoinHash, OutPoint, TransactionId, apply, async_trait_maybe_send, push_db_pair_items,
69    runtime, secp256k1,
70};
71use fedimint_derive_secret::{ChildId, DerivableSecret};
72use fedimint_logging::LOG_CLIENT_MODULE_WALLET;
73pub use fedimint_wallet_common as common;
74use fedimint_wallet_common::config::{FeeConsensus, WalletClientConfig};
75use fedimint_wallet_common::tweakable::Tweakable;
76pub use fedimint_wallet_common::*;
77use futures::{Stream, StreamExt, TryStreamExt as _};
78use rand::{Rng, thread_rng};
79use secp256k1::Keypair;
80use serde::{Deserialize, Serialize};
81use strum::IntoEnumIterator;
82use tokio::sync::watch;
83use tracing::{debug, instrument, warn};
84
85use crate::api::WalletFederationApi;
86use crate::backup::{FEDERATION_RECOVER_MAX_GAP, RecoveryStateV2, WalletRecovery};
87use crate::client_db::{
88    ClaimedPegInData, ClaimedPegInKey, ClaimedPegInPrefix, NextPegInTweakIndexKey,
89    PegInPoolCursorKey, PegInTweakIndexData, PegInTweakIndexPrefix, RecoveryFinalizedKey,
90    RecoveryStateKey, SupportsSafeDepositPrefix,
91};
92use crate::deposit::DepositStateMachine;
93pub use crate::error::{
94    ConsensusVersionVotingError, DepositAddressError, MaxWithdrawableAmountError, PegInError,
95    PegOutError, SubscribeDepositError, SubscribeWithdrawError, WithdrawFeesError,
96};
97use crate::withdraw::{CreatedWithdrawState, WithdrawStateMachine, WithdrawStates};
98
99const WALLET_TWEAK_CHILD_ID: ChildId = ChildId(0);
100
101#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
102pub struct BitcoinTransactionData {
103    /// The bitcoin transaction is saved as soon as we see it so the transaction
104    /// can be re-transmitted if it's evicted from the mempool.
105    pub btc_transaction: bitcoin::Transaction,
106    /// Index of the deposit output
107    pub out_idx: u32,
108}
109
110#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
111pub enum DepositStateV1 {
112    WaitingForTransaction,
113    WaitingForConfirmation(BitcoinTransactionData),
114    Confirmed(BitcoinTransactionData),
115    Claimed(BitcoinTransactionData),
116    Failed(String),
117}
118
119#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
120pub enum DepositStateV2 {
121    WaitingForTransaction,
122    WaitingForConfirmation {
123        #[serde(with = "bitcoin::amount::serde::as_sat")]
124        btc_deposited: bitcoin::Amount,
125        btc_out_point: bitcoin::OutPoint,
126    },
127    Confirmed {
128        #[serde(with = "bitcoin::amount::serde::as_sat")]
129        btc_deposited: bitcoin::Amount,
130        btc_out_point: bitcoin::OutPoint,
131    },
132    Claimed {
133        #[serde(with = "bitcoin::amount::serde::as_sat")]
134        btc_deposited: bitcoin::Amount,
135        btc_out_point: bitcoin::OutPoint,
136    },
137    Failed(String),
138}
139
140/// Deposit address allocated by this client.
141#[derive(Debug, Clone, PartialEq, Eq)]
142pub struct DepositAddressInfo {
143    pub operation_id: OperationId,
144    pub address: Address,
145    pub tweak_idx: TweakIdx,
146}
147
148/// Result of [`WalletClientModule::allocate_deposit_address_pooled_stateless`].
149///
150/// Callers that need a custom address-picking strategy can use
151/// [`MaybeNewAddress::TooManyUnusedAddresses`] to see all currently reusable
152/// addresses instead of relying on this crate's round-robin policy.
153#[allow(clippy::enum_variant_names)]
154#[derive(Debug, Clone, PartialEq, Eq)]
155pub enum MaybeNewAddress {
156    /// A new tweak/operation was created on this call.
157    NewAddress(DepositAddressInfo),
158    /// The unused-address gap is full. No new address was allocated.
159    ///
160    /// Reusable unused addresses are ordered by `creation_time` ascending.
161    TooManyUnusedAddresses(Vec<DepositAddressInfo>),
162}
163
164/// Outcome of [`WalletClientModule::allocate_deposit_address_pooled`], so
165/// callers can decide whether to perform per-operation initialization (notes,
166/// fee bookkeeping, metadata writes) — for `Reused` returns, that work was
167/// already done at the time of the original `Fresh` allocation and must not be
168/// repeated.
169#[derive(Debug, Clone, Copy, PartialEq, Eq)]
170pub enum AllocateDepositOutcome {
171    /// A new tweak/operation was created on this call.
172    Fresh,
173    /// An existing unused address was returned. The carried `TweakIdx` is the
174    /// same as the one returned in the tuple and refers to the original
175    /// allocation; it's exposed here as well for explicit diagnostic use.
176    Reused { original_tweak_idx: TweakIdx },
177}
178
179#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
180pub enum WithdrawState {
181    Created,
182    Succeeded(bitcoin::Txid),
183    Failed(String),
184    // TODO: track refund
185    // Refunded,
186    // RefundFailed(String),
187}
188
189async fn next_withdraw_state<S>(stream: &mut S) -> Option<WithdrawStates>
190where
191    S: Stream<Item = WalletClientStates> + Unpin,
192{
193    loop {
194        if let WalletClientStates::Withdraw(ds) = stream.next().await? {
195            return Some(ds.state);
196        }
197        tokio::task::yield_now().await;
198    }
199}
200
201#[derive(Debug, Clone, Default)]
202// TODO: should probably move to DB
203pub struct WalletClientInit(pub Option<DynBitcoindRpc>);
204
205const SLICE_SIZE: u64 = 1000;
206
207impl WalletClientInit {
208    pub fn new(rpc: DynBitcoindRpc) -> Self {
209        Self(Some(rpc))
210    }
211
212    async fn recover_from_slices(
213        &self,
214        args: &ClientModuleRecoverArgs<Self>,
215        total_items: u64,
216    ) -> Option<fedimint_core::Amount> {
217        let data = WalletClientModuleData {
218            cfg: args.cfg().clone(),
219            module_root_secret: args.module_root_secret().clone(),
220        };
221
222        let mut state = RecoveryStateV2::new();
223
224        state.refill_pending_pool_up_to(&data, TweakIdx(FEDERATION_RECOVER_MAX_GAP));
225
226        for start in (0..total_items).step_by(SLICE_SIZE as usize) {
227            let end = std::cmp::min(start + SLICE_SIZE, total_items);
228
229            let items = args.module_api().fetch_recovery_slice(start, end).await;
230
231            for item in &items {
232                match item {
233                    RecoveryItem::Input { outpoint, script } => {
234                        state.handle_item(*outpoint, script, &data);
235                    }
236                }
237            }
238
239            args.update_recovery_progress(RecoveryProgress {
240                complete: end.try_into().unwrap_or(u32::MAX),
241                total: total_items.try_into().unwrap_or(u32::MAX),
242            });
243        }
244
245        let mut dbtx = args.db().begin_transaction().await;
246
247        for tweak_idx in 0..state.new_start_idx().0 {
248            let operation_id = data.derive_peg_in_script(TweakIdx(tweak_idx)).3;
249
250            let claimed = state
251                .claimed_outpoints
252                .get(&TweakIdx(tweak_idx))
253                .cloned()
254                .unwrap_or_default();
255
256            dbtx.insert_new_entry(
257                &PegInTweakIndexKey(TweakIdx(tweak_idx)),
258                &PegInTweakIndexData {
259                    operation_id,
260                    creation_time: fedimint_core::time::now(),
261                    last_check_time: None,
262                    next_check_time: Some(fedimint_core::time::now()),
263                    claimed,
264                },
265            )
266            .await;
267        }
268
269        dbtx.insert_new_entry(&NextPegInTweakIndexKey, &state.new_start_idx())
270            .await;
271
272        dbtx.commit_tx().await;
273
274        // The wallet only discovers which on-chain outputs belonged to the
275        // client during recovery; their value isn't known until the deposits
276        // are later claimed, so no amount is reported here.
277        None
278    }
279}
280
281impl ModuleInit for WalletClientInit {
282    type Common = WalletCommonInit;
283
284    async fn dump_database(
285        &self,
286        dbtx: &mut DatabaseTransaction<'_>,
287        prefix_names: Vec<String>,
288    ) -> Box<dyn Iterator<Item = (String, Box<dyn erased_serde::Serialize + Send>)> + '_> {
289        let mut wallet_client_items: BTreeMap<String, Box<dyn erased_serde::Serialize + Send>> =
290            BTreeMap::new();
291        let filtered_prefixes = DbKeyPrefix::iter().filter(|f| {
292            prefix_names.is_empty() || prefix_names.contains(&f.to_string().to_lowercase())
293        });
294
295        for table in filtered_prefixes {
296            match table {
297                DbKeyPrefix::NextPegInTweakIndex => {
298                    if let Some(index) = dbtx.get_value(&NextPegInTweakIndexKey).await {
299                        wallet_client_items
300                            .insert("NextPegInTweakIndex".to_string(), Box::new(index));
301                    }
302                }
303                DbKeyPrefix::PegInTweakIndex => {
304                    push_db_pair_items!(
305                        dbtx,
306                        PegInTweakIndexPrefix,
307                        PegInTweakIndexKey,
308                        PegInTweakIndexData,
309                        wallet_client_items,
310                        "Peg-In Tweak Index"
311                    );
312                }
313                DbKeyPrefix::ClaimedPegIn => {
314                    push_db_pair_items!(
315                        dbtx,
316                        ClaimedPegInPrefix,
317                        ClaimedPegInKey,
318                        ClaimedPegInData,
319                        wallet_client_items,
320                        "Claimed Peg-In"
321                    );
322                }
323                DbKeyPrefix::RecoveryFinalized => {
324                    if let Some(val) = dbtx.get_value(&RecoveryFinalizedKey).await {
325                        wallet_client_items.insert("RecoveryFinalized".to_string(), Box::new(val));
326                    }
327                }
328                DbKeyPrefix::SupportsSafeDeposit => {
329                    push_db_pair_items!(
330                        dbtx,
331                        SupportsSafeDepositPrefix,
332                        SupportsSafeDepositKey,
333                        (),
334                        wallet_client_items,
335                        "Supports Safe Deposit"
336                    );
337                }
338                DbKeyPrefix::PegInPoolCursor => {
339                    if let Some(cursor) = dbtx.get_value(&PegInPoolCursorKey).await {
340                        wallet_client_items.insert("PegInPoolCursor".to_string(), Box::new(cursor));
341                    }
342                }
343                DbKeyPrefix::RecoveryState
344                | DbKeyPrefix::ExternalReservedStart
345                | DbKeyPrefix::CoreInternalReservedStart
346                | DbKeyPrefix::CoreInternalReservedEnd => {}
347            }
348        }
349
350        Box::new(wallet_client_items.into_iter())
351    }
352}
353
354#[apply(async_trait_maybe_send!)]
355impl ClientModuleInit for WalletClientInit {
356    type Module = WalletClientModule;
357
358    fn supported_api_versions(&self) -> MultiApiVersion {
359        MultiApiVersion::try_from_iter([ApiVersion { major: 0, minor: 0 }])
360            .expect("no version conflicts")
361    }
362
363    async fn init(
364        &self,
365        args: &ClientModuleInitArgs<Self>,
366    ) -> Result<Self::Module, ClientModuleError> {
367        let data = WalletClientModuleData {
368            cfg: args.cfg().clone(),
369            module_root_secret: args.module_root_secret().clone(),
370        };
371
372        let db = args.db().clone();
373
374        let rpc_config = WalletClientModule::get_rpc_config(args.cfg());
375
376        // Priority:
377        // 1. user-provided bitcoind RPC from ClientBuilder::with_bitcoind_rpc
378        // 2. user-provided no-chain-id factory from
379        //    ClientBuilder::with_bitcoind_rpc_no_chain_id
380        // 3. WalletClientInit constructor
381        // 4. create from config (esplora)
382        let btc_rpc = if let Some(user_rpc) = args.user_bitcoind_rpc() {
383            user_rpc.clone()
384        } else if let Some(factory) = args.user_bitcoind_rpc_no_chain_id() {
385            if let Some(rpc) = factory(rpc_config.url.clone()).await {
386                rpc
387            } else {
388                self.0.clone().unwrap_or(
389                    create_esplora_rpc(&rpc_config.url).map_err(ClientModuleError::other)?,
390                )
391            }
392        } else {
393            self.0
394                .clone()
395                .unwrap_or(create_esplora_rpc(&rpc_config.url).map_err(ClientModuleError::other)?)
396        };
397        let btc_rpc = BitcoindTracked::new(btc_rpc, "wallet-client").into_dyn();
398
399        let module_api = args.module_api().clone();
400
401        let (pegin_claimed_sender, pegin_claimed_receiver) = watch::channel(());
402        let (pegin_monitor_wakeup_sender, pegin_monitor_wakeup_receiver) = watch::channel(());
403
404        Ok(WalletClientModule {
405            db,
406            data,
407            module_api,
408            notifier: args.notifier().clone(),
409            rpc: btc_rpc,
410            client_ctx: args.context(),
411            pegin_monitor_wakeup_sender,
412            pegin_monitor_wakeup_receiver,
413            pegin_claimed_receiver,
414            pegin_claimed_sender,
415            task_group: args.task_group().clone(),
416            client_span: args.client_span().clone(),
417            admin_auth: args.admin_auth().cloned(),
418        })
419    }
420
421    fn recovery_mode(&self) -> RecoveryMode {
422        RecoveryMode::Unusable
423    }
424
425    /// Wallet recovery
426    ///
427    /// Uses slice-based recovery if supported by the federation, otherwise
428    /// falls back to session-based history recovery.
429    async fn recover(
430        &self,
431        args: &ClientModuleRecoverArgs<Self>,
432        snapshot: Option<&<Self::Module as ClientModule>::Backup>,
433    ) -> Result<Option<fedimint_core::Amount>, ClientModuleError> {
434        // Check if V1 (session-based) recovery state exists (resuming interrupted
435        // recovery)
436        if args
437            .db()
438            .begin_transaction_nc()
439            .await
440            .get_value(&RecoveryStateKey)
441            .await
442            .is_some()
443        {
444            return args
445                .recover_from_history::<WalletRecovery>(self, snapshot)
446                .await;
447        }
448
449        // Determine which method to use based on endpoint availability. The
450        // count fetched here is reused by the slice recovery so that it is not
451        // requested a second time (and cannot transiently fail there).
452        match args.module_api().fetch_recovery_count().await {
453            Ok(total_items) => Ok(self.recover_from_slices(args, total_items).await),
454            Err(_) => {
455                args.recover_from_history::<WalletRecovery>(self, snapshot)
456                    .await
457            }
458        }
459    }
460
461    fn used_db_prefixes(&self) -> Option<BTreeSet<u8>> {
462        Some(
463            DbKeyPrefix::iter()
464                .map(|p| p as u8)
465                .chain(
466                    DbKeyPrefix::ExternalReservedStart as u8
467                        ..=DbKeyPrefix::CoreInternalReservedEnd as u8,
468                )
469                .collect(),
470        )
471    }
472}
473
474#[derive(Debug, Clone, Serialize, Deserialize)]
475pub struct WalletOperationMeta {
476    pub variant: WalletOperationMetaVariant,
477    pub extra_meta: serde_json::Value,
478}
479
480#[derive(Debug, Clone, Serialize, Deserialize)]
481#[serde(rename_all = "snake_case")]
482pub enum WalletOperationMetaVariant {
483    Deposit {
484        address: Address<NetworkUnchecked>,
485        /// Added in 0.4.2, can be `None` for old deposits or `Some` for ones
486        /// using the pegin monitor. The value is the child index of the key
487        /// used to generate the address, so we can re-generate the secret key
488        /// from our root secret.
489        #[serde(default)]
490        tweak_idx: Option<TweakIdx>,
491        #[serde(default, skip_serializing_if = "Option::is_none")]
492        expires_at: Option<SystemTime>,
493    },
494    Withdraw {
495        address: Address<NetworkUnchecked>,
496        #[serde(with = "bitcoin::amount::serde::as_sat")]
497        amount: bitcoin::Amount,
498        fee: PegOutFees,
499        change: Vec<OutPoint>,
500    },
501
502    RbfWithdraw {
503        rbf: Rbf,
504        change: Vec<OutPoint>,
505    },
506}
507
508/// The non-resource, just plain-data parts of [`WalletClientModule`]
509#[derive(Debug, Clone)]
510pub struct WalletClientModuleData {
511    cfg: WalletClientConfig,
512    module_root_secret: DerivableSecret,
513}
514
515impl WalletClientModuleData {
516    fn derive_deposit_address(
517        &self,
518        idx: TweakIdx,
519    ) -> (Keypair, secp256k1::PublicKey, Address, OperationId) {
520        let idx = ChildId(idx.0);
521
522        let secret_tweak_key = self
523            .module_root_secret
524            .child_key(WALLET_TWEAK_CHILD_ID)
525            .child_key(idx)
526            .to_secp_key(fedimint_core::secp256k1::SECP256K1);
527
528        let public_tweak_key = secret_tweak_key.public_key();
529
530        let address = self
531            .cfg
532            .peg_in_descriptor
533            .tweak(&public_tweak_key, bitcoin::secp256k1::SECP256K1)
534            .address(self.cfg.network.0)
535            .unwrap();
536
537        // TODO: make hash?
538        let operation_id = OperationId(public_tweak_key.x_only_public_key().0.serialize());
539
540        (secret_tweak_key, public_tweak_key, address, operation_id)
541    }
542
543    fn derive_peg_in_script(
544        &self,
545        idx: TweakIdx,
546    ) -> (ScriptBuf, bitcoin::Address, Keypair, OperationId) {
547        let (secret_tweak_key, _, address, operation_id) = self.derive_deposit_address(idx);
548
549        (
550            self.cfg
551                .peg_in_descriptor
552                .tweak(&secret_tweak_key.public_key(), SECP256K1)
553                .script_pubkey(),
554            address,
555            secret_tweak_key,
556            operation_id,
557        )
558    }
559}
560
561#[derive(Debug)]
562pub struct WalletClientModule {
563    data: WalletClientModuleData,
564    db: Database,
565    module_api: DynModuleApi,
566    notifier: ModuleNotifier<WalletClientStates>,
567    rpc: DynBitcoindRpc,
568    client_ctx: ClientContext<Self>,
569    /// Updated to wake up pegin monitor
570    pegin_monitor_wakeup_sender: watch::Sender<()>,
571    pegin_monitor_wakeup_receiver: watch::Receiver<()>,
572    /// Called every time a peg-in was claimed
573    pegin_claimed_sender: watch::Sender<()>,
574    pegin_claimed_receiver: watch::Receiver<()>,
575    task_group: TaskGroup,
576    client_span: tracing::Span,
577    admin_auth: Option<ApiAuth>,
578}
579
580#[apply(async_trait_maybe_send!)]
581impl ClientModule for WalletClientModule {
582    type Init = WalletClientInit;
583    type Common = WalletModuleTypes;
584    type Backup = WalletModuleBackup;
585    type ModuleStateMachineContext = WalletClientContext;
586    type States = WalletClientStates;
587
588    fn context(&self) -> Self::ModuleStateMachineContext {
589        WalletClientContext {
590            rpc: self.rpc.clone(),
591            wallet_descriptor: self.cfg().peg_in_descriptor.clone(),
592            wallet_decoder: self.decoder(),
593            secp: Secp256k1::default(),
594            client_ctx: self.client_ctx.clone(),
595        }
596    }
597
598    async fn start(&self) {
599        self.task_group
600            .spawn_cancellable_with_span(self.client_span.clone(), "peg-in monitor", {
601                let client_ctx = self.client_ctx.clone();
602                let db = self.db.clone();
603                let btc_rpc = self.rpc.clone();
604                let module_api = self.module_api.clone();
605                let data = self.data.clone();
606                let pegin_claimed_sender = self.pegin_claimed_sender.clone();
607                let pegin_monitor_wakeup_receiver = self.pegin_monitor_wakeup_receiver.clone();
608                pegin_monitor::run_peg_in_monitor(
609                    client_ctx,
610                    db,
611                    btc_rpc,
612                    module_api,
613                    data,
614                    pegin_claimed_sender,
615                    pegin_monitor_wakeup_receiver,
616                )
617            });
618
619        self.task_group.spawn_cancellable_with_span(
620            self.client_span.clone(),
621            "supports-safe-deposit-version",
622            {
623                let db = self.db.clone();
624                let module_api = self.module_api.clone();
625
626                poll_supports_safe_deposit_version(db, module_api)
627            },
628        );
629    }
630
631    fn supports_backup(&self) -> bool {
632        true
633    }
634
635    async fn backup(&self) -> Result<backup::WalletModuleBackup, ClientModuleError> {
636        // fetch consensus height first
637        let session_count = self
638            .client_ctx
639            .global_api()
640            .session_count()
641            .await
642            .map_err(ClientModuleError::other)?;
643
644        let mut dbtx = self.db.begin_transaction_nc().await;
645        let next_pegin_tweak_idx = dbtx
646            .get_value(&NextPegInTweakIndexKey)
647            .await
648            .unwrap_or_default();
649        let claimed = dbtx
650            .find_by_prefix(&PegInTweakIndexPrefix)
651            .await
652            .filter_map(|(k, v)| async move {
653                if v.claimed.is_empty() {
654                    None
655                } else {
656                    Some(k.0)
657                }
658            })
659            .collect()
660            .await;
661        Ok(backup::WalletModuleBackup::new_v1(
662            session_count,
663            next_pegin_tweak_idx,
664            claimed,
665        ))
666    }
667
668    fn input_fee(
669        &self,
670        _amount: &Amounts,
671        _input: &<Self::Common as ModuleCommon>::Input,
672    ) -> Option<Amounts> {
673        Some(Amounts::new_bitcoin(self.cfg().fee_consensus.peg_in_abs))
674    }
675
676    fn output_fee(
677        &self,
678        _amount: &Amounts,
679        _output: &<Self::Common as ModuleCommon>::Output,
680    ) -> Option<Amounts> {
681        Some(Amounts::new_bitcoin(self.cfg().fee_consensus.peg_out_abs))
682    }
683
684    async fn handle_rpc(
685        &self,
686        method: String,
687        request: serde_json::Value,
688    ) -> BoxStream<'_, Result<serde_json::Value, ClientModuleError>> {
689        let stream: BoxStream<'_, Result<serde_json::Value, RpcError>> = Box::pin(try_stream! {
690            match method.as_str() {
691                "get_wallet_summary" => {
692                    let _req: WalletSummaryRequest = serde_json::from_value(request)?;
693                    let wallet_summary = self.get_wallet_summary()
694                        .await
695                        .expect("Failed to fetch wallet summary");
696                    let result = serde_json::to_value(&wallet_summary)
697                        .expect("Serialization error");
698                    yield result;
699                }
700                "get_block_count_local" => {
701                    let block_count = self.get_block_count_local().await
702                        .expect("Failed to fetch block count");
703                    yield serde_json::to_value(block_count)?;
704                }
705                "peg_in" => {
706                    let req: PegInRequest = serde_json::from_value(request)?;
707                    let response = self.peg_in(req)
708                        .await
709                        .map_err(RpcError::PegIn)?;
710                    let result = serde_json::to_value(&response)?;
711                    yield result;
712                },
713                "peg_out" => {
714                    let req: PegOutRequest = serde_json::from_value(request)?;
715                    let response = self.peg_out(req)
716                        .await
717                        .map_err(RpcError::PegOut)?;
718                    let result = serde_json::to_value(&response)?;
719                    yield result;
720                },
721                "subscribe_deposit" => {
722                    let req: SubscribeDepositRequest = serde_json::from_value(request)?;
723                    for await state in self.subscribe_deposit(req.operation_id).await?.into_stream() {
724                        yield serde_json::to_value(state)?;
725                    }
726                },
727                "subscribe_withdraw" => {
728                    let req: SubscribeWithdrawRequest = serde_json::from_value(request)?;
729                    for await state in self.subscribe_withdraw_updates(req.operation_id).await?.into_stream(){
730                        yield serde_json::to_value(state)?;
731                    }
732                }
733                _ => {
734                    Err(RpcError::UnknownMethod { method: method.clone() })?;
735                }
736            }
737        });
738        Box::pin(stream.map_err(ClientModuleError::other))
739    }
740
741    #[cfg(feature = "cli")]
742    async fn handle_cli_command(
743        &self,
744        args: &[std::ffi::OsString],
745    ) -> Result<serde_json::Value, ClientModuleError> {
746        cli::handle_cli_command(self, args)
747            .await
748            .map_err(ClientModuleError::other)
749    }
750}
751
752/// A failure of a wallet module RPC request.
753#[derive(Debug, thiserror::Error)]
754enum RpcError {
755    /// The request's parameters do not fit the method, or its response could
756    /// not be serialized.
757    #[error(transparent)]
758    Json(#[from] serde_json::Error),
759
760    /// The peg-in could not be started.
761    #[error("peg_in failed")]
762    PegIn(#[source] DepositAddressError),
763
764    /// The peg-out could not be started.
765    #[error("peg_out failed")]
766    PegOut(#[source] PegOutError),
767
768    /// The deposit's updates could not be followed.
769    #[error(transparent)]
770    SubscribeDeposit(#[from] SubscribeDepositError),
771
772    /// The withdrawal's updates could not be followed.
773    #[error(transparent)]
774    SubscribeWithdraw(#[from] SubscribeWithdrawError),
775
776    /// The request names a method the module does not have.
777    #[error("Unknown method: {method}")]
778    UnknownMethod { method: String },
779}
780
781#[derive(Deserialize)]
782struct WalletSummaryRequest {}
783
784#[derive(Debug, Clone)]
785pub struct WalletClientContext {
786    rpc: DynBitcoindRpc,
787    wallet_descriptor: PegInDescriptor,
788    wallet_decoder: Decoder,
789    secp: Secp256k1<All>,
790    pub client_ctx: ClientContext<WalletClientModule>,
791}
792
793#[derive(Debug, Clone, Serialize, Deserialize)]
794pub struct PegInRequest {
795    pub extra_meta: serde_json::Value,
796}
797
798#[derive(Deserialize)]
799struct SubscribeDepositRequest {
800    operation_id: OperationId,
801}
802
803#[derive(Deserialize)]
804struct SubscribeWithdrawRequest {
805    operation_id: OperationId,
806}
807
808#[derive(Debug, Clone, Serialize, Deserialize)]
809pub struct PegInResponse {
810    pub deposit_address: Address<NetworkUnchecked>,
811    pub operation_id: OperationId,
812}
813
814#[derive(Debug, Clone, Serialize, Deserialize)]
815pub struct PegOutRequest {
816    pub amount_sat: u64,
817    pub destination_address: Address<NetworkUnchecked>,
818    pub extra_meta: serde_json::Value,
819}
820
821#[derive(Debug, Clone, Serialize, Deserialize)]
822pub struct PegOutResponse {
823    pub operation_id: OperationId,
824}
825
826impl Context for WalletClientContext {
827    const KIND: Option<ModuleKind> = Some(KIND);
828}
829
830impl WalletClientModule {
831    fn cfg(&self) -> &WalletClientConfig {
832        &self.data.cfg
833    }
834
835    fn get_rpc_config(cfg: &WalletClientConfig) -> BitcoinRpcConfig {
836        match BitcoinRpcConfig::get_defaults_from_env_vars() {
837            Ok(rpc_config) => {
838                // TODO: Wallet client cannot support bitcoind RPC until the bitcoin dep is
839                // updated to 0.30
840                if rpc_config.kind == "bitcoind" {
841                    cfg.default_bitcoin_rpc.clone()
842                } else {
843                    rpc_config
844                }
845            }
846            _ => cfg.default_bitcoin_rpc.clone(),
847        }
848    }
849
850    pub fn get_network(&self) -> Network {
851        self.cfg().network.0
852    }
853
854    pub fn get_finality_delay(&self) -> u32 {
855        self.cfg().finality_delay
856    }
857
858    pub fn get_fee_consensus(&self) -> FeeConsensus {
859        self.cfg().fee_consensus
860    }
861
862    async fn allocate_deposit_address_inner(
863        &self,
864        dbtx: &mut DatabaseTransaction<'_>,
865    ) -> DepositAddressInfo {
866        dbtx.ensure_isolated().expect("Must be isolated db");
867
868        let tweak_idx = get_next_peg_in_tweak_child_id(dbtx).await;
869        let (_secret_tweak_key, _, address, operation_id) =
870            self.data.derive_deposit_address(tweak_idx);
871
872        let now = fedimint_core::time::now();
873
874        dbtx.insert_new_entry(
875            &PegInTweakIndexKey(tweak_idx),
876            &PegInTweakIndexData {
877                creation_time: now,
878                next_check_time: Some(now),
879                last_check_time: None,
880                operation_id,
881                claimed: vec![],
882            },
883        )
884        .await;
885
886        DepositAddressInfo {
887            operation_id,
888            address,
889            tweak_idx,
890        }
891    }
892
893    /// Fetches the fees that would need to be paid to make the withdraw request
894    /// using [`Self::withdraw`] work *right now*.
895    ///
896    /// Note that we do not receive a guarantee that these fees will be valid in
897    /// the future, thus even the next second using these fees *may* fail.
898    /// The caller should be prepared to retry with a new fee estimate.
899    pub async fn get_withdraw_fees(
900        &self,
901        address: &bitcoin::Address,
902        amount: bitcoin::Amount,
903    ) -> Result<PegOutFees, WithdrawFeesError> {
904        self.module_api
905            .fetch_peg_out_fees(address, amount)
906            .await?
907            .ok_or(WithdrawFeesError::NoQuote)
908    }
909
910    /// Computes the federation fee a peg-out of an on-chain output worth
911    /// `amount` (the withdrawal amount plus the on-chain miner fee) would
912    /// incur, without submitting anything.
913    ///
914    /// A peg-out submits a single wallet output worth `amount`; the primary
915    /// module balances it by spending ecash to fund the output and minting any
916    /// change. This quotes the fee of that transaction — the wallet (peg-out)
917    /// output fee, the mint input fees on the funding notes, any mint change
918    /// output fees, and sub-denomination dust — via the shared, module-agnostic
919    /// fee quote.
920    ///
921    /// The on-chain Bitcoin miner fee is deliberately excluded: it is part of
922    /// the output `amount` (see [`Self::get_withdraw_fees`]), not the
923    /// on-federation transaction fee.
924    pub async fn send_fee_quote(
925        &self,
926        amount: bitcoin::Amount,
927    ) -> Result<FeeQuote, TransactionSubmitError> {
928        let amount = fedimint_core::Amount::from_sats(amount.to_sat());
929        self.client_ctx
930            .fee_quote(
931                OperationId::new_random(),
932                FeeQuoteRequest {
933                    input_amount: Amounts::ZERO,
934                    output_amount: Amounts::new_bitcoin(amount),
935                    input_fee: Amounts::ZERO,
936                    output_fee: Amounts::new_bitcoin(self.cfg().fee_consensus.peg_out_abs),
937                },
938            )
939            .await
940    }
941
942    /// Finds the largest amount that can be withdrawn in full out of
943    /// `balance` — the amount a "withdraw everything" sweep should use —
944    /// together with the peg-out fees to submit it with.
945    ///
946    /// Withdrawing `amount` submits a peg-out output worth `amount +
947    /// peg_out_fees` (the on-chain miner fee is carried inside the output, see
948    /// [`WalletOutputV0::amount`]) and funding that output costs an additional
949    /// federation fee — the peg-out output fee, the mint input fees on the
950    /// funding notes, any mint change output fees and sub-denomination dust —
951    /// as quoted by [`Self::send_fee_quote`].
952    ///
953    /// Both returned values must be passed to [`Self::withdraw`] together. The
954    /// federation rebuilds the peg-out for the exact amount submitted and
955    /// rejects it unless the declared `total_weight` matches what it computed
956    /// (`WalletOutputError::TxWeightIncorrect`), so the fees are re-quoted here
957    /// at the amount being returned rather than at the balance.
958    ///
959    /// This costs exactly two [`Self::get_withdraw_fees`] round trips. The
960    /// first, at the full balance, is an upper bound on the miner fee — the
961    /// federation selects UTXOs until they cover the amount, so the weight is
962    /// monotone in it — and sizing the withdrawal against that upper bound can
963    /// only leave the result more affordable once the true, smaller fee
964    /// replaces it. The maximum itself is then found by binary search over the
965    /// real fee quote (see [`max_affordable_send_amount`]), which needs no
966    /// further round trips because [`Self::send_fee_quote`] is a local dry-run.
967    ///
968    /// Sizing against the upper bound means a sweep may leave a small
969    /// remainder behind. The quote is point-in-time: if the federation's UTXO
970    /// set or feerate moves before the peg-out is processed it is rejected, and
971    /// the caller should retry with a fresh quote.
972    ///
973    /// Returns [`MaxWithdrawableAmountError::BalanceTooLow`] if the balance
974    /// cannot cover the destination's dust limit plus fees, or
975    /// [`MaxWithdrawableAmountError::Quote`] if the fee probe itself failed.
976    pub async fn max_withdrawable_amount(
977        &self,
978        address: &bitcoin::Address,
979        balance: fedimint_core::Amount,
980    ) -> Result<(bitcoin::Amount, PegOutFees), MaxWithdrawableAmountError> {
981        // Upper bound on the miner fee: the weight only grows as the federation
982        // has to reach for more UTXOs, so no smaller withdrawal costs more.
983        let max_fees = self
984            .get_withdraw_fees(
985                address,
986                bitcoin::Amount::from_sat(balance.sats_round_down()),
987            )
988            .await?;
989        let max_fee_msats = fedimint_core::Amount::from_sats(max_fees.amount().to_sat());
990        let dust_limit = address.script_pubkey().minimal_non_dust();
991
992        let max = max_affordable_send_amount(
993            balance,
994            fedimint_core::Amount::from_sats(dust_limit.to_sat()),
995            balance,
996            // A peg-out funds a whole number of sats, so round the probe up to
997            // the next sat before adding the miner fee; the solver searches
998            // millisatoshis.
999            |amount: fedimint_core::Amount| {
1000                fedimint_core::Amount::from_sats(amount.msats.div_ceil(1000)) + max_fee_msats
1001            },
1002            |funded: fedimint_core::Amount| {
1003                self.send_fee_quote(bitcoin::Amount::from_sat(funded.msats / 1000))
1004            },
1005        )
1006        .await
1007        .map_err(MaxWithdrawableAmountError::Quote)?
1008        .ok_or(MaxWithdrawableAmountError::BalanceTooLow {
1009            balance,
1010            dust_limit,
1011        })?;
1012
1013        // `gross_up` rounded up to whole satoshis, so the largest affordable
1014        // amount already sits on a satoshi boundary; no value is lost here.
1015        let amount = bitcoin::Amount::from_sat(max.msats.div_ceil(1000));
1016
1017        // Re-quote at the amount actually being withdrawn. This is the fee the
1018        // federation recomputes for that amount, so the declared weight
1019        // matches; it is at most `max_fees`, which is what keeps `amount`
1020        // affordable.
1021        let fees = self.get_withdraw_fees(address, amount).await?;
1022
1023        Ok((amount, fees))
1024    }
1025
1026    /// Returns a summary of the wallet's coins
1027    pub async fn get_wallet_summary(&self) -> FederationResult<WalletSummary> {
1028        self.module_api.fetch_wallet_summary().await
1029    }
1030
1031    pub async fn get_block_count_local(&self) -> FederationResult<u32> {
1032        self.module_api.fetch_block_count_local().await
1033    }
1034
1035    pub fn create_withdraw_output(
1036        &self,
1037        operation_id: OperationId,
1038        address: bitcoin::Address,
1039        amount: bitcoin::Amount,
1040        fees: PegOutFees,
1041    ) -> ClientOutputBundle<WalletOutput, WalletClientStates> {
1042        let output = WalletOutput::new_v0_peg_out(address, amount, fees);
1043
1044        let amount = output.maybe_v0_ref().expect("v0 output").amount().into();
1045
1046        let sm_gen = move |out_point_range: OutPointRange| {
1047            assert_eq!(out_point_range.count(), 1);
1048            let out_idx = out_point_range.start_idx();
1049            vec![WalletClientStates::Withdraw(WithdrawStateMachine {
1050                operation_id,
1051                state: WithdrawStates::Created(CreatedWithdrawState {
1052                    fm_outpoint: OutPoint {
1053                        txid: out_point_range.txid(),
1054                        out_idx,
1055                    },
1056                }),
1057            })]
1058        };
1059
1060        ClientOutputBundle::new(
1061            vec![ClientOutput::<WalletOutput> {
1062                output,
1063                amounts: Amounts::new_bitcoin(amount),
1064            }],
1065            vec![ClientOutputSM::<WalletClientStates> {
1066                state_machines: Arc::new(sm_gen),
1067            }],
1068        )
1069    }
1070
1071    pub async fn peg_in(&self, req: PegInRequest) -> Result<PegInResponse, DepositAddressError> {
1072        let deposit_address = self.safe_allocate_deposit_address(req.extra_meta).await?;
1073
1074        Ok(PegInResponse {
1075            deposit_address: deposit_address.address.into_unchecked(),
1076            operation_id: deposit_address.operation_id,
1077        })
1078    }
1079
1080    pub async fn peg_out(&self, req: PegOutRequest) -> Result<PegOutResponse, PegOutError> {
1081        let amount = bitcoin::Amount::from_sat(req.amount_sat);
1082        let network = self.get_network();
1083        let destination = req
1084            .destination_address
1085            .require_network(network)
1086            .map_err(|_| PegOutError::WrongNetwork { expected: network })?;
1087
1088        let fees = self.get_withdraw_fees(&destination, amount).await?;
1089        let operation_id = self
1090            .withdraw(&destination, amount, fees, req.extra_meta)
1091            .await?;
1092
1093        Ok(PegOutResponse { operation_id })
1094    }
1095
1096    pub fn create_rbf_withdraw_output(
1097        &self,
1098        operation_id: OperationId,
1099        rbf: &Rbf,
1100    ) -> ClientOutputBundle<WalletOutput, WalletClientStates> {
1101        let output = WalletOutput::new_v0_rbf(rbf.fees, rbf.txid);
1102
1103        let amount = output.maybe_v0_ref().expect("v0 output").amount().into();
1104
1105        let sm_gen = move |out_point_range: OutPointRange| {
1106            assert_eq!(out_point_range.count(), 1);
1107            let out_idx = out_point_range.start_idx();
1108            vec![WalletClientStates::Withdraw(WithdrawStateMachine {
1109                operation_id,
1110                state: WithdrawStates::Created(CreatedWithdrawState {
1111                    fm_outpoint: OutPoint {
1112                        txid: out_point_range.txid(),
1113                        out_idx,
1114                    },
1115                }),
1116            })]
1117        };
1118
1119        ClientOutputBundle::new(
1120            vec![ClientOutput::<WalletOutput> {
1121                output,
1122                amounts: Amounts::new_bitcoin(amount),
1123            }],
1124            vec![ClientOutputSM::<WalletClientStates> {
1125                state_machines: Arc::new(sm_gen),
1126            }],
1127        )
1128    }
1129
1130    pub async fn btc_tx_has_no_size_limit(&self) -> FederationResult<bool> {
1131        Ok(self.module_api.module_consensus_version().await? >= ModuleConsensusVersion::new(2, 2))
1132    }
1133
1134    /// Returns true if the federation's wallet module consensus version
1135    /// supports processing all deposits.
1136    ///
1137    /// This method is safe to call offline, since it first attempts to read a
1138    /// key from the db that represents the client has previously been able to
1139    /// verify the wallet module consensus version. If the client has not
1140    /// verified the version, it must be online to fetch the latest wallet
1141    /// module consensus version.
1142    pub async fn supports_safe_deposit(&self) -> bool {
1143        if supports_safe_deposit_verified(&self.db).await {
1144            return true;
1145        }
1146
1147        verify_supports_safe_deposit(&self.db, &self.module_api)
1148            .await
1149            .unwrap_or(false)
1150    }
1151
1152    /// Allocates a deposit address controlled by the federation, guaranteeing
1153    /// safe handling of all deposits, including on-chain transactions exceeding
1154    /// `ALEPH_BFT_UNIT_BYTE_LIMIT`.
1155    ///
1156    /// Returns an error if the client has never been online to verify the
1157    /// federation's wallet module consensus version supports processing all
1158    /// deposits.
1159    pub async fn safe_allocate_deposit_address<M>(
1160        &self,
1161        extra_meta: M,
1162    ) -> Result<DepositAddressInfo, DepositAddressError>
1163    where
1164        M: Serialize + MaybeSend + MaybeSync,
1165    {
1166        if !self.supports_safe_deposit().await {
1167            return Err(DepositAddressError::SafeDepositUnverified);
1168        }
1169
1170        self.allocate_deposit_address_expert_only(extra_meta).await
1171    }
1172
1173    /// Allocates a deposit address that is controlled by the federation.
1174    ///
1175    /// This is an EXPERT ONLY method intended for power users such as Lightning
1176    /// gateways allocating liquidity, and we discourage exposing peg-in
1177    /// functionality to everyday users of a Fedimint wallet due to the
1178    /// following two limitations:
1179    ///
1180    /// The transaction sending to this address needs to be smaller than 40KB in
1181    /// order for the peg-in to be claimable. If the transaction is too large,
1182    /// funds will be lost.
1183    ///
1184    /// In the future, federations will also enforce a minimum peg-in amount to
1185    /// prevent accumulation of dust UTXOs. Peg-ins under this minimum cannot be
1186    /// claimed and funds will be lost.
1187    ///
1188    /// Everyday users should rely on Lightning to move funds into the
1189    /// federation.
1190    pub async fn allocate_deposit_address_expert_only<M>(
1191        &self,
1192        extra_meta: M,
1193    ) -> Result<DepositAddressInfo, DepositAddressError>
1194    where
1195        M: Serialize + MaybeSend + MaybeSync,
1196    {
1197        let extra_meta_value =
1198            serde_json::to_value(extra_meta).expect("Failed to serialize extra meta");
1199        let deposit_address = self
1200            .db
1201            .autocommit(
1202                move |dbtx, _| {
1203                    let extra_meta_value_inner = extra_meta_value.clone();
1204                    Box::pin(async move {
1205                        let deposit_address = self.allocate_deposit_address_inner(dbtx).await;
1206
1207                        self.client_ctx
1208                            .manual_operation_start_dbtx(
1209                                dbtx,
1210                                deposit_address.operation_id,
1211                                WalletCommonInit::KIND.as_str(),
1212                                WalletOperationMeta {
1213                                    variant: WalletOperationMetaVariant::Deposit {
1214                                        address: deposit_address.address.clone().into_unchecked(),
1215                                        tweak_idx: Some(deposit_address.tweak_idx),
1216                                        expires_at: None,
1217                                    },
1218                                    extra_meta: extra_meta_value_inner,
1219                                },
1220                                vec![],
1221                            )
1222                            .await?;
1223
1224                        debug!(
1225                            target: LOG_CLIENT_MODULE_WALLET,
1226                            tweak_idx = %deposit_address.tweak_idx,
1227                            address = %deposit_address.address,
1228                            "Derived a new deposit address"
1229                        );
1230
1231                        // Begin watching the script address
1232                        self.rpc
1233                            .watch_script_history(&deposit_address.address.script_pubkey())
1234                            .await?;
1235
1236                        let sender = self.pegin_monitor_wakeup_sender.clone();
1237                        dbtx.on_commit(move || {
1238                            sender.send_replace(());
1239                        });
1240
1241                        Ok(deposit_address)
1242                    })
1243                },
1244                Some(100),
1245            )
1246            .await?;
1247
1248        Ok(deposit_address)
1249    }
1250
1251    /// Allocate a deposit address, bounding the gap of consecutive unused
1252    /// addresses past the last-used one, without selecting a reusable address.
1253    ///
1254    /// "Unused" here means `PegInTweakIndexData::claimed.is_empty()` — the
1255    /// `pegin_monitor` has not yet successfully claimed any deposit on this
1256    /// address. There is a small race window where a deposit has been
1257    /// observed in the mempool / awaiting confirmations but not yet claimed,
1258    /// in which case the address still looks unused. Reusing such an address
1259    /// is benign in practice: `pegin_monitor` claims any number of deposits
1260    /// per address.
1261    ///
1262    /// Let `gap` be the count of unused tweak indices strictly greater than
1263    /// the highest used index (or the count of all allocated tweaks if none
1264    /// have been used yet). Semantics:
1265    ///   - If `gap >= max_gap_size` and there is at least one unused address,
1266    ///     returns all those addresses with
1267    ///     [`MaybeNewAddress::TooManyUnusedAddresses`].
1268    ///   - Otherwise, allocates a new address (equivalent to
1269    ///     [`Self::allocate_deposit_address_expert_only`]) and returns it with
1270    ///     [`MaybeNewAddress::NewAddress`].
1271    ///
1272    /// `max_gap_size == 0` therefore means "only allocate fresh when the
1273    /// previous address was already used". The very first call still
1274    /// allocates fresh because there are no addresses to reuse.
1275    ///
1276    /// `max_gap_size == usize::MAX` makes this method behave like
1277    /// [`Self::allocate_deposit_address_expert_only`] — always allocating a new
1278    /// address.
1279    ///
1280    /// Reusable addresses are ordered by `creation_time` ascending. This lets
1281    /// callers that need a custom selection strategy choose from all available
1282    /// candidates.
1283    ///
1284    /// Caveats inherited from
1285    /// [`Self::allocate_deposit_address_expert_only`] (40KB tx limit, future
1286    /// minimum peg-in amount) apply equally to newly allocated and reused
1287    /// addresses.
1288    pub async fn allocate_deposit_address_pooled_stateless(
1289        &self,
1290        max_gap_size: usize,
1291    ) -> Result<MaybeNewAddress, DepositAddressError> {
1292        let max_gap_size_u64 = u64::try_from(max_gap_size).unwrap_or(u64::MAX);
1293        let extra_meta_value = serde_json::Value::Null;
1294        let result = self
1295            .db
1296            .autocommit(
1297                move |dbtx, _| {
1298                    let extra_meta_value_inner = extra_meta_value.clone();
1299                    Box::pin(async move {
1300                        let unused = self.unused_pooled_deposit_addresses(dbtx).await;
1301
1302                        if max_gap_size_u64 <= unused.len() as u64 && !unused.is_empty() {
1303                            let addresses = unused
1304                                .into_iter()
1305                                .map(|(tweak_idx, data)| {
1306                                    let (_script, address, _key, operation_id) =
1307                                        self.data.derive_peg_in_script(tweak_idx);
1308
1309                                    debug_assert_eq!(operation_id, data.operation_id);
1310
1311                                    DepositAddressInfo {
1312                                        operation_id,
1313                                        address,
1314                                        tweak_idx,
1315                                    }
1316                                })
1317                                .collect();
1318
1319                            return Ok::<_, DepositAddressError>(
1320                                MaybeNewAddress::TooManyUnusedAddresses(addresses),
1321                            );
1322                        }
1323
1324                        let deposit_address = self.allocate_deposit_address_inner(dbtx).await;
1325
1326                        self.client_ctx
1327                            .manual_operation_start_dbtx(
1328                                dbtx,
1329                                deposit_address.operation_id,
1330                                WalletCommonInit::KIND.as_str(),
1331                                WalletOperationMeta {
1332                                    variant: WalletOperationMetaVariant::Deposit {
1333                                        address: deposit_address.address.clone().into_unchecked(),
1334                                        tweak_idx: Some(deposit_address.tweak_idx),
1335                                        expires_at: None,
1336                                    },
1337                                    extra_meta: extra_meta_value_inner,
1338                                },
1339                                vec![],
1340                            )
1341                            .await?;
1342
1343                        debug!(
1344                            target: LOG_CLIENT_MODULE_WALLET,
1345                            tweak_idx = %deposit_address.tweak_idx,
1346                            address = %deposit_address.address,
1347                            "Derived a new pooled deposit address"
1348                        );
1349
1350                        self.rpc
1351                            .watch_script_history(&deposit_address.address.script_pubkey())
1352                            .await?;
1353
1354                        let sender = self.pegin_monitor_wakeup_sender.clone();
1355                        dbtx.on_commit(move || {
1356                            sender.send_replace(());
1357                        });
1358
1359                        Ok(MaybeNewAddress::NewAddress(deposit_address))
1360                    })
1361                },
1362                Some(100),
1363            )
1364            .await?;
1365
1366        Ok(result)
1367    }
1368
1369    async fn unused_pooled_deposit_addresses(
1370        &self,
1371        dbtx: &mut DatabaseTransaction<'_>,
1372    ) -> Vec<(TweakIdx, PegInTweakIndexData)> {
1373        // Walk peg-in tweaks in descending order, taking unused ones
1374        // (`claimed.is_empty()`) until we hit a used one. That gives us exactly
1375        // the trailing gap — `gap + 1` reads regardless of total tweak count.
1376        // No RPC and no separately-maintained `last_used` index needed.
1377        let mut unused: Vec<(TweakIdx, PegInTweakIndexData)> = dbtx
1378            .find_by_prefix_sorted_descending(&PegInTweakIndexPrefix)
1379            .await
1380            .take_while(|(_, d)| std::future::ready(d.claimed.is_empty()))
1381            .map(|(k, v)| (k.0, v))
1382            .collect()
1383            .await;
1384
1385        // Order by `creation_time` ascending for caller selection and stable
1386        // wraparound after stateful reuse mutates `creation_time`.
1387        unused.sort_by_key(|(t, d)| (d.creation_time, *t));
1388        unused
1389    }
1390
1391    /// Allocate a deposit address, bounding the gap of consecutive unused
1392    /// addresses past the last-used one.
1393    ///
1394    /// This is a stateful wrapper over
1395    /// [`Self::allocate_deposit_address_pooled_stateless`]. If the stateless
1396    /// method returns [`MaybeNewAddress::TooManyUnusedAddresses`], this method
1397    /// picks one of those addresses with a persisted round-robin cursor and
1398    /// returns it with [`AllocateDepositOutcome::Reused`].
1399    ///
1400    /// Selection policy when reusing: round-robin over unused addresses
1401    /// ordered by `creation_time` ascending, with the cursor persisted
1402    /// per-client in the wallet module's DB. The cursor stores the next
1403    /// tweak index to consider; after picking idx X it advances to X+1, and
1404    /// when no candidate has `tweak_idx >= cursor` it wraps to the
1405    /// oldest-by-`creation_time` candidate. This gives the caller the
1406    /// predictable "cycle through the unused addresses" behavior.
1407    ///
1408    /// On reuse the `creation_time` and check-schedule fields of the
1409    /// underlying tweak entry are reset to "now", so `pegin_monitor`'s
1410    /// age-proportional polling delay restarts from zero. Without this an
1411    /// old entry would keep its long stale next-check delay and a user
1412    /// sending to it could wait a long time before the client noticed.
1413    #[allow(clippy::too_many_lines)]
1414    pub async fn allocate_deposit_address_pooled(
1415        &self,
1416        max_gap_size: usize,
1417    ) -> Result<(DepositAddressInfo, AllocateDepositOutcome), DepositAddressError> {
1418        let stateless = self
1419            .allocate_deposit_address_pooled_stateless(max_gap_size)
1420            .await?;
1421
1422        let reused_addresses = match stateless {
1423            MaybeNewAddress::NewAddress(deposit_address) => {
1424                return Ok((deposit_address, AllocateDepositOutcome::Fresh));
1425            }
1426            MaybeNewAddress::TooManyUnusedAddresses(addresses) => addresses,
1427        };
1428
1429        let result = self
1430            .db
1431            .autocommit(
1432                move |dbtx, _| {
1433                    let reused_addresses = reused_addresses.clone();
1434                    Box::pin(async move {
1435                        let cursor = dbtx
1436                            .get_value(&PegInPoolCursorKey)
1437                            .await
1438                            .unwrap_or(TweakIdx(0));
1439
1440                        let pick_pos = reused_addresses
1441                            .iter()
1442                            .position(|a| cursor <= a.tweak_idx)
1443                            .unwrap_or(0);
1444                        let reused_address = reused_addresses[pick_pos].clone();
1445
1446                        let existing_tweak_idx = reused_address.tweak_idx;
1447                        let existing = dbtx
1448                            .get_value(&PegInTweakIndexKey(reused_address.tweak_idx))
1449                            .await
1450                            .ok_or(DepositAddressError::PooledAddressDisappeared {
1451                                tweak_idx: reused_address.tweak_idx,
1452                            })?;
1453
1454                        if !existing.claimed.is_empty() {
1455                            return Err(DepositAddressError::PooledAddressUsed {
1456                                tweak_idx: reused_address.tweak_idx,
1457                            });
1458                        }
1459
1460                        dbtx.insert_entry(&PegInPoolCursorKey, &reused_address.tweak_idx.next())
1461                            .await;
1462
1463                        // Reset the monitoring schedule so the reused address
1464                        // gets checked as aggressively as a freshly-allocated
1465                        // one. Without this, an old entry would retain its
1466                        // `age/10` next-check delay from `pegin_monitor`,
1467                        // meaning a user who sends to a long-idle pool address
1468                        // could wait hours before the client polls for the
1469                        // deposit.
1470                        let now = fedimint_core::time::now();
1471                        dbtx.insert_entry(
1472                            &PegInTweakIndexKey(reused_address.tweak_idx),
1473                            &PegInTweakIndexData {
1474                                creation_time: now,
1475                                last_check_time: None,
1476                                next_check_time: Some(now),
1477                                operation_id: existing.operation_id,
1478                                claimed: existing.claimed,
1479                            },
1480                        )
1481                        .await;
1482
1483                        let sender = self.pegin_monitor_wakeup_sender.clone();
1484                        dbtx.on_commit(move || {
1485                            sender.send_replace(());
1486                        });
1487
1488                        Ok::<_, DepositAddressError>((
1489                            reused_address,
1490                            AllocateDepositOutcome::Reused {
1491                                original_tweak_idx: existing_tweak_idx,
1492                            },
1493                        ))
1494                    })
1495                },
1496                Some(100),
1497            )
1498            .await?;
1499
1500        Ok(result)
1501    }
1502
1503    /// Returns a stream of updates about an ongoing deposit operation created
1504    /// with [`WalletClientModule::allocate_deposit_address_expert_only`].
1505    /// Returns an error for old deposit operations created prior to the 0.4
1506    /// release and not driven to completion yet. This should be rare enough
1507    /// that an indeterminate state is ok here.
1508    pub async fn subscribe_deposit(
1509        &self,
1510        operation_id: OperationId,
1511    ) -> Result<UpdateStreamOrOutcome<DepositStateV2>, SubscribeDepositError> {
1512        let operation = self.client_ctx.get_operation(operation_id).await?;
1513
1514        let operation_meta = operation.meta::<WalletOperationMeta>();
1515
1516        let WalletOperationMetaVariant::Deposit {
1517            address, tweak_idx, ..
1518        } = operation_meta.variant
1519        else {
1520            return Err(SubscribeDepositError::NotADeposit);
1521        };
1522
1523        let network = self.cfg().network.0;
1524        let address = address
1525            .require_network(network)
1526            .map_err(|_| SubscribeDepositError::WrongNetwork { expected: network })?;
1527
1528        // The old deposit operations don't have tweak_idx set
1529        let Some(tweak_idx) = tweak_idx else {
1530            // In case we are dealing with an old deposit that still uses state machines we
1531            // don't have the logic here anymore to subscribe to updates. We can still read
1532            // the final state though if it reached any.
1533            let outcome_v1 = operation
1534                .outcome::<DepositStateV1>()
1535                .ok_or(SubscribeDepositError::OldPendingDeposit)?;
1536
1537            let outcome_v2 = match outcome_v1 {
1538                DepositStateV1::Claimed(tx_info) => DepositStateV2::Claimed {
1539                    btc_deposited: tx_info.btc_transaction.output[tx_info.out_idx as usize].value,
1540                    btc_out_point: bitcoin::OutPoint {
1541                        txid: tx_info.btc_transaction.compute_txid(),
1542                        vout: tx_info.out_idx,
1543                    },
1544                },
1545                DepositStateV1::Failed(error) => DepositStateV2::Failed(error),
1546                _ => return Err(SubscribeDepositError::NonFinalOutcome),
1547            };
1548
1549            return Ok(UpdateStreamOrOutcome::Outcome(outcome_v2));
1550        };
1551
1552        Ok(self.client_ctx.outcome_or_updates(
1553            &operation,
1554            operation_id,
1555            |state| match state {
1556                DepositStateV2::WaitingForTransaction
1557                | DepositStateV2::WaitingForConfirmation { .. }
1558                | DepositStateV2::Confirmed { .. } => false,
1559                DepositStateV2::Claimed { .. } | DepositStateV2::Failed(_) => true,
1560            },
1561            {
1562            let stream_rpc = self.rpc.clone();
1563            let stream_client_ctx = self.client_ctx.clone();
1564            let stream_script_pub_key = address.script_pubkey();
1565            move || {
1566
1567            stream! {
1568                yield DepositStateV2::WaitingForTransaction;
1569
1570                retry(
1571                    "subscribe script history",
1572                    background_backoff(),
1573                    || stream_rpc.watch_script_history(&stream_script_pub_key)
1574                ).await.expect("Will never give up");
1575                let (btc_out_point, btc_deposited) = retry(
1576                    "fetch history",
1577                    background_backoff(),
1578                    || async {
1579                        let history = stream_rpc.get_script_history(&stream_script_pub_key).await?;
1580                        history.first().and_then(|tx| {
1581                            let (out_idx, amount) = tx.output
1582                                .iter()
1583                                .enumerate()
1584                                .find_map(|(idx, output)| (output.script_pubkey == stream_script_pub_key).then_some((idx, output.value)))?;
1585                            let txid = tx.compute_txid();
1586
1587                            Some((
1588                                bitcoin::OutPoint {
1589                                    txid,
1590                                    vout: out_idx as u32,
1591                                },
1592                                amount
1593                            ))
1594                        }).ok_or(FetchDepositTransactionError::NotFound)
1595                    }
1596                ).await.expect("Will never give up");
1597
1598                yield DepositStateV2::WaitingForConfirmation {
1599                    btc_deposited,
1600                    btc_out_point
1601                };
1602
1603                let claim_data = stream_client_ctx.module_db().wait_key_exists(&ClaimedPegInKey {
1604                    peg_in_index: tweak_idx,
1605                    btc_out_point,
1606                }).await;
1607
1608                yield DepositStateV2::Confirmed {
1609                    btc_deposited,
1610                    btc_out_point
1611                };
1612
1613                match stream_client_ctx.await_primary_module_outputs(operation_id, claim_data.change).await {
1614                    Ok(()) => yield DepositStateV2::Claimed {
1615                        btc_deposited,
1616                        btc_out_point
1617                    },
1618                    Err(e) => yield DepositStateV2::Failed(e.fmt_compact().to_string())
1619                }
1620            }
1621        }}))
1622    }
1623
1624    pub async fn list_peg_in_tweak_idxes(&self) -> BTreeMap<TweakIdx, PegInTweakIndexData> {
1625        self.client_ctx
1626            .module_db()
1627            .clone()
1628            .begin_transaction_nc()
1629            .await
1630            .find_by_prefix(&PegInTweakIndexPrefix)
1631            .await
1632            .map(|(key, data)| (key.0, data))
1633            .collect()
1634            .await
1635    }
1636
1637    pub async fn find_tweak_idx_by_address(
1638        &self,
1639        address: bitcoin::Address<NetworkUnchecked>,
1640    ) -> Result<TweakIdx, PegInError> {
1641        let data = self.data.clone();
1642        let Some((tweak_idx, _)) = self
1643            .db
1644            .begin_transaction_nc()
1645            .await
1646            .find_by_prefix(&PegInTweakIndexPrefix)
1647            .await
1648            .filter(|(k, _)| {
1649                let (_, derived_address, _tweak_key, _) = data.derive_peg_in_script(k.0);
1650                future::ready(derived_address.into_unchecked() == address)
1651            })
1652            .next()
1653            .await
1654        else {
1655            return Err(PegInError::AddressNotDerived);
1656        };
1657
1658        Ok(tweak_idx.0)
1659    }
1660    pub async fn find_tweak_idx_by_operation_id(
1661        &self,
1662        operation_id: OperationId,
1663    ) -> Result<TweakIdx, PegInError> {
1664        Ok(self
1665            .client_ctx
1666            .module_db()
1667            .clone()
1668            .begin_transaction_nc()
1669            .await
1670            .find_by_prefix(&PegInTweakIndexPrefix)
1671            .await
1672            .filter(|(_k, v)| future::ready(v.operation_id == operation_id))
1673            .next()
1674            .await
1675            .ok_or(PegInError::NoAddressForOperation { operation_id })?
1676            .0
1677            .0)
1678    }
1679
1680    pub async fn get_pegin_tweak_idx(
1681        &self,
1682        tweak_idx: TweakIdx,
1683    ) -> Result<PegInTweakIndexData, PegInError> {
1684        self.client_ctx
1685            .module_db()
1686            .clone()
1687            .begin_transaction_nc()
1688            .await
1689            .get_value(&PegInTweakIndexKey(tweak_idx))
1690            .await
1691            .ok_or(PegInError::TweakIdxNotFound { tweak_idx })
1692    }
1693
1694    pub async fn get_claimed_pegins(
1695        &self,
1696        dbtx: &mut DatabaseTransaction<'_>,
1697        tweak_idx: TweakIdx,
1698    ) -> Vec<(
1699        bitcoin::OutPoint,
1700        TransactionId,
1701        Vec<fedimint_core::OutPoint>,
1702    )> {
1703        let outpoints = dbtx
1704            .get_value(&PegInTweakIndexKey(tweak_idx))
1705            .await
1706            .map(|v| v.claimed)
1707            .unwrap_or_default();
1708
1709        let mut res = vec![];
1710
1711        for outpoint in outpoints {
1712            let claimed_peg_in_data = dbtx
1713                .get_value(&ClaimedPegInKey {
1714                    peg_in_index: tweak_idx,
1715                    btc_out_point: outpoint,
1716                })
1717                .await
1718                .expect("Must have a corresponding claim record");
1719            res.push((
1720                outpoint,
1721                claimed_peg_in_data.claim_txid,
1722                claimed_peg_in_data.change,
1723            ));
1724        }
1725
1726        res
1727    }
1728
1729    /// Like [`Self::recheck_pegin_address`] but by `operation_id`
1730    pub async fn recheck_pegin_address_by_op_id(
1731        &self,
1732        operation_id: OperationId,
1733    ) -> Result<(), PegInError> {
1734        let tweak_idx = self.find_tweak_idx_by_operation_id(operation_id).await?;
1735
1736        self.recheck_pegin_address(tweak_idx).await
1737    }
1738
1739    /// Schedule given address for immediate re-check for deposits
1740    pub async fn recheck_pegin_address_by_address(
1741        &self,
1742        address: bitcoin::Address<NetworkUnchecked>,
1743    ) -> Result<(), PegInError> {
1744        self.recheck_pegin_address(self.find_tweak_idx_by_address(address).await?)
1745            .await
1746    }
1747
1748    /// Schedule given address for immediate re-check for deposits
1749    pub async fn recheck_pegin_address(&self, tweak_idx: TweakIdx) -> Result<(), PegInError> {
1750        self.db
1751            .autocommit(
1752                |dbtx, _| {
1753                    Box::pin(async {
1754                        let db_key = PegInTweakIndexKey(tweak_idx);
1755                        let db_val = dbtx
1756                            .get_value(&db_key)
1757                            .await
1758                            .ok_or(PegInError::TweakIdxNotFound { tweak_idx })?;
1759
1760                        dbtx.insert_entry(
1761                            &db_key,
1762                            &PegInTweakIndexData {
1763                                next_check_time: Some(fedimint_core::time::now()),
1764                                ..db_val
1765                            },
1766                        )
1767                        .await;
1768
1769                        let sender = self.pegin_monitor_wakeup_sender.clone();
1770                        dbtx.on_commit(move || {
1771                            sender.send_replace(());
1772                        });
1773
1774                        Ok::<_, PegInError>(())
1775                    })
1776                },
1777                Some(100),
1778            )
1779            .await?;
1780
1781        Ok(())
1782    }
1783
1784    /// Await for num deposit by [`OperationId`]
1785    pub async fn await_num_deposits_by_operation_id(
1786        &self,
1787        operation_id: OperationId,
1788        num_deposits: usize,
1789    ) -> Result<(), PegInError> {
1790        let tweak_idx = self.find_tweak_idx_by_operation_id(operation_id).await?;
1791        self.await_num_deposits(tweak_idx, num_deposits).await
1792    }
1793
1794    pub async fn await_num_deposits_by_address(
1795        &self,
1796        address: bitcoin::Address<NetworkUnchecked>,
1797        num_deposits: usize,
1798    ) -> Result<(), PegInError> {
1799        self.await_num_deposits(self.find_tweak_idx_by_address(address).await?, num_deposits)
1800            .await
1801    }
1802
1803    #[instrument(target = LOG_CLIENT_MODULE_WALLET, skip_all, fields(tweak_idx=?tweak_idx, num_deposists=num_deposits))]
1804    pub async fn await_num_deposits(
1805        &self,
1806        tweak_idx: TweakIdx,
1807        num_deposits: usize,
1808    ) -> Result<(), PegInError> {
1809        let operation_id = self.get_pegin_tweak_idx(tweak_idx).await?.operation_id;
1810
1811        let mut receiver = self.pegin_claimed_receiver.clone();
1812        let mut backoff = backoff_util::aggressive_backoff();
1813
1814        loop {
1815            let pegins = self
1816                .get_claimed_pegins(
1817                    &mut self.client_ctx.module_db().begin_transaction_nc().await,
1818                    tweak_idx,
1819                )
1820                .await;
1821
1822            if pegins.len() < num_deposits {
1823                debug!(target: LOG_CLIENT_MODULE_WALLET, has=pegins.len(), "Not enough deposits");
1824                self.recheck_pegin_address(tweak_idx).await?;
1825                runtime::sleep(backoff.next().unwrap_or_default()).await;
1826                receiver
1827                    .changed()
1828                    .await
1829                    .map_err(|_| PegInError::MonitorStopped)?;
1830                continue;
1831            }
1832
1833            debug!(target: LOG_CLIENT_MODULE_WALLET, has=pegins.len(), "Enough deposits detected");
1834
1835            for (_outpoint, transaction_id, change) in pegins {
1836                if transaction_id == TransactionId::from_byte_array([0; 32]) && change.is_empty() {
1837                    debug!(target: LOG_CLIENT_MODULE_WALLET, "Deposited amount was too low, skipping");
1838                    continue;
1839                }
1840
1841                debug!(target: LOG_CLIENT_MODULE_WALLET, out_points=?change, "Ensuring deposists claimed");
1842                let tx_subscriber = self.client_ctx.transaction_updates(operation_id).await;
1843
1844                if let Err(reason) = tx_subscriber.await_tx_accepted(transaction_id).await {
1845                    return Err(PegInError::TransactionRejected { reason });
1846                }
1847
1848                debug!(target: LOG_CLIENT_MODULE_WALLET, out_points=?change, "Ensuring outputs claimed");
1849                self.client_ctx
1850                    .await_primary_module_outputs(operation_id, change)
1851                    .await
1852                    .expect("Cannot fail if tx was accepted and federation is honest");
1853            }
1854
1855            return Ok(());
1856        }
1857    }
1858
1859    /// Attempt to withdraw a given `amount` of Bitcoin to a destination
1860    /// `address`. The caller has to supply the fee rate to be used which can be
1861    /// fetched using [`Self::get_withdraw_fees`] and should be
1862    /// acknowledged by the user since it can be unexpectedly high.
1863    pub async fn withdraw<M: Serialize + MaybeSend + MaybeSync>(
1864        &self,
1865        address: &bitcoin::Address,
1866        amount: bitcoin::Amount,
1867        fee: PegOutFees,
1868        extra_meta: M,
1869    ) -> Result<OperationId, TransactionSubmitError> {
1870        {
1871            let operation_id = OperationId(thread_rng().r#gen());
1872
1873            let withdraw_output =
1874                self.create_withdraw_output(operation_id, address.clone(), amount, fee);
1875            let tx_builder = TransactionBuilder::new()
1876                .with_outputs(self.client_ctx.make_client_outputs(withdraw_output));
1877
1878            let extra_meta =
1879                serde_json::to_value(extra_meta).expect("Failed to serialize extra meta");
1880            self.client_ctx
1881                .finalize_and_submit_transaction(
1882                    operation_id,
1883                    WalletCommonInit::KIND.as_str(),
1884                    {
1885                        let address = address.clone();
1886                        move |change_range: OutPointRange| WalletOperationMeta {
1887                            variant: WalletOperationMetaVariant::Withdraw {
1888                                address: address.clone().into_unchecked(),
1889                                amount,
1890                                fee,
1891                                change: change_range.into_iter().collect(),
1892                            },
1893                            extra_meta: extra_meta.clone(),
1894                        }
1895                    },
1896                    tx_builder,
1897                )
1898                .await?;
1899
1900            let mut dbtx = self.client_ctx.module_db().begin_transaction().await;
1901
1902            self.client_ctx
1903                .log_event(
1904                    &mut dbtx,
1905                    SendPaymentEvent {
1906                        operation_id,
1907                        amount: amount + fee.amount(),
1908                        fee: fee.amount(),
1909                    },
1910                )
1911                .await;
1912
1913            dbtx.commit_tx().await;
1914
1915            Ok(operation_id)
1916        }
1917    }
1918
1919    /// Attempt to increase the fee of a onchain withdraw transaction using
1920    /// replace by fee (RBF).
1921    /// This can prevent transactions from getting stuck
1922    /// in the mempool
1923    #[deprecated(
1924        since = "0.4.0",
1925        note = "RBF withdrawals are rejected by the federation"
1926    )]
1927    pub async fn rbf_withdraw<M: Serialize + MaybeSync + MaybeSend>(
1928        &self,
1929        rbf: Rbf,
1930        extra_meta: M,
1931    ) -> Result<OperationId, TransactionSubmitError> {
1932        let operation_id = OperationId(thread_rng().r#gen());
1933
1934        let withdraw_output = self.create_rbf_withdraw_output(operation_id, &rbf);
1935        let tx_builder = TransactionBuilder::new()
1936            .with_outputs(self.client_ctx.make_client_outputs(withdraw_output));
1937
1938        let extra_meta = serde_json::to_value(extra_meta).expect("Failed to serialize extra meta");
1939        self.client_ctx
1940            .finalize_and_submit_transaction(
1941                operation_id,
1942                WalletCommonInit::KIND.as_str(),
1943                move |change_range: OutPointRange| WalletOperationMeta {
1944                    variant: WalletOperationMetaVariant::RbfWithdraw {
1945                        rbf: rbf.clone(),
1946                        change: change_range.into_iter().collect(),
1947                    },
1948                    extra_meta: extra_meta.clone(),
1949                },
1950                tx_builder,
1951            )
1952            .await?;
1953
1954        Ok(operation_id)
1955    }
1956
1957    pub async fn subscribe_withdraw_updates(
1958        &self,
1959        operation_id: OperationId,
1960    ) -> Result<UpdateStreamOrOutcome<WithdrawState>, SubscribeWithdrawError> {
1961        let operation = self.client_ctx.get_operation(operation_id).await?;
1962
1963        let operation_meta = operation.meta::<WalletOperationMeta>();
1964
1965        let (WalletOperationMetaVariant::Withdraw { change, .. }
1966        | WalletOperationMetaVariant::RbfWithdraw { change, .. }) = operation_meta.variant
1967        else {
1968            return Err(SubscribeWithdrawError::NotAWithdrawal);
1969        };
1970
1971        let mut operation_stream = self.notifier.subscribe(operation_id).await;
1972        let client_ctx = self.client_ctx.clone();
1973
1974        Ok(self.client_ctx.outcome_or_updates(
1975            &operation,
1976            operation_id,
1977            |state| match state {
1978                WithdrawState::Created => false,
1979                WithdrawState::Succeeded(_) | WithdrawState::Failed(_) => true,
1980            },
1981            move || {
1982                stream! {
1983                    match next_withdraw_state(&mut operation_stream).await {
1984                        Some(WithdrawStates::Created(_)) => {
1985                            yield WithdrawState::Created;
1986                        },
1987                        Some(s) => {
1988                            panic!("Unexpected state {s:?}")
1989                        },
1990                        None => return,
1991                    }
1992
1993                    // TODO: get rid of awaiting change here, there has to be a better way to make tests deterministic
1994
1995                        // Swallowing potential errors since the transaction failing  is handled by
1996                        // output outcome fetching already
1997                        let _ = client_ctx
1998                            .await_primary_module_outputs(operation_id, change)
1999                            .await;
2000
2001
2002                    match next_withdraw_state(&mut operation_stream).await {
2003                        Some(WithdrawStates::Aborted(inner)) => {
2004                            yield WithdrawState::Failed(inner.error);
2005                        },
2006                        Some(WithdrawStates::Success(inner)) => {
2007                            yield WithdrawState::Succeeded(inner.txid);
2008                        },
2009                        Some(s) => {
2010                            panic!("Unexpected state {s:?}")
2011                        },
2012                        None => {},
2013                    }
2014                }
2015            },
2016        ))
2017    }
2018
2019    fn admin_auth(&self) -> Result<ApiAuth, ConsensusVersionVotingError> {
2020        self.admin_auth
2021            .clone()
2022            .ok_or(ConsensusVersionVotingError::AdminAuthMissing)
2023    }
2024
2025    pub async fn activate_consensus_version_voting(
2026        &self,
2027    ) -> Result<(), ConsensusVersionVotingError> {
2028        self.module_api
2029            .activate_consensus_version_voting(self.admin_auth()?)
2030            .await?;
2031
2032        Ok(())
2033    }
2034}
2035
2036/// Why a deposit address's funding transaction was not found, so that it is
2037/// looked up again.
2038#[derive(Debug, thiserror::Error)]
2039enum FetchDepositTransactionError {
2040    /// The Bitcoin backend could not report the address's history.
2041    #[error(transparent)]
2042    BitcoinRpc(#[from] BitcoinRpcError),
2043
2044    /// The address's history holds no transaction paying to it yet.
2045    #[error("No deposit transaction found")]
2046    NotFound,
2047}
2048
2049/// Polls the federation checking if the activated module consensus version
2050/// supports safe deposits, saving the result in the db once it does.
2051async fn poll_supports_safe_deposit_version(db: Database, module_api: DynModuleApi) {
2052    loop {
2053        if supports_safe_deposit_verified(&db).await {
2054            break;
2055        }
2056
2057        module_api.wait_for_initialized_connections().await;
2058
2059        if verify_supports_safe_deposit(&db, &module_api).await == Some(true) {
2060            break;
2061        }
2062
2063        if is_running_in_test_env() {
2064            // Even in tests we don't want to spam the federation with requests about it
2065            sleep(Duration::from_secs(10)).await;
2066        } else {
2067            sleep(Duration::from_hours(1)).await;
2068        }
2069    }
2070}
2071
2072/// Whether the federation's wallet module consensus version was already
2073/// verified to support safe deposits.
2074async fn supports_safe_deposit_verified(db: &Database) -> bool {
2075    db.begin_transaction_nc()
2076        .await
2077        .get_value(&SupportsSafeDepositKey)
2078        .await
2079        .is_some()
2080}
2081
2082/// Fetches the federation's wallet module consensus version and records the
2083/// marker if it supports safe deposits.
2084///
2085/// Returns `None` if the version could not be fetched, e.g. while offline.
2086async fn verify_supports_safe_deposit(db: &Database, module_api: &DynModuleApi) -> Option<bool> {
2087    let module_consensus_version = match module_api.module_consensus_version().await {
2088        Ok(module_consensus_version) => module_consensus_version,
2089        Err(err) => {
2090            debug!(
2091                target: LOG_CLIENT_MODULE_WALLET,
2092                err = %err.fmt_compact(),
2093                "Could not fetch the wallet module consensus version"
2094            );
2095            return None;
2096        }
2097    };
2098
2099    let supported_version = SAFE_DEPOSIT_MODULE_CONSENSUS_VERSION <= module_consensus_version;
2100
2101    if supported_version {
2102        store_supports_safe_deposit(db.begin_transaction().await).await;
2103    }
2104
2105    Some(supported_version)
2106}
2107
2108/// Records in `dbtx` that the federation's wallet module consensus version
2109/// was verified to support safe deposits, and commits it.
2110///
2111/// The marker is written by both the background version poller and
2112/// [`WalletClientModule::supports_safe_deposit`], which race right after
2113/// joining a federation. The loser's commit fails with a write conflict, but
2114/// the winner stored the very same marker and nothing ever removes it, so the
2115/// conflict is treated as success. That is only sound for a fresh `dbtx` whose
2116/// sole write is the marker: any other write in it would be lost as well.
2117///
2118/// Any other commit error is logged rather than escalated: the marker only
2119/// lets future checks skip re-verifying the version, callers already have
2120/// their answer, and the next verification simply writes it again — an
2121/// optional write is not worth crashing over.
2122async fn store_supports_safe_deposit(mut dbtx: DatabaseTransaction<'_, Committable>) {
2123    if dbtx.get_value(&SupportsSafeDepositKey).await.is_some() {
2124        // The other writer won before this transaction started.
2125        return;
2126    }
2127
2128    dbtx.insert_entry(&SupportsSafeDepositKey, &()).await;
2129
2130    match dbtx.commit_tx_result().await {
2131        Ok(()) | Err(DatabaseError::WriteConflict) => {}
2132        Err(err) => {
2133            warn!(
2134                target: LOG_CLIENT_MODULE_WALLET,
2135                err = %err.fmt_compact(),
2136                "Failed to store the safe-deposit marker"
2137            );
2138        }
2139    }
2140}
2141
2142/// Returns the child index to derive the next peg-in tweak key from.
2143async fn get_next_peg_in_tweak_child_id(dbtx: &mut DatabaseTransaction<'_>) -> TweakIdx {
2144    let index = dbtx
2145        .get_value(&NextPegInTweakIndexKey)
2146        .await
2147        .unwrap_or_default();
2148    dbtx.insert_entry(&NextPegInTweakIndexKey, &(index.next()))
2149        .await;
2150    index
2151}
2152
2153#[derive(Debug, Clone, Eq, PartialEq, Hash, Decodable, Encodable)]
2154pub enum WalletClientStates {
2155    Deposit(DepositStateMachine),
2156    Withdraw(WithdrawStateMachine),
2157}
2158
2159impl IntoDynInstance for WalletClientStates {
2160    type DynType = DynState;
2161
2162    fn into_dyn(self, instance_id: ModuleInstanceId) -> Self::DynType {
2163        DynState::from_typed(instance_id, self)
2164    }
2165}
2166
2167impl State for WalletClientStates {
2168    type ModuleContext = WalletClientContext;
2169
2170    fn transitions(
2171        &self,
2172        context: &Self::ModuleContext,
2173        global_context: &DynGlobalClientContext,
2174    ) -> Vec<StateTransition<Self>> {
2175        match self {
2176            WalletClientStates::Deposit(sm) => {
2177                sm_enum_variant_translation!(
2178                    sm.transitions(context, global_context),
2179                    WalletClientStates::Deposit
2180                )
2181            }
2182            WalletClientStates::Withdraw(sm) => {
2183                sm_enum_variant_translation!(
2184                    sm.transitions(context, global_context),
2185                    WalletClientStates::Withdraw
2186                )
2187            }
2188        }
2189    }
2190
2191    fn operation_id(&self) -> OperationId {
2192        match self {
2193            WalletClientStates::Deposit(sm) => sm.operation_id(),
2194            WalletClientStates::Withdraw(sm) => sm.operation_id(),
2195        }
2196    }
2197}
2198
2199#[cfg(all(test, not(target_family = "wasm")))]
2200mod tests;