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;
18pub mod events;
19/// Peg-in monitor: a task monitoring deposit addresses for peg-ins.
20mod pegin_monitor;
21mod withdraw;
22
23use std::collections::{BTreeMap, BTreeSet};
24use std::future;
25use std::sync::Arc;
26use std::time::{Duration, SystemTime};
27
28use anyhow::{Context as AnyhowContext, anyhow, bail, ensure};
29use async_stream::{stream, try_stream};
30use backup::WalletModuleBackup;
31use bitcoin::address::NetworkUnchecked;
32use bitcoin::secp256k1::{All, SECP256K1, Secp256k1};
33use bitcoin::{Address, Network, ScriptBuf};
34use client_db::{DbKeyPrefix, PegInTweakIndexKey, SupportsSafeDepositKey, TweakIdx};
35use fedimint_api_client::api::{DynModuleApi, FederationResult};
36use fedimint_bitcoind::{DynBitcoindRpc, create_esplora_rpc};
37use fedimint_client_module::module::init::{
38    ClientModuleInit, ClientModuleInitArgs, ClientModuleRecoverArgs,
39};
40use fedimint_client_module::module::{ClientContext, ClientModule, IClientModule, OutPointRange};
41use fedimint_client_module::oplog::UpdateStreamOrOutcome;
42use fedimint_client_module::sm::{Context, DynState, ModuleNotifier, State, StateTransition};
43use fedimint_client_module::transaction::{
44    ClientOutput, ClientOutputBundle, ClientOutputSM, TransactionBuilder,
45};
46use fedimint_client_module::{DynGlobalClientContext, sm_enum_variant_translation};
47use fedimint_core::core::{Decoder, IntoDynInstance, ModuleInstanceId, ModuleKind, OperationId};
48use fedimint_core::db::{
49    AutocommitError, Database, DatabaseTransaction, IDatabaseTransactionOpsCoreTyped,
50};
51use fedimint_core::encoding::{Decodable, Encodable};
52use fedimint_core::envs::{BitcoinRpcConfig, is_running_in_test_env};
53use fedimint_core::module::{
54    ApiAuth, ApiVersion, CommonModuleInit, ModuleCommon, ModuleConsensusVersion, ModuleInit,
55    MultiApiVersion,
56};
57use fedimint_core::task::{MaybeSend, MaybeSync, TaskGroup, sleep};
58use fedimint_core::util::backoff_util::background_backoff;
59use fedimint_core::util::{BoxStream, backoff_util, retry};
60use fedimint_core::{
61    Amount, BitcoinHash, OutPoint, TransactionId, apply, async_trait_maybe_send,
62    push_db_pair_items, runtime, secp256k1,
63};
64use fedimint_derive_secret::{ChildId, DerivableSecret};
65use fedimint_logging::LOG_CLIENT_MODULE_WALLET;
66use fedimint_wallet_common::config::{FeeConsensus, WalletClientConfig};
67use fedimint_wallet_common::tweakable::Tweakable;
68pub use fedimint_wallet_common::*;
69use futures::{Stream, StreamExt};
70use rand::{Rng, thread_rng};
71use secp256k1::Keypair;
72use serde::{Deserialize, Serialize};
73use strum::IntoEnumIterator;
74use tokio::sync::watch;
75use tracing::{debug, instrument};
76
77use crate::api::WalletFederationApi;
78use crate::backup::WalletRecovery;
79use crate::client_db::{
80    ClaimedPegInData, ClaimedPegInKey, ClaimedPegInPrefix, NextPegInTweakIndexKey,
81    PegInTweakIndexData, PegInTweakIndexPrefix, RecoveryFinalizedKey, SupportsSafeDepositPrefix,
82};
83use crate::deposit::DepositStateMachine;
84use crate::withdraw::{CreatedWithdrawState, WithdrawStateMachine, WithdrawStates};
85
86const WALLET_TWEAK_CHILD_ID: ChildId = ChildId(0);
87
88#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
89pub struct BitcoinTransactionData {
90    /// The bitcoin transaction is saved as soon as we see it so the transaction
91    /// can be re-transmitted if it's evicted from the mempool.
92    pub btc_transaction: bitcoin::Transaction,
93    /// Index of the deposit output
94    pub out_idx: u32,
95}
96
97#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
98pub enum DepositStateV1 {
99    WaitingForTransaction,
100    WaitingForConfirmation(BitcoinTransactionData),
101    Confirmed(BitcoinTransactionData),
102    Claimed(BitcoinTransactionData),
103    Failed(String),
104}
105
106#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
107pub enum DepositStateV2 {
108    WaitingForTransaction,
109    WaitingForConfirmation {
110        #[serde(with = "bitcoin::amount::serde::as_sat")]
111        btc_deposited: bitcoin::Amount,
112        btc_out_point: bitcoin::OutPoint,
113    },
114    Confirmed {
115        #[serde(with = "bitcoin::amount::serde::as_sat")]
116        btc_deposited: bitcoin::Amount,
117        btc_out_point: bitcoin::OutPoint,
118    },
119    Claimed {
120        #[serde(with = "bitcoin::amount::serde::as_sat")]
121        btc_deposited: bitcoin::Amount,
122        btc_out_point: bitcoin::OutPoint,
123    },
124    Failed(String),
125}
126
127#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
128pub enum WithdrawState {
129    Created,
130    Succeeded(bitcoin::Txid),
131    Failed(String),
132    // TODO: track refund
133    // Refunded,
134    // RefundFailed(String),
135}
136
137async fn next_withdraw_state<S>(stream: &mut S) -> Option<WithdrawStates>
138where
139    S: Stream<Item = WalletClientStates> + Unpin,
140{
141    loop {
142        if let WalletClientStates::Withdraw(ds) = stream.next().await? {
143            return Some(ds.state);
144        }
145        tokio::task::yield_now().await;
146    }
147}
148
149#[derive(Debug, Clone, Default)]
150// TODO: should probably move to DB
151pub struct WalletClientInit(pub Option<DynBitcoindRpc>);
152
153impl WalletClientInit {
154    pub fn new(rpc: DynBitcoindRpc) -> Self {
155        Self(Some(rpc))
156    }
157}
158
159impl ModuleInit for WalletClientInit {
160    type Common = WalletCommonInit;
161
162    async fn dump_database(
163        &self,
164        dbtx: &mut DatabaseTransaction<'_>,
165        prefix_names: Vec<String>,
166    ) -> Box<dyn Iterator<Item = (String, Box<dyn erased_serde::Serialize + Send>)> + '_> {
167        let mut wallet_client_items: BTreeMap<String, Box<dyn erased_serde::Serialize + Send>> =
168            BTreeMap::new();
169        let filtered_prefixes = DbKeyPrefix::iter().filter(|f| {
170            prefix_names.is_empty() || prefix_names.contains(&f.to_string().to_lowercase())
171        });
172
173        for table in filtered_prefixes {
174            match table {
175                DbKeyPrefix::NextPegInTweakIndex => {
176                    if let Some(index) = dbtx.get_value(&NextPegInTweakIndexKey).await {
177                        wallet_client_items
178                            .insert("NextPegInTweakIndex".to_string(), Box::new(index));
179                    }
180                }
181                DbKeyPrefix::PegInTweakIndex => {
182                    push_db_pair_items!(
183                        dbtx,
184                        PegInTweakIndexPrefix,
185                        PegInTweakIndexKey,
186                        PegInTweakIndexData,
187                        wallet_client_items,
188                        "Peg-In Tweak Index"
189                    );
190                }
191                DbKeyPrefix::ClaimedPegIn => {
192                    push_db_pair_items!(
193                        dbtx,
194                        ClaimedPegInPrefix,
195                        ClaimedPegInKey,
196                        ClaimedPegInData,
197                        wallet_client_items,
198                        "Claimed Peg-In"
199                    );
200                }
201                DbKeyPrefix::RecoveryFinalized => {
202                    if let Some(val) = dbtx.get_value(&RecoveryFinalizedKey).await {
203                        wallet_client_items.insert("RecoveryFinalized".to_string(), Box::new(val));
204                    }
205                }
206                DbKeyPrefix::SupportsSafeDeposit => {
207                    push_db_pair_items!(
208                        dbtx,
209                        SupportsSafeDepositPrefix,
210                        SupportsSafeDepositKey,
211                        (),
212                        wallet_client_items,
213                        "Supports Safe Deposit"
214                    );
215                }
216                DbKeyPrefix::RecoveryState
217                | DbKeyPrefix::ExternalReservedStart
218                | DbKeyPrefix::CoreInternalReservedStart
219                | DbKeyPrefix::CoreInternalReservedEnd => {}
220            }
221        }
222
223        Box::new(wallet_client_items.into_iter())
224    }
225}
226
227#[apply(async_trait_maybe_send!)]
228impl ClientModuleInit for WalletClientInit {
229    type Module = WalletClientModule;
230
231    fn supported_api_versions(&self) -> MultiApiVersion {
232        MultiApiVersion::try_from_iter([ApiVersion { major: 0, minor: 0 }])
233            .expect("no version conflicts")
234    }
235
236    async fn init(&self, args: &ClientModuleInitArgs<Self>) -> anyhow::Result<Self::Module> {
237        let data = WalletClientModuleData {
238            cfg: args.cfg().clone(),
239            module_root_secret: args.module_root_secret().clone(),
240        };
241
242        let db = args.db().clone();
243
244        let btc_rpc = self.0.clone().unwrap_or(create_esplora_rpc(
245            &WalletClientModule::get_rpc_config(args.cfg()).url,
246        )?);
247
248        let module_api = args.module_api().clone();
249
250        let (pegin_claimed_sender, pegin_claimed_receiver) = watch::channel(());
251        let (pegin_monitor_wakeup_sender, pegin_monitor_wakeup_receiver) = watch::channel(());
252
253        Ok(WalletClientModule {
254            db,
255            data,
256            module_api,
257            notifier: args.notifier().clone(),
258            rpc: btc_rpc,
259            client_ctx: args.context(),
260            pegin_monitor_wakeup_sender,
261            pegin_monitor_wakeup_receiver,
262            pegin_claimed_receiver,
263            pegin_claimed_sender,
264            task_group: args.task_group().clone(),
265            admin_auth: args.admin_auth().cloned(),
266        })
267    }
268
269    /// Wallet recovery
270    ///
271    /// Query bitcoin rpc for history of addresses from last known used
272    /// addresses (or index 0) until `MAX_GAP` unused ones.
273    ///
274    /// Notably does not persist the progress of addresses being queried,
275    /// because it is not expected that it would take long enough to bother.
276    async fn recover(
277        &self,
278        args: &ClientModuleRecoverArgs<Self>,
279        snapshot: Option<&<Self::Module as ClientModule>::Backup>,
280    ) -> anyhow::Result<()> {
281        args.recover_from_history::<WalletRecovery>(self, snapshot)
282            .await
283    }
284
285    fn used_db_prefixes(&self) -> Option<BTreeSet<u8>> {
286        Some(
287            DbKeyPrefix::iter()
288                .map(|p| p as u8)
289                .chain(
290                    DbKeyPrefix::ExternalReservedStart as u8
291                        ..=DbKeyPrefix::CoreInternalReservedEnd as u8,
292                )
293                .collect(),
294        )
295    }
296}
297
298#[derive(Debug, Clone, Serialize, Deserialize)]
299pub struct WalletOperationMeta {
300    pub variant: WalletOperationMetaVariant,
301    pub extra_meta: serde_json::Value,
302}
303
304#[derive(Debug, Clone, Serialize, Deserialize)]
305#[serde(rename_all = "snake_case")]
306pub enum WalletOperationMetaVariant {
307    Deposit {
308        address: Address<NetworkUnchecked>,
309        /// Added in 0.4.2, can be `None` for old deposits or `Some` for ones
310        /// using the pegin monitor. The value is the child index of the key
311        /// used to generate the address, so we can re-generate the secret key
312        /// from our root secret.
313        #[serde(default)]
314        tweak_idx: Option<TweakIdx>,
315        #[serde(default, skip_serializing_if = "Option::is_none")]
316        expires_at: Option<SystemTime>,
317    },
318    Withdraw {
319        address: Address<NetworkUnchecked>,
320        #[serde(with = "bitcoin::amount::serde::as_sat")]
321        amount: bitcoin::Amount,
322        fee: PegOutFees,
323        change: Vec<OutPoint>,
324    },
325
326    RbfWithdraw {
327        rbf: Rbf,
328        change: Vec<OutPoint>,
329    },
330}
331
332/// The non-resource, just plain-data parts of [`WalletClientModule`]
333#[derive(Debug, Clone)]
334pub struct WalletClientModuleData {
335    cfg: WalletClientConfig,
336    module_root_secret: DerivableSecret,
337}
338
339impl WalletClientModuleData {
340    fn derive_deposit_address(
341        &self,
342        idx: TweakIdx,
343    ) -> (Keypair, secp256k1::PublicKey, Address, OperationId) {
344        let idx = ChildId(idx.0);
345
346        let secret_tweak_key = self
347            .module_root_secret
348            .child_key(WALLET_TWEAK_CHILD_ID)
349            .child_key(idx)
350            .to_secp_key(fedimint_core::secp256k1::SECP256K1);
351
352        let public_tweak_key = secret_tweak_key.public_key();
353
354        let address = self
355            .cfg
356            .peg_in_descriptor
357            .tweak(&public_tweak_key, bitcoin::secp256k1::SECP256K1)
358            .address(self.cfg.network.0)
359            .unwrap();
360
361        // TODO: make hash?
362        let operation_id = OperationId(public_tweak_key.x_only_public_key().0.serialize());
363
364        (secret_tweak_key, public_tweak_key, address, operation_id)
365    }
366
367    fn derive_peg_in_script(
368        &self,
369        idx: TweakIdx,
370    ) -> (ScriptBuf, bitcoin::Address, Keypair, OperationId) {
371        let (secret_tweak_key, _, address, operation_id) = self.derive_deposit_address(idx);
372
373        (
374            self.cfg
375                .peg_in_descriptor
376                .tweak(&secret_tweak_key.public_key(), SECP256K1)
377                .script_pubkey(),
378            address,
379            secret_tweak_key,
380            operation_id,
381        )
382    }
383}
384
385#[derive(Debug)]
386pub struct WalletClientModule {
387    data: WalletClientModuleData,
388    db: Database,
389    module_api: DynModuleApi,
390    notifier: ModuleNotifier<WalletClientStates>,
391    rpc: DynBitcoindRpc,
392    client_ctx: ClientContext<Self>,
393    /// Updated to wake up pegin monitor
394    pegin_monitor_wakeup_sender: watch::Sender<()>,
395    pegin_monitor_wakeup_receiver: watch::Receiver<()>,
396    /// Called every time a peg-in was claimed
397    pegin_claimed_sender: watch::Sender<()>,
398    pegin_claimed_receiver: watch::Receiver<()>,
399    task_group: TaskGroup,
400    admin_auth: Option<ApiAuth>,
401}
402
403#[apply(async_trait_maybe_send!)]
404impl ClientModule for WalletClientModule {
405    type Init = WalletClientInit;
406    type Common = WalletModuleTypes;
407    type Backup = WalletModuleBackup;
408    type ModuleStateMachineContext = WalletClientContext;
409    type States = WalletClientStates;
410
411    fn context(&self) -> Self::ModuleStateMachineContext {
412        WalletClientContext {
413            rpc: self.rpc.clone(),
414            wallet_descriptor: self.cfg().peg_in_descriptor.clone(),
415            wallet_decoder: self.decoder(),
416            secp: Secp256k1::default(),
417            client_ctx: self.client_ctx.clone(),
418        }
419    }
420
421    async fn start(&self) {
422        self.task_group.spawn_cancellable("peg-in monitor", {
423            let client_ctx = self.client_ctx.clone();
424            let db = self.db.clone();
425            let btc_rpc = self.rpc.clone();
426            let module_api = self.module_api.clone();
427            let data = self.data.clone();
428            let pegin_claimed_sender = self.pegin_claimed_sender.clone();
429            let pegin_monitor_wakeup_receiver = self.pegin_monitor_wakeup_receiver.clone();
430            pegin_monitor::run_peg_in_monitor(
431                client_ctx,
432                db,
433                btc_rpc,
434                module_api,
435                data,
436                pegin_claimed_sender,
437                pegin_monitor_wakeup_receiver,
438            )
439        });
440
441        self.task_group
442            .spawn_cancellable("supports-safe-deposit-version", {
443                let db = self.db.clone();
444                let module_api = self.module_api.clone();
445
446                poll_supports_safe_deposit_version(db, module_api)
447            });
448    }
449
450    fn supports_backup(&self) -> bool {
451        true
452    }
453
454    async fn backup(&self) -> anyhow::Result<backup::WalletModuleBackup> {
455        // fetch consensus height first
456        let session_count = self.client_ctx.global_api().session_count().await?;
457
458        let mut dbtx = self.db.begin_transaction_nc().await;
459        let next_pegin_tweak_idx = dbtx
460            .get_value(&NextPegInTweakIndexKey)
461            .await
462            .unwrap_or_default();
463        let claimed = dbtx
464            .find_by_prefix(&PegInTweakIndexPrefix)
465            .await
466            .filter_map(|(k, v)| async move {
467                if v.claimed.is_empty() {
468                    None
469                } else {
470                    Some(k.0)
471                }
472            })
473            .collect()
474            .await;
475        Ok(backup::WalletModuleBackup::new_v1(
476            session_count,
477            next_pegin_tweak_idx,
478            claimed,
479        ))
480    }
481
482    fn input_fee(
483        &self,
484        _amount: Amount,
485        _input: &<Self::Common as ModuleCommon>::Input,
486    ) -> Option<Amount> {
487        Some(self.cfg().fee_consensus.peg_in_abs)
488    }
489
490    fn output_fee(
491        &self,
492        _amount: Amount,
493        _output: &<Self::Common as ModuleCommon>::Output,
494    ) -> Option<Amount> {
495        Some(self.cfg().fee_consensus.peg_out_abs)
496    }
497
498    async fn handle_rpc(
499        &self,
500        method: String,
501        request: serde_json::Value,
502    ) -> BoxStream<'_, anyhow::Result<serde_json::Value>> {
503        Box::pin(try_stream! {
504            match method.as_str() {
505                "get_wallet_summary" => {
506                    let _req: WalletSummaryRequest = serde_json::from_value(request)?;
507                    let wallet_summary = self.get_wallet_summary()
508                        .await
509                        .expect("Failed to fetch wallet summary");
510                    let result = serde_json::to_value(&wallet_summary)
511                        .expect("Serialization error");
512                    yield result;
513                }
514                "get_block_count_local" => {
515                    let block_count = self.get_block_count_local().await
516                        .expect("Failed to fetch block count");
517                    yield serde_json::to_value(block_count)?;
518                }
519                "peg_in" => {
520                    let req: PegInRequest = serde_json::from_value(request)?;
521                    let response = self.peg_in(req)
522                        .await
523                        .map_err(|e| anyhow::anyhow!("peg_in failed: {}", e))?;
524                    let result = serde_json::to_value(&response)?;
525                    yield result;
526                },
527                "peg_out" => {
528                    let req: PegOutRequest = serde_json::from_value(request)?;
529                    let response = self.peg_out(req)
530                        .await
531                        .map_err(|e| anyhow::anyhow!("peg_out failed: {}", e))?;
532                    let result = serde_json::to_value(&response)?;
533                    yield result;
534                },
535                "subscribe_deposit" => {
536                    let req: SubscribeDepositRequest = serde_json::from_value(request)?;
537                    for await state in self.subscribe_deposit(req.operation_id).await?.into_stream() {
538                        yield serde_json::to_value(state)?;
539                    }
540                },
541                "subscribe_withdraw" => {
542                    let req: SubscribeWithdrawRequest = serde_json::from_value(request)?;
543                    for await state in self.subscribe_withdraw_updates(req.operation_id).await?.into_stream(){
544                        yield serde_json::to_value(state)?;
545                    }
546                }
547                _ => {
548                    Err(anyhow::format_err!("Unknown method: {}", method))?;
549                }
550            }
551        })
552    }
553
554    #[cfg(feature = "cli")]
555    async fn handle_cli_command(
556        &self,
557        args: &[std::ffi::OsString],
558    ) -> anyhow::Result<serde_json::Value> {
559        cli::handle_cli_command(self, args).await
560    }
561}
562
563#[derive(Deserialize)]
564struct WalletSummaryRequest {}
565
566#[derive(Debug, Clone)]
567pub struct WalletClientContext {
568    rpc: DynBitcoindRpc,
569    wallet_descriptor: PegInDescriptor,
570    wallet_decoder: Decoder,
571    secp: Secp256k1<All>,
572    pub client_ctx: ClientContext<WalletClientModule>,
573}
574
575#[derive(Debug, Clone, Serialize, Deserialize)]
576pub struct PegInRequest {
577    pub extra_meta: serde_json::Value,
578}
579
580#[derive(Deserialize)]
581struct SubscribeDepositRequest {
582    operation_id: OperationId,
583}
584
585#[derive(Deserialize)]
586struct SubscribeWithdrawRequest {
587    operation_id: OperationId,
588}
589
590#[derive(Debug, Clone, Serialize, Deserialize)]
591pub struct PegInResponse {
592    pub deposit_address: Address<NetworkUnchecked>,
593    pub operation_id: OperationId,
594}
595
596#[derive(Debug, Clone, Serialize, Deserialize)]
597pub struct PegOutRequest {
598    pub amount_sat: u64,
599    pub destination_address: Address<NetworkUnchecked>,
600    pub extra_meta: serde_json::Value,
601}
602
603#[derive(Debug, Clone, Serialize, Deserialize)]
604pub struct PegOutResponse {
605    pub operation_id: OperationId,
606}
607
608impl Context for WalletClientContext {
609    const KIND: Option<ModuleKind> = Some(KIND);
610}
611
612impl WalletClientModule {
613    fn cfg(&self) -> &WalletClientConfig {
614        &self.data.cfg
615    }
616
617    fn get_rpc_config(cfg: &WalletClientConfig) -> BitcoinRpcConfig {
618        match BitcoinRpcConfig::get_defaults_from_env_vars() {
619            Ok(rpc_config) => {
620                // TODO: Wallet client cannot support bitcoind RPC until the bitcoin dep is
621                // updated to 0.30
622                if rpc_config.kind == "bitcoind" {
623                    cfg.default_bitcoin_rpc.clone()
624                } else {
625                    rpc_config
626                }
627            }
628            _ => cfg.default_bitcoin_rpc.clone(),
629        }
630    }
631
632    pub fn get_network(&self) -> Network {
633        self.cfg().network.0
634    }
635
636    pub fn get_finality_delay(&self) -> u32 {
637        self.cfg().finality_delay
638    }
639
640    pub fn get_fee_consensus(&self) -> FeeConsensus {
641        self.cfg().fee_consensus
642    }
643
644    async fn allocate_deposit_address_inner(
645        &self,
646        dbtx: &mut DatabaseTransaction<'_>,
647    ) -> (OperationId, Address, TweakIdx) {
648        dbtx.ensure_isolated().expect("Must be isolated db");
649
650        let tweak_idx = get_next_peg_in_tweak_child_id(dbtx).await;
651        let (_secret_tweak_key, _, address, operation_id) =
652            self.data.derive_deposit_address(tweak_idx);
653
654        let now = fedimint_core::time::now();
655
656        dbtx.insert_new_entry(
657            &PegInTweakIndexKey(tweak_idx),
658            &PegInTweakIndexData {
659                creation_time: now,
660                next_check_time: Some(now),
661                last_check_time: None,
662                operation_id,
663                claimed: vec![],
664            },
665        )
666        .await;
667
668        (operation_id, address, tweak_idx)
669    }
670
671    /// Fetches the fees that would need to be paid to make the withdraw request
672    /// using [`Self::withdraw`] work *right now*.
673    ///
674    /// Note that we do not receive a guarantee that these fees will be valid in
675    /// the future, thus even the next second using these fees *may* fail.
676    /// The caller should be prepared to retry with a new fee estimate.
677    pub async fn get_withdraw_fees(
678        &self,
679        address: &bitcoin::Address,
680        amount: bitcoin::Amount,
681    ) -> anyhow::Result<PegOutFees> {
682        self.module_api
683            .fetch_peg_out_fees(address, amount)
684            .await?
685            .context("Federation didn't return peg-out fees")
686    }
687
688    /// Returns a summary of the wallet's coins
689    pub async fn get_wallet_summary(&self) -> anyhow::Result<WalletSummary> {
690        Ok(self.module_api.fetch_wallet_summary().await?)
691    }
692
693    pub async fn get_block_count_local(&self) -> anyhow::Result<u32> {
694        Ok(self.module_api.fetch_block_count_local().await?)
695    }
696
697    pub fn create_withdraw_output(
698        &self,
699        operation_id: OperationId,
700        address: bitcoin::Address,
701        amount: bitcoin::Amount,
702        fees: PegOutFees,
703    ) -> anyhow::Result<ClientOutputBundle<WalletOutput, WalletClientStates>> {
704        let output = WalletOutput::new_v0_peg_out(address, amount, fees);
705
706        let amount = output.maybe_v0_ref().expect("v0 output").amount().into();
707
708        let sm_gen = move |out_point_range: OutPointRange| {
709            assert_eq!(out_point_range.count(), 1);
710            let out_idx = out_point_range.start_idx();
711            vec![WalletClientStates::Withdraw(WithdrawStateMachine {
712                operation_id,
713                state: WithdrawStates::Created(CreatedWithdrawState {
714                    fm_outpoint: OutPoint {
715                        txid: out_point_range.txid(),
716                        out_idx,
717                    },
718                }),
719            })]
720        };
721
722        Ok(ClientOutputBundle::new(
723            vec![ClientOutput::<WalletOutput> { output, amount }],
724            vec![ClientOutputSM::<WalletClientStates> {
725                state_machines: Arc::new(sm_gen),
726            }],
727        ))
728    }
729
730    pub async fn peg_in(&self, req: PegInRequest) -> anyhow::Result<PegInResponse> {
731        let (operation_id, address, _) = self.safe_allocate_deposit_address(req.extra_meta).await?;
732
733        Ok(PegInResponse {
734            deposit_address: Address::from_script(&address.script_pubkey(), self.get_network())?
735                .as_unchecked()
736                .clone(),
737            operation_id,
738        })
739    }
740
741    pub async fn peg_out(&self, req: PegOutRequest) -> anyhow::Result<PegOutResponse> {
742        let amount = bitcoin::Amount::from_sat(req.amount_sat);
743        let destination = req
744            .destination_address
745            .require_network(self.get_network())?;
746
747        let fees = self.get_withdraw_fees(&destination, amount).await?;
748        let operation_id = self
749            .withdraw(&destination, amount, fees, req.extra_meta)
750            .await
751            .context("Failed to initiate withdraw")?;
752
753        Ok(PegOutResponse { operation_id })
754    }
755
756    pub fn create_rbf_withdraw_output(
757        &self,
758        operation_id: OperationId,
759        rbf: &Rbf,
760    ) -> anyhow::Result<ClientOutputBundle<WalletOutput, WalletClientStates>> {
761        let output = WalletOutput::new_v0_rbf(rbf.fees, rbf.txid);
762
763        let amount = output.maybe_v0_ref().expect("v0 output").amount().into();
764
765        let sm_gen = move |out_point_range: OutPointRange| {
766            assert_eq!(out_point_range.count(), 1);
767            let out_idx = out_point_range.start_idx();
768            vec![WalletClientStates::Withdraw(WithdrawStateMachine {
769                operation_id,
770                state: WithdrawStates::Created(CreatedWithdrawState {
771                    fm_outpoint: OutPoint {
772                        txid: out_point_range.txid(),
773                        out_idx,
774                    },
775                }),
776            })]
777        };
778
779        Ok(ClientOutputBundle::new(
780            vec![ClientOutput::<WalletOutput> { output, amount }],
781            vec![ClientOutputSM::<WalletClientStates> {
782                state_machines: Arc::new(sm_gen),
783            }],
784        ))
785    }
786
787    pub async fn btc_tx_has_no_size_limit(&self) -> FederationResult<bool> {
788        Ok(self.module_api.module_consensus_version().await? >= ModuleConsensusVersion::new(2, 2))
789    }
790
791    /// Returns true if the federation's wallet module consensus version
792    /// supports processing all deposits.
793    ///
794    /// This method is safe to call offline, since it first attempts to read a
795    /// key from the db that represents the client has previously been able to
796    /// verify the wallet module consensus version. If the client has not
797    /// verified the version, it must be online to fetch the latest wallet
798    /// module consensus version.
799    pub async fn supports_safe_deposit(&self) -> bool {
800        let mut dbtx = self.db.begin_transaction().await;
801
802        let already_verified_supports_safe_deposit =
803            dbtx.get_value(&SupportsSafeDepositKey).await.is_some();
804
805        already_verified_supports_safe_deposit || {
806            match self.module_api.module_consensus_version().await {
807                Ok(module_consensus_version) => {
808                    let supported_version =
809                        SAFE_DEPOSIT_MODULE_CONSENSUS_VERSION <= module_consensus_version;
810
811                    if supported_version {
812                        dbtx.insert_new_entry(&SupportsSafeDepositKey, &()).await;
813                        dbtx.commit_tx().await;
814                    }
815
816                    supported_version
817                }
818                Err(_) => false,
819            }
820        }
821    }
822
823    /// Allocates a deposit address controlled by the federation, guaranteeing
824    /// safe handling of all deposits, including on-chain transactions exceeding
825    /// `ALEPH_BFT_UNIT_BYTE_LIMIT`.
826    ///
827    /// Returns an error if the client has never been online to verify the
828    /// federation's wallet module consensus version supports processing all
829    /// deposits.
830    pub async fn safe_allocate_deposit_address<M>(
831        &self,
832        extra_meta: M,
833    ) -> anyhow::Result<(OperationId, Address, TweakIdx)>
834    where
835        M: Serialize + MaybeSend + MaybeSync,
836    {
837        ensure!(
838            self.supports_safe_deposit().await,
839            "Wallet module consensus version doesn't support safe deposits",
840        );
841
842        self.allocate_deposit_address_expert_only(extra_meta).await
843    }
844
845    /// Allocates a deposit address that is controlled by the federation.
846    ///
847    /// This is an EXPERT ONLY method intended for power users such as Lightning
848    /// gateways allocating liquidity, and we discourage exposing peg-in
849    /// functionality to everyday users of a Fedimint wallet due to the
850    /// following two limitations:
851    ///
852    /// The transaction sending to this address needs to be smaller than 40KB in
853    /// order for the peg-in to be claimable. If the transaction is too large,
854    /// funds will be lost.
855    ///
856    /// In the future, federations will also enforce a minimum peg-in amount to
857    /// prevent accumulation of dust UTXOs. Peg-ins under this minimum cannot be
858    /// claimed and funds will be lost.
859    ///
860    /// Everyday users should rely on Lightning to move funds into the
861    /// federation.
862    pub async fn allocate_deposit_address_expert_only<M>(
863        &self,
864        extra_meta: M,
865    ) -> anyhow::Result<(OperationId, Address, TweakIdx)>
866    where
867        M: Serialize + MaybeSend + MaybeSync,
868    {
869        let extra_meta_value =
870            serde_json::to_value(extra_meta).expect("Failed to serialize extra meta");
871        let (operation_id, address, tweak_idx) = self
872            .db
873            .autocommit(
874                move |dbtx, _| {
875                    let extra_meta_value_inner = extra_meta_value.clone();
876                    Box::pin(async move {
877                        let (operation_id, address, tweak_idx) = self
878                            .allocate_deposit_address_inner(dbtx)
879                            .await;
880
881                        self.client_ctx.manual_operation_start_dbtx(
882                            dbtx,
883                            operation_id,
884                            WalletCommonInit::KIND.as_str(),
885                            WalletOperationMeta {
886                                variant: WalletOperationMetaVariant::Deposit {
887                                    address: address.clone().into_unchecked(),
888                                    tweak_idx: Some(tweak_idx),
889                                    expires_at: None,
890                                },
891                                extra_meta: extra_meta_value_inner,
892                            },
893                            vec![]
894                        ).await?;
895
896                        debug!(target: LOG_CLIENT_MODULE_WALLET, %tweak_idx, %address, "Derived a new deposit address");
897
898                        // Begin watching the script address
899                        self.rpc.watch_script_history(&address.script_pubkey()).await?;
900
901                        let sender = self.pegin_monitor_wakeup_sender.clone();
902                        dbtx.on_commit(move || {
903                            sender.send_replace(());
904                        });
905
906                        Ok((operation_id, address, tweak_idx))
907                    })
908                },
909                Some(100),
910            )
911            .await
912            .map_err(|e| match e {
913                AutocommitError::CommitFailed {
914                    last_error,
915                    attempts,
916                } => last_error.context(format!("Failed to commit after {attempts} attempts")),
917                AutocommitError::ClosureError { error, .. } => error,
918            })?;
919
920        Ok((operation_id, address, tweak_idx))
921    }
922
923    /// Returns a stream of updates about an ongoing deposit operation created
924    /// with [`WalletClientModule::allocate_deposit_address_expert_only`].
925    /// Returns an error for old deposit operations created prior to the 0.4
926    /// release and not driven to completion yet. This should be rare enough
927    /// that an indeterminate state is ok here.
928    pub async fn subscribe_deposit(
929        &self,
930        operation_id: OperationId,
931    ) -> anyhow::Result<UpdateStreamOrOutcome<DepositStateV2>> {
932        let operation = self
933            .client_ctx
934            .get_operation(operation_id)
935            .await
936            .with_context(|| anyhow!("Operation not found: {}", operation_id.fmt_short()))?;
937
938        if operation.operation_module_kind() != WalletCommonInit::KIND.as_str() {
939            bail!("Operation is not a wallet operation");
940        }
941
942        let operation_meta = operation.meta::<WalletOperationMeta>();
943
944        let WalletOperationMetaVariant::Deposit {
945            address, tweak_idx, ..
946        } = operation_meta.variant
947        else {
948            bail!("Operation is not a deposit operation");
949        };
950
951        let address = address.require_network(self.cfg().network.0)?;
952
953        // The old deposit operations don't have tweak_idx set
954        let Some(tweak_idx) = tweak_idx else {
955            // In case we are dealing with an old deposit that still uses state machines we
956            // don't have the logic here anymore to subscribe to updates. We can still read
957            // the final state though if it reached any.
958            let outcome_v1 = operation
959                .outcome::<DepositStateV1>()
960                .context("Old pending deposit, can't subscribe to updates")?;
961
962            let outcome_v2 = match outcome_v1 {
963                DepositStateV1::Claimed(tx_info) => DepositStateV2::Claimed {
964                    btc_deposited: tx_info.btc_transaction.output[tx_info.out_idx as usize].value,
965                    btc_out_point: bitcoin::OutPoint {
966                        txid: tx_info.btc_transaction.compute_txid(),
967                        vout: tx_info.out_idx,
968                    },
969                },
970                DepositStateV1::Failed(error) => DepositStateV2::Failed(error),
971                _ => bail!("Non-final outcome in operation log"),
972            };
973
974            return Ok(UpdateStreamOrOutcome::Outcome(outcome_v2));
975        };
976
977        Ok(self.client_ctx.outcome_or_updates(operation, operation_id, {
978            let stream_rpc = self.rpc.clone();
979            let stream_client_ctx = self.client_ctx.clone();
980            let stream_script_pub_key = address.script_pubkey();
981            move || {
982
983            stream! {
984                yield DepositStateV2::WaitingForTransaction;
985
986                retry(
987                    "subscribe script history",
988                    background_backoff(),
989                    || stream_rpc.watch_script_history(&stream_script_pub_key)
990                ).await.expect("Will never give up");
991                let (btc_out_point, btc_deposited) = retry(
992                    "fetch history",
993                    background_backoff(),
994                    || async {
995                        let history = stream_rpc.get_script_history(&stream_script_pub_key).await?;
996                        history.first().and_then(|tx| {
997                            let (out_idx, amount) = tx.output
998                                .iter()
999                                .enumerate()
1000                                .find_map(|(idx, output)| (output.script_pubkey == stream_script_pub_key).then_some((idx, output.value)))?;
1001                            let txid = tx.compute_txid();
1002
1003                            Some((
1004                                bitcoin::OutPoint {
1005                                    txid,
1006                                    vout: out_idx as u32,
1007                                },
1008                                amount
1009                            ))
1010                        }).context("No deposit transaction found")
1011                    }
1012                ).await.expect("Will never give up");
1013
1014                yield DepositStateV2::WaitingForConfirmation {
1015                    btc_deposited,
1016                    btc_out_point
1017                };
1018
1019                let claim_data = stream_client_ctx.module_db().wait_key_exists(&ClaimedPegInKey {
1020                    peg_in_index: tweak_idx,
1021                    btc_out_point,
1022                }).await;
1023
1024                yield DepositStateV2::Confirmed {
1025                    btc_deposited,
1026                    btc_out_point
1027                };
1028
1029                match stream_client_ctx.await_primary_module_outputs(operation_id, claim_data.change).await {
1030                    Ok(()) => yield DepositStateV2::Claimed {
1031                        btc_deposited,
1032                        btc_out_point
1033                    },
1034                    Err(e) => yield DepositStateV2::Failed(e.to_string())
1035                }
1036            }
1037        }}))
1038    }
1039
1040    pub async fn list_peg_in_tweak_idxes(&self) -> BTreeMap<TweakIdx, PegInTweakIndexData> {
1041        self.client_ctx
1042            .module_db()
1043            .clone()
1044            .begin_transaction_nc()
1045            .await
1046            .find_by_prefix(&PegInTweakIndexPrefix)
1047            .await
1048            .map(|(key, data)| (key.0, data))
1049            .collect()
1050            .await
1051    }
1052
1053    pub async fn find_tweak_idx_by_address(
1054        &self,
1055        address: bitcoin::Address<NetworkUnchecked>,
1056    ) -> anyhow::Result<TweakIdx> {
1057        let data = self.data.clone();
1058        let Some((tweak_idx, _)) = self
1059            .db
1060            .begin_transaction_nc()
1061            .await
1062            .find_by_prefix(&PegInTweakIndexPrefix)
1063            .await
1064            .filter(|(k, _)| {
1065                let (_, derived_address, _tweak_key, _) = data.derive_peg_in_script(k.0);
1066                future::ready(derived_address.into_unchecked() == address)
1067            })
1068            .next()
1069            .await
1070        else {
1071            bail!("Address not found in the list of derived keys");
1072        };
1073
1074        Ok(tweak_idx.0)
1075    }
1076    pub async fn find_tweak_idx_by_operation_id(
1077        &self,
1078        operation_id: OperationId,
1079    ) -> anyhow::Result<TweakIdx> {
1080        Ok(self
1081            .client_ctx
1082            .module_db()
1083            .clone()
1084            .begin_transaction_nc()
1085            .await
1086            .find_by_prefix(&PegInTweakIndexPrefix)
1087            .await
1088            .filter(|(_k, v)| future::ready(v.operation_id == operation_id))
1089            .next()
1090            .await
1091            .ok_or_else(|| anyhow::format_err!("OperationId not found"))?
1092            .0
1093            .0)
1094    }
1095
1096    pub async fn get_pegin_tweak_idx(
1097        &self,
1098        tweak_idx: TweakIdx,
1099    ) -> anyhow::Result<PegInTweakIndexData> {
1100        self.client_ctx
1101            .module_db()
1102            .clone()
1103            .begin_transaction_nc()
1104            .await
1105            .get_value(&PegInTweakIndexKey(tweak_idx))
1106            .await
1107            .ok_or_else(|| anyhow::format_err!("TweakIdx not found"))
1108    }
1109
1110    pub async fn get_claimed_pegins(
1111        &self,
1112        dbtx: &mut DatabaseTransaction<'_>,
1113        tweak_idx: TweakIdx,
1114    ) -> Vec<(
1115        bitcoin::OutPoint,
1116        TransactionId,
1117        Vec<fedimint_core::OutPoint>,
1118    )> {
1119        let outpoints = dbtx
1120            .get_value(&PegInTweakIndexKey(tweak_idx))
1121            .await
1122            .map(|v| v.claimed)
1123            .unwrap_or_default();
1124
1125        let mut res = vec![];
1126
1127        for outpoint in outpoints {
1128            let claimed_peg_in_data = dbtx
1129                .get_value(&ClaimedPegInKey {
1130                    peg_in_index: tweak_idx,
1131                    btc_out_point: outpoint,
1132                })
1133                .await
1134                .expect("Must have a corresponding claim record");
1135            res.push((
1136                outpoint,
1137                claimed_peg_in_data.claim_txid,
1138                claimed_peg_in_data.change,
1139            ));
1140        }
1141
1142        res
1143    }
1144
1145    /// Like [`Self::recheck_pegin_address`] but by `operation_id`
1146    pub async fn recheck_pegin_address_by_op_id(
1147        &self,
1148        operation_id: OperationId,
1149    ) -> anyhow::Result<()> {
1150        let tweak_idx = self.find_tweak_idx_by_operation_id(operation_id).await?;
1151
1152        self.recheck_pegin_address(tweak_idx).await
1153    }
1154
1155    /// Schedule given address for immediate re-check for deposits
1156    pub async fn recheck_pegin_address_by_address(
1157        &self,
1158        address: bitcoin::Address<NetworkUnchecked>,
1159    ) -> anyhow::Result<()> {
1160        self.recheck_pegin_address(self.find_tweak_idx_by_address(address).await?)
1161            .await
1162    }
1163
1164    /// Schedule given address for immediate re-check for deposits
1165    pub async fn recheck_pegin_address(&self, tweak_idx: TweakIdx) -> anyhow::Result<()> {
1166        self.db
1167            .autocommit(
1168                |dbtx, _| {
1169                    Box::pin(async {
1170                        let db_key = PegInTweakIndexKey(tweak_idx);
1171                        let db_val = dbtx
1172                            .get_value(&db_key)
1173                            .await
1174                            .ok_or_else(|| anyhow::format_err!("DBKey not found"))?;
1175
1176                        dbtx.insert_entry(
1177                            &db_key,
1178                            &PegInTweakIndexData {
1179                                next_check_time: Some(fedimint_core::time::now()),
1180                                ..db_val
1181                            },
1182                        )
1183                        .await;
1184
1185                        let sender = self.pegin_monitor_wakeup_sender.clone();
1186                        dbtx.on_commit(move || {
1187                            sender.send_replace(());
1188                        });
1189
1190                        Ok::<_, anyhow::Error>(())
1191                    })
1192                },
1193                Some(100),
1194            )
1195            .await?;
1196
1197        Ok(())
1198    }
1199
1200    /// Await for num deposit by [`OperationId`]
1201    pub async fn await_num_deposits_by_operation_id(
1202        &self,
1203        operation_id: OperationId,
1204        num_deposits: usize,
1205    ) -> anyhow::Result<()> {
1206        let tweak_idx = self.find_tweak_idx_by_operation_id(operation_id).await?;
1207        self.await_num_deposits(tweak_idx, num_deposits).await
1208    }
1209
1210    pub async fn await_num_deposits_by_address(
1211        &self,
1212        address: bitcoin::Address<NetworkUnchecked>,
1213        num_deposits: usize,
1214    ) -> anyhow::Result<()> {
1215        self.await_num_deposits(self.find_tweak_idx_by_address(address).await?, num_deposits)
1216            .await
1217    }
1218
1219    #[instrument(target = LOG_CLIENT_MODULE_WALLET, skip_all, fields(tweak_idx=?tweak_idx, num_deposists=num_deposits))]
1220    pub async fn await_num_deposits(
1221        &self,
1222        tweak_idx: TweakIdx,
1223        num_deposits: usize,
1224    ) -> anyhow::Result<()> {
1225        let operation_id = self.get_pegin_tweak_idx(tweak_idx).await?.operation_id;
1226
1227        let mut receiver = self.pegin_claimed_receiver.clone();
1228        let mut backoff = backoff_util::aggressive_backoff();
1229
1230        loop {
1231            let pegins = self
1232                .get_claimed_pegins(
1233                    &mut self.client_ctx.module_db().begin_transaction_nc().await,
1234                    tweak_idx,
1235                )
1236                .await;
1237
1238            if pegins.len() < num_deposits {
1239                debug!(target: LOG_CLIENT_MODULE_WALLET, has=pegins.len(), "Not enough deposits");
1240                self.recheck_pegin_address(tweak_idx).await?;
1241                runtime::sleep(backoff.next().unwrap_or_default()).await;
1242                receiver.changed().await?;
1243                continue;
1244            }
1245
1246            debug!(target: LOG_CLIENT_MODULE_WALLET, has=pegins.len(), "Enough deposits detected");
1247
1248            for (_outpoint, transaction_id, change) in pegins {
1249                if transaction_id == TransactionId::from_byte_array([0; 32]) && change.is_empty() {
1250                    debug!(target: LOG_CLIENT_MODULE_WALLET, "Deposited amount was too low, skipping");
1251                    continue;
1252                }
1253
1254                debug!(target: LOG_CLIENT_MODULE_WALLET, out_points=?change, "Ensuring deposists claimed");
1255                let tx_subscriber = self.client_ctx.transaction_updates(operation_id).await;
1256
1257                if let Err(e) = tx_subscriber.await_tx_accepted(transaction_id).await {
1258                    bail!("{}", e);
1259                }
1260
1261                debug!(target: LOG_CLIENT_MODULE_WALLET, out_points=?change, "Ensuring outputs claimed");
1262                self.client_ctx
1263                    .await_primary_module_outputs(operation_id, change)
1264                    .await
1265                    .expect("Cannot fail if tx was accepted and federation is honest");
1266            }
1267
1268            return Ok(());
1269        }
1270    }
1271
1272    /// Attempt to withdraw a given `amount` of Bitcoin to a destination
1273    /// `address`. The caller has to supply the fee rate to be used which can be
1274    /// fetched using [`Self::get_withdraw_fees`] and should be
1275    /// acknowledged by the user since it can be unexpectedly high.
1276    pub async fn withdraw<M: Serialize + MaybeSend + MaybeSync>(
1277        &self,
1278        address: &bitcoin::Address,
1279        amount: bitcoin::Amount,
1280        fee: PegOutFees,
1281        extra_meta: M,
1282    ) -> anyhow::Result<OperationId> {
1283        {
1284            let operation_id = OperationId(thread_rng().r#gen());
1285
1286            let withdraw_output =
1287                self.create_withdraw_output(operation_id, address.clone(), amount, fee)?;
1288            let tx_builder = TransactionBuilder::new()
1289                .with_outputs(self.client_ctx.make_client_outputs(withdraw_output));
1290
1291            let extra_meta =
1292                serde_json::to_value(extra_meta).expect("Failed to serialize extra meta");
1293            self.client_ctx
1294                .finalize_and_submit_transaction(
1295                    operation_id,
1296                    WalletCommonInit::KIND.as_str(),
1297                    {
1298                        let address = address.clone();
1299                        move |change_range: OutPointRange| WalletOperationMeta {
1300                            variant: WalletOperationMetaVariant::Withdraw {
1301                                address: address.clone().into_unchecked(),
1302                                amount,
1303                                fee,
1304                                change: change_range.into_iter().collect(),
1305                            },
1306                            extra_meta: extra_meta.clone(),
1307                        }
1308                    },
1309                    tx_builder,
1310                )
1311                .await?;
1312
1313            Ok(operation_id)
1314        }
1315    }
1316
1317    /// Attempt to increase the fee of a onchain withdraw transaction using
1318    /// replace by fee (RBF).
1319    /// This can prevent transactions from getting stuck
1320    /// in the mempool
1321    #[deprecated(
1322        since = "0.4.0",
1323        note = "RBF withdrawals are rejected by the federation"
1324    )]
1325    pub async fn rbf_withdraw<M: Serialize + MaybeSync + MaybeSend>(
1326        &self,
1327        rbf: Rbf,
1328        extra_meta: M,
1329    ) -> anyhow::Result<OperationId> {
1330        let operation_id = OperationId(thread_rng().r#gen());
1331
1332        let withdraw_output = self.create_rbf_withdraw_output(operation_id, &rbf)?;
1333        let tx_builder = TransactionBuilder::new()
1334            .with_outputs(self.client_ctx.make_client_outputs(withdraw_output));
1335
1336        let extra_meta = serde_json::to_value(extra_meta).expect("Failed to serialize extra meta");
1337        self.client_ctx
1338            .finalize_and_submit_transaction(
1339                operation_id,
1340                WalletCommonInit::KIND.as_str(),
1341                move |change_range: OutPointRange| WalletOperationMeta {
1342                    variant: WalletOperationMetaVariant::RbfWithdraw {
1343                        rbf: rbf.clone(),
1344                        change: change_range.into_iter().collect(),
1345                    },
1346                    extra_meta: extra_meta.clone(),
1347                },
1348                tx_builder,
1349            )
1350            .await?;
1351
1352        Ok(operation_id)
1353    }
1354
1355    pub async fn subscribe_withdraw_updates(
1356        &self,
1357        operation_id: OperationId,
1358    ) -> anyhow::Result<UpdateStreamOrOutcome<WithdrawState>> {
1359        let operation = self
1360            .client_ctx
1361            .get_operation(operation_id)
1362            .await
1363            .with_context(|| anyhow!("Operation not found: {}", operation_id.fmt_short()))?;
1364
1365        if operation.operation_module_kind() != WalletCommonInit::KIND.as_str() {
1366            bail!("Operation is not a wallet operation");
1367        }
1368
1369        let operation_meta = operation.meta::<WalletOperationMeta>();
1370
1371        let (WalletOperationMetaVariant::Withdraw { change, .. }
1372        | WalletOperationMetaVariant::RbfWithdraw { change, .. }) = operation_meta.variant
1373        else {
1374            bail!("Operation is not a withdraw operation");
1375        };
1376
1377        let mut operation_stream = self.notifier.subscribe(operation_id).await;
1378        let client_ctx = self.client_ctx.clone();
1379
1380        Ok(self
1381            .client_ctx
1382            .outcome_or_updates(operation, operation_id, move || {
1383                stream! {
1384                    match next_withdraw_state(&mut operation_stream).await {
1385                        Some(WithdrawStates::Created(_)) => {
1386                            yield WithdrawState::Created;
1387                        },
1388                        Some(s) => {
1389                            panic!("Unexpected state {s:?}")
1390                        },
1391                        None => return,
1392                    }
1393
1394                    // TODO: get rid of awaiting change here, there has to be a better way to make tests deterministic
1395
1396                        // Swallowing potential errors since the transaction failing  is handled by
1397                        // output outcome fetching already
1398                        let _ = client_ctx
1399                            .await_primary_module_outputs(operation_id, change)
1400                            .await;
1401
1402
1403                    match next_withdraw_state(&mut operation_stream).await {
1404                        Some(WithdrawStates::Aborted(inner)) => {
1405                            yield WithdrawState::Failed(inner.error);
1406                        },
1407                        Some(WithdrawStates::Success(inner)) => {
1408                            yield WithdrawState::Succeeded(inner.txid);
1409                        },
1410                        Some(s) => {
1411                            panic!("Unexpected state {s:?}")
1412                        },
1413                        None => {},
1414                    }
1415                }
1416            }))
1417    }
1418
1419    fn admin_auth(&self) -> anyhow::Result<ApiAuth> {
1420        self.admin_auth
1421            .clone()
1422            .ok_or_else(|| anyhow::format_err!("Admin auth not set"))
1423    }
1424
1425    pub async fn activate_consensus_version_voting(&self) -> anyhow::Result<()> {
1426        self.module_api
1427            .activate_consensus_version_voting(self.admin_auth()?)
1428            .await?;
1429
1430        Ok(())
1431    }
1432}
1433
1434/// Polls the federation checking if the activated module consensus version
1435/// supports safe deposits, saving the result in the db once it does.
1436async fn poll_supports_safe_deposit_version(db: Database, module_api: DynModuleApi) {
1437    loop {
1438        let mut dbtx = db.begin_transaction().await;
1439
1440        if dbtx.get_value(&SupportsSafeDepositKey).await.is_some() {
1441            break;
1442        }
1443
1444        if let Ok(module_consensus_version) = module_api.module_consensus_version().await
1445            && SAFE_DEPOSIT_MODULE_CONSENSUS_VERSION <= module_consensus_version
1446        {
1447            dbtx.insert_new_entry(&SupportsSafeDepositKey, &()).await;
1448            dbtx.commit_tx().await;
1449            break;
1450        }
1451
1452        drop(dbtx);
1453
1454        if is_running_in_test_env() {
1455            // Even in tests we don't want to spam the federation with requests about it
1456            sleep(Duration::from_secs(10)).await;
1457        } else {
1458            sleep(Duration::from_secs(3600)).await;
1459        }
1460    }
1461}
1462
1463/// Returns the child index to derive the next peg-in tweak key from.
1464async fn get_next_peg_in_tweak_child_id(dbtx: &mut DatabaseTransaction<'_>) -> TweakIdx {
1465    let index = dbtx
1466        .get_value(&NextPegInTweakIndexKey)
1467        .await
1468        .unwrap_or_default();
1469    dbtx.insert_entry(&NextPegInTweakIndexKey, &(index.next()))
1470        .await;
1471    index
1472}
1473
1474#[derive(Debug, Clone, Eq, PartialEq, Hash, Decodable, Encodable)]
1475pub enum WalletClientStates {
1476    Deposit(DepositStateMachine),
1477    Withdraw(WithdrawStateMachine),
1478}
1479
1480impl IntoDynInstance for WalletClientStates {
1481    type DynType = DynState;
1482
1483    fn into_dyn(self, instance_id: ModuleInstanceId) -> Self::DynType {
1484        DynState::from_typed(instance_id, self)
1485    }
1486}
1487
1488impl State for WalletClientStates {
1489    type ModuleContext = WalletClientContext;
1490
1491    fn transitions(
1492        &self,
1493        context: &Self::ModuleContext,
1494        global_context: &DynGlobalClientContext,
1495    ) -> Vec<StateTransition<Self>> {
1496        match self {
1497            WalletClientStates::Deposit(sm) => {
1498                sm_enum_variant_translation!(
1499                    sm.transitions(context, global_context),
1500                    WalletClientStates::Deposit
1501                )
1502            }
1503            WalletClientStates::Withdraw(sm) => {
1504                sm_enum_variant_translation!(
1505                    sm.transitions(context, global_context),
1506                    WalletClientStates::Withdraw
1507                )
1508            }
1509        }
1510    }
1511
1512    fn operation_id(&self) -> OperationId {
1513        match self {
1514            WalletClientStates::Deposit(sm) => sm.operation_id(),
1515            WalletClientStates::Withdraw(sm) => sm.operation_id(),
1516        }
1517    }
1518}
1519
1520#[cfg(all(test, not(target_family = "wasm")))]
1521mod tests {
1522    use std::collections::BTreeSet;
1523    use std::sync::atomic::{AtomicBool, Ordering};
1524
1525    use super::*;
1526    use crate::backup::{
1527        RECOVER_NUM_IDX_ADD_TO_LAST_USED, RecoverScanOutcome, recover_scan_idxes_for_activity,
1528    };
1529
1530    #[allow(clippy::too_many_lines)] // shut-up clippy, it's a test
1531    #[tokio::test(flavor = "multi_thread")]
1532    async fn sanity_test_recover_inner() {
1533        {
1534            let last_checked = AtomicBool::new(false);
1535            let last_checked = &last_checked;
1536            assert_eq!(
1537                recover_scan_idxes_for_activity(
1538                    TweakIdx(0),
1539                    &BTreeSet::new(),
1540                    |cur_idx| async move {
1541                        Ok(match cur_idx {
1542                            TweakIdx(9) => {
1543                                last_checked.store(true, Ordering::SeqCst);
1544                                vec![]
1545                            }
1546                            TweakIdx(10) => panic!("Shouldn't happen"),
1547                            TweakIdx(11) => {
1548                                vec![0usize] /* just for type inference */
1549                            }
1550                            _ => vec![],
1551                        })
1552                    }
1553                )
1554                .await
1555                .unwrap(),
1556                RecoverScanOutcome {
1557                    last_used_idx: None,
1558                    new_start_idx: TweakIdx(RECOVER_NUM_IDX_ADD_TO_LAST_USED),
1559                    tweak_idxes_with_pegins: BTreeSet::from([])
1560                }
1561            );
1562            assert!(last_checked.load(Ordering::SeqCst));
1563        }
1564
1565        {
1566            let last_checked = AtomicBool::new(false);
1567            let last_checked = &last_checked;
1568            assert_eq!(
1569                recover_scan_idxes_for_activity(
1570                    TweakIdx(0),
1571                    &BTreeSet::from([TweakIdx(1), TweakIdx(2)]),
1572                    |cur_idx| async move {
1573                        Ok(match cur_idx {
1574                            TweakIdx(1) => panic!("Shouldn't happen: already used (1)"),
1575                            TweakIdx(2) => panic!("Shouldn't happen: already used (2)"),
1576                            TweakIdx(11) => {
1577                                last_checked.store(true, Ordering::SeqCst);
1578                                vec![]
1579                            }
1580                            TweakIdx(12) => panic!("Shouldn't happen"),
1581                            TweakIdx(13) => {
1582                                vec![0usize] /* just for type inference */
1583                            }
1584                            _ => vec![],
1585                        })
1586                    }
1587                )
1588                .await
1589                .unwrap(),
1590                RecoverScanOutcome {
1591                    last_used_idx: Some(TweakIdx(2)),
1592                    new_start_idx: TweakIdx(2 + RECOVER_NUM_IDX_ADD_TO_LAST_USED),
1593                    tweak_idxes_with_pegins: BTreeSet::from([])
1594                }
1595            );
1596            assert!(last_checked.load(Ordering::SeqCst));
1597        }
1598
1599        {
1600            let last_checked = AtomicBool::new(false);
1601            let last_checked = &last_checked;
1602            assert_eq!(
1603                recover_scan_idxes_for_activity(
1604                    TweakIdx(10),
1605                    &BTreeSet::new(),
1606                    |cur_idx| async move {
1607                        Ok(match cur_idx {
1608                            TweakIdx(10) => vec![()],
1609                            TweakIdx(19) => {
1610                                last_checked.store(true, Ordering::SeqCst);
1611                                vec![]
1612                            }
1613                            TweakIdx(20) => panic!("Shouldn't happen"),
1614                            _ => vec![],
1615                        })
1616                    }
1617                )
1618                .await
1619                .unwrap(),
1620                RecoverScanOutcome {
1621                    last_used_idx: Some(TweakIdx(10)),
1622                    new_start_idx: TweakIdx(10 + RECOVER_NUM_IDX_ADD_TO_LAST_USED),
1623                    tweak_idxes_with_pegins: BTreeSet::from([TweakIdx(10)])
1624                }
1625            );
1626            assert!(last_checked.load(Ordering::SeqCst));
1627        }
1628
1629        assert_eq!(
1630            recover_scan_idxes_for_activity(TweakIdx(0), &BTreeSet::new(), |cur_idx| async move {
1631                Ok(match cur_idx {
1632                    TweakIdx(6 | 15) => vec![()],
1633                    _ => vec![],
1634                })
1635            })
1636            .await
1637            .unwrap(),
1638            RecoverScanOutcome {
1639                last_used_idx: Some(TweakIdx(15)),
1640                new_start_idx: TweakIdx(15 + RECOVER_NUM_IDX_ADD_TO_LAST_USED),
1641                tweak_idxes_with_pegins: BTreeSet::from([TweakIdx(6), TweakIdx(15)])
1642            }
1643        );
1644        assert_eq!(
1645            recover_scan_idxes_for_activity(TweakIdx(10), &BTreeSet::new(), |cur_idx| async move {
1646                Ok(match cur_idx {
1647                    TweakIdx(8) => {
1648                        vec![()] /* for type inference only */
1649                    }
1650                    TweakIdx(9) => {
1651                        panic!("Shouldn't happen")
1652                    }
1653                    _ => vec![],
1654                })
1655            })
1656            .await
1657            .unwrap(),
1658            RecoverScanOutcome {
1659                last_used_idx: None,
1660                new_start_idx: TweakIdx(9 + RECOVER_NUM_IDX_ADD_TO_LAST_USED),
1661                tweak_idxes_with_pegins: BTreeSet::from([])
1662            }
1663        );
1664        assert_eq!(
1665            recover_scan_idxes_for_activity(TweakIdx(10), &BTreeSet::new(), |cur_idx| async move {
1666                Ok(match cur_idx {
1667                    TweakIdx(9) => panic!("Shouldn't happen"),
1668                    TweakIdx(15) => vec![()],
1669                    _ => vec![],
1670                })
1671            })
1672            .await
1673            .unwrap(),
1674            RecoverScanOutcome {
1675                last_used_idx: Some(TweakIdx(15)),
1676                new_start_idx: TweakIdx(15 + RECOVER_NUM_IDX_ADD_TO_LAST_USED),
1677                tweak_idxes_with_pegins: BTreeSet::from([TweakIdx(15)])
1678            }
1679        );
1680    }
1681}