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;
15mod deposit;
18pub mod error;
20pub mod events;
21use events::SendPaymentEvent;
22mod 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 pub btc_transaction: bitcoin::Transaction,
106 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#[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#[allow(clippy::enum_variant_names)]
154#[derive(Debug, Clone, PartialEq, Eq)]
155pub enum MaybeNewAddress {
156 NewAddress(DepositAddressInfo),
158 TooManyUnusedAddresses(Vec<DepositAddressInfo>),
162}
163
164#[derive(Debug, Clone, Copy, PartialEq, Eq)]
170pub enum AllocateDepositOutcome {
171 Fresh,
173 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 }
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)]
202pub 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 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 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 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 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 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 #[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#[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 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 pegin_monitor_wakeup_sender: watch::Sender<()>,
571 pegin_monitor_wakeup_receiver: watch::Receiver<()>,
572 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 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#[derive(Debug, thiserror::Error)]
754enum RpcError {
755 #[error(transparent)]
758 Json(#[from] serde_json::Error),
759
760 #[error("peg_in failed")]
762 PegIn(#[source] DepositAddressError),
763
764 #[error("peg_out failed")]
766 PegOut(#[source] PegOutError),
767
768 #[error(transparent)]
770 SubscribeDeposit(#[from] SubscribeDepositError),
771
772 #[error(transparent)]
774 SubscribeWithdraw(#[from] SubscribeWithdrawError),
775
776 #[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 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 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 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 pub async fn max_withdrawable_amount(
977 &self,
978 address: &bitcoin::Address,
979 balance: fedimint_core::Amount,
980 ) -> Result<(bitcoin::Amount, PegOutFees), MaxWithdrawableAmountError> {
981 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 |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 let amount = bitcoin::Amount::from_sat(max.msats.div_ceil(1000));
1016
1017 let fees = self.get_withdraw_fees(address, amount).await?;
1022
1023 Ok((amount, fees))
1024 }
1025
1026 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 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 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 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 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 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 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 unused.sort_by_key(|(t, d)| (d.creation_time, *t));
1388 unused
1389 }
1390
1391 #[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 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 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 let Some(tweak_idx) = tweak_idx else {
1530 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 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 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 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 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 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 #[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 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#[derive(Debug, thiserror::Error)]
2039enum FetchDepositTransactionError {
2040 #[error(transparent)]
2042 BitcoinRpc(#[from] BitcoinRpcError),
2043
2044 #[error("No deposit transaction found")]
2046 NotFound,
2047}
2048
2049async 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 sleep(Duration::from_secs(10)).await;
2066 } else {
2067 sleep(Duration::from_hours(1)).await;
2068 }
2069 }
2070}
2071
2072async 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
2082async 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
2108async fn store_supports_safe_deposit(mut dbtx: DatabaseTransaction<'_, Committable>) {
2123 if dbtx.get_value(&SupportsSafeDepositKey).await.is_some() {
2124 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
2142async 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;