1#![deny(clippy::pedantic)]
2#![allow(clippy::missing_errors_doc)]
3#![allow(clippy::missing_panics_doc)]
4#![allow(clippy::module_name_repetitions)]
5#![allow(clippy::must_use_candidate)]
6
7pub use fedimint_lnv2_common as common;
8
9mod api;
10#[cfg(feature = "cli")]
11mod cli;
12pub mod db;
13pub mod events;
14mod receive_sm;
15mod send_sm;
16
17use std::collections::{BTreeMap, BTreeSet};
18use std::sync::Arc;
19
20use async_stream::stream;
21use bitcoin::hashes::{Hash, sha256};
22use bitcoin::secp256k1;
23use db::{DbKeyPrefix, GatewayKey, IncomingContractStreamIndexKey};
24use fedimint_api_client::api::DynModuleApi;
25use fedimint_client_module::module::init::{ClientModuleInit, ClientModuleInitArgs};
26use fedimint_client_module::module::recovery::NoModuleBackup;
27use fedimint_client_module::module::{ClientContext, ClientModule, OutPointRange};
28use fedimint_client_module::oplog::UpdateStreamOrOutcome;
29use fedimint_client_module::sm::{Context, DynState, ModuleNotifier, State, StateTransition};
30use fedimint_client_module::transaction::{
31 ClientOutput, ClientOutputBundle, ClientOutputSM, FeeQuote, FeeQuoteRequest,
32 TransactionBuilder, max_affordable_send_amount,
33};
34use fedimint_client_module::{DynGlobalClientContext, sm_enum_variant_translation};
35use fedimint_core::config::FederationId;
36use fedimint_core::core::{IntoDynInstance, ModuleInstanceId, ModuleKind, OperationId};
37use fedimint_core::db::{DatabaseTransaction, IDatabaseTransactionOpsCoreTyped};
38use fedimint_core::encoding::{Decodable, Encodable};
39use fedimint_core::module::{
40 Amounts, ApiAuth, ApiVersion, CommonModuleInit, ModuleCommon, ModuleInit, MultiApiVersion,
41};
42use fedimint_core::secp256k1::SECP256K1;
43use fedimint_core::task::TaskGroup;
44use fedimint_core::time::duration_since_epoch;
45use fedimint_core::util::SafeUrl;
46use fedimint_core::{Amount, PeerId, apply, async_trait_maybe_send};
47use fedimint_derive_secret::{ChildId, DerivableSecret};
48use fedimint_lnv2_common::config::LightningClientConfig;
49use fedimint_lnv2_common::contracts::{IncomingContract, OutgoingContract, PaymentImage};
50use fedimint_lnv2_common::gateway_api::{
51 GatewayConnection, PaymentFee, RealGatewayConnection, RoutingInfo,
52};
53use fedimint_lnv2_common::{
54 Bolt11InvoiceDescription, GatewayApi, KIND, LightningCommonInit, LightningInvoice,
55 LightningModuleTypes, LightningOutput, LightningOutputV0, MINIMUM_INCOMING_CONTRACT_AMOUNT,
56 lnurl, tweak,
57};
58use futures::StreamExt;
59use lightning_invoice::{Bolt11Invoice, Currency};
60use secp256k1::{Keypair, PublicKey, Scalar, SecretKey, ecdh};
61use serde::{Deserialize, Serialize};
62use serde_json::Value;
63use strum::IntoEnumIterator as _;
64use thiserror::Error;
65use tpe::{AggregateDecryptionKey, derive_agg_dk};
66use tracing::warn;
67
68use crate::api::LightningFederationApi;
69use crate::events::SendPaymentEvent;
70use crate::receive_sm::{ReceiveSMCommon, ReceiveSMState, ReceiveStateMachine};
71use crate::send_sm::{SendSMCommon, SendSMState, SendStateMachine};
72
73const EXPIRATION_DELTA_LIMIT: u64 = 1440;
76
77const CONTRACT_CONFIRMATION_BUFFER: u64 = 12;
79
80#[allow(clippy::large_enum_variant)]
81#[derive(Debug, Clone, Serialize, Deserialize)]
82pub enum LightningOperationMeta {
83 Send(SendOperationMeta),
84 Receive(ReceiveOperationMeta),
85 LnurlReceive(LnurlReceiveOperationMeta),
86}
87
88#[derive(Debug, Clone, Serialize, Deserialize)]
89pub struct SendOperationMeta {
90 pub change_outpoint_range: OutPointRange,
91 pub gateway: SafeUrl,
92 pub contract: OutgoingContract,
93 pub invoice: LightningInvoice,
94 pub custom_meta: Value,
95}
96
97impl SendOperationMeta {
98 pub fn gateway_fee(&self) -> Amount {
100 match &self.invoice {
101 LightningInvoice::Bolt11(invoice) => self.contract.amount.saturating_sub(
102 Amount::from_msats(invoice.amount_milli_satoshis().expect("Invoice has amount")),
103 ),
104 }
105 }
106}
107
108#[derive(Debug, Clone, Serialize, Deserialize)]
109pub struct ReceiveOperationMeta {
110 pub gateway: SafeUrl,
111 pub contract: IncomingContract,
112 pub invoice: LightningInvoice,
113 pub custom_meta: Value,
114}
115
116impl ReceiveOperationMeta {
117 pub fn gateway_fee(&self) -> Amount {
119 match &self.invoice {
120 LightningInvoice::Bolt11(invoice) => {
121 Amount::from_msats(invoice.amount_milli_satoshis().expect("Invoice has amount"))
122 .saturating_sub(self.contract.commitment.amount)
123 }
124 }
125 }
126}
127
128#[derive(Debug, Clone, Serialize, Deserialize)]
129pub struct LnurlReceiveOperationMeta {
130 pub contract: IncomingContract,
131 pub custom_meta: Value,
132}
133
134#[cfg_attr(doc, aquamarine::aquamarine)]
135#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
153pub enum SendOperationState {
154 Funding,
156 Funded,
158 Success([u8; 32]),
160 Refunding,
162 Refunded,
164 Failure,
166}
167
168#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
170pub enum FinalSendOperationState {
171 Success(#[serde(with = "fedimint_core::hex::serde")] [u8; 32]),
174 Refunded,
176 Failure,
178}
179
180pub type SendResult = Result<OperationId, SendPaymentError>;
181
182#[cfg_attr(doc, aquamarine::aquamarine)]
183#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
195pub enum ReceiveOperationState {
196 Pending,
198 Expired,
200 Claiming,
202 Claimed,
204 Failure,
206}
207
208#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
210pub enum FinalReceiveOperationState {
211 Expired,
213 Claimed,
215 Failure,
217}
218
219pub type ReceiveResult = Result<(Bolt11Invoice, OperationId), ReceiveError>;
220
221#[derive(Clone)]
222pub struct LightningClientInit {
223 pub gateway_conn: Option<Arc<dyn GatewayConnection + Send + Sync>>,
224 pub custom_meta_fn: Arc<dyn Fn() -> Value + Send + Sync>,
225}
226
227impl std::fmt::Debug for LightningClientInit {
228 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
229 f.debug_struct("LightningClientInit")
230 .field("gateway_conn", &self.gateway_conn)
231 .field("custom_meta_fn", &"<function>")
232 .finish()
233 }
234}
235
236impl Default for LightningClientInit {
237 fn default() -> Self {
238 LightningClientInit {
239 gateway_conn: None,
240 custom_meta_fn: Arc::new(|| Value::Null),
241 }
242 }
243}
244
245impl ModuleInit for LightningClientInit {
246 type Common = LightningCommonInit;
247
248 async fn dump_database(
249 &self,
250 _dbtx: &mut DatabaseTransaction<'_>,
251 _prefix_names: Vec<String>,
252 ) -> Box<dyn Iterator<Item = (String, Box<dyn erased_serde::Serialize + Send>)> + '_> {
253 Box::new(BTreeMap::new().into_iter())
254 }
255}
256
257#[apply(async_trait_maybe_send!)]
258impl ClientModuleInit for LightningClientInit {
259 type Module = LightningClientModule;
260
261 fn supported_api_versions(&self) -> MultiApiVersion {
262 MultiApiVersion::try_from_iter([ApiVersion { major: 0, minor: 0 }])
263 .expect("no version conflicts")
264 }
265
266 async fn init(&self, args: &ClientModuleInitArgs<Self>) -> anyhow::Result<Self::Module> {
267 let gateway_conn = if let Some(gateway_conn) = self.gateway_conn.clone() {
268 gateway_conn
269 } else {
270 let api = GatewayApi::new(None, args.connector_registry.clone());
271 Arc::new(RealGatewayConnection { api })
272 };
273 Ok(LightningClientModule::new(
274 *args.federation_id(),
275 args.cfg().clone(),
276 args.notifier().clone(),
277 args.context(),
278 args.module_api().clone(),
279 args.module_root_secret(),
280 gateway_conn,
281 self.custom_meta_fn.clone(),
282 args.admin_auth().cloned(),
283 args.task_group(),
284 args.client_span(),
285 ))
286 }
287
288 fn used_db_prefixes(&self) -> Option<BTreeSet<u8>> {
289 Some(
290 DbKeyPrefix::iter()
291 .map(|p| p as u8)
292 .chain(
293 DbKeyPrefix::ExternalReservedStart as u8
294 ..=DbKeyPrefix::CoreInternalReservedEnd as u8,
295 )
296 .collect(),
297 )
298 }
299}
300
301#[derive(Debug, Clone)]
302pub struct LightningClientContext {
303 federation_id: FederationId,
304 gateway_conn: Arc<dyn GatewayConnection + Send + Sync>,
305 pub(crate) client_ctx: ClientContext<LightningClientModule>,
306}
307
308impl Context for LightningClientContext {
309 const KIND: Option<ModuleKind> = Some(KIND);
310}
311
312#[derive(Debug, Clone)]
313pub struct LightningClientModule {
314 federation_id: FederationId,
315 cfg: LightningClientConfig,
316 notifier: ModuleNotifier<LightningClientStateMachines>,
317 client_ctx: ClientContext<Self>,
318 module_api: DynModuleApi,
319 keypair: Keypair,
320 lnurl_keypair: Keypair,
321 gateway_conn: Arc<dyn GatewayConnection + Send + Sync>,
322 #[allow(unused)] admin_auth: Option<ApiAuth>,
324}
325
326#[apply(async_trait_maybe_send!)]
327impl ClientModule for LightningClientModule {
328 type Init = LightningClientInit;
329 type Common = LightningModuleTypes;
330 type Backup = NoModuleBackup;
331 type ModuleStateMachineContext = LightningClientContext;
332 type States = LightningClientStateMachines;
333
334 fn context(&self) -> Self::ModuleStateMachineContext {
335 LightningClientContext {
336 federation_id: self.federation_id,
337 gateway_conn: self.gateway_conn.clone(),
338 client_ctx: self.client_ctx.clone(),
339 }
340 }
341
342 fn input_fee(
343 &self,
344 amounts: &Amounts,
345 _input: &<Self::Common as ModuleCommon>::Input,
346 ) -> Option<Amounts> {
347 Some(Amounts::new_bitcoin(
348 self.cfg.fee_consensus.fee(amounts.expect_only_bitcoin()),
349 ))
350 }
351
352 fn output_fee(
353 &self,
354 amounts: &Amounts,
355 _output: &<Self::Common as ModuleCommon>::Output,
356 ) -> Option<Amounts> {
357 Some(Amounts::new_bitcoin(
358 self.cfg.fee_consensus.fee(amounts.expect_only_bitcoin()),
359 ))
360 }
361
362 #[cfg(feature = "cli")]
363 async fn handle_cli_command(
364 &self,
365 args: &[std::ffi::OsString],
366 ) -> anyhow::Result<serde_json::Value> {
367 cli::handle_cli_command(self, args).await
368 }
369}
370
371impl LightningClientModule {
372 #[allow(clippy::too_many_arguments)]
373 fn new(
374 federation_id: FederationId,
375 cfg: LightningClientConfig,
376 notifier: ModuleNotifier<LightningClientStateMachines>,
377 client_ctx: ClientContext<Self>,
378 module_api: DynModuleApi,
379 module_root_secret: &DerivableSecret,
380 gateway_conn: Arc<dyn GatewayConnection + Send + Sync>,
381 custom_meta_fn: Arc<dyn Fn() -> Value + Send + Sync>,
382 admin_auth: Option<ApiAuth>,
383 task_group: &TaskGroup,
384 client_span: &tracing::Span,
385 ) -> Self {
386 let module = Self {
387 federation_id,
388 cfg,
389 notifier,
390 client_ctx,
391 module_api,
392 keypair: module_root_secret
393 .child_key(ChildId(0))
394 .to_secp_key(SECP256K1),
395 lnurl_keypair: module_root_secret
396 .child_key(ChildId(1))
397 .to_secp_key(SECP256K1),
398 gateway_conn,
399 admin_auth,
400 };
401
402 module.spawn_receive_lnurl_task(custom_meta_fn, task_group, client_span);
403
404 module.spawn_gateway_map_update_task(task_group, client_span);
405
406 module
407 }
408
409 fn spawn_gateway_map_update_task(&self, task_group: &TaskGroup, client_span: &tracing::Span) {
410 let module = self.clone();
411 let api = self.module_api.clone();
412
413 task_group.spawn_cancellable_with_span(
414 client_span.clone(),
415 "gateway_map_update_task",
416 async move {
417 api.wait_for_initialized_connections().await;
418 module.update_gateway_map().await;
419 },
420 );
421 }
422
423 async fn update_gateway_map(&self) {
424 if let Ok(gateways) = self.module_api.gateways().await {
431 let mut dbtx = self.client_ctx.module_db().begin_transaction().await;
432
433 for gateway in gateways {
434 if let Ok(Some(routing_info)) = self
435 .gateway_conn
436 .routing_info(gateway.clone(), &self.federation_id)
437 .await
438 {
439 dbtx.insert_entry(&GatewayKey(routing_info.lightning_public_key), &gateway)
440 .await;
441 }
442 }
443
444 if let Err(e) = dbtx.commit_tx_result().await {
445 warn!("Failed to commit the updated gateway mapping to the database: {e}");
446 }
447 }
448 }
449
450 pub async fn select_gateway(
455 &self,
456 invoice: Option<Bolt11Invoice>,
457 ) -> Result<(SafeUrl, RoutingInfo), SelectGatewayError> {
458 let gateways = self
459 .module_api
460 .gateways()
461 .await
462 .map_err(|e| SelectGatewayError::FailedToRequestGateways(e.to_string()))?;
463
464 if gateways.is_empty() {
465 return Err(SelectGatewayError::NoGatewaysAvailable);
466 }
467
468 if let Some(invoice) = invoice
469 && let Some(gateway) = self
470 .client_ctx
471 .module_db()
472 .begin_transaction_nc()
473 .await
474 .get_value(&GatewayKey(invoice.recover_payee_pub_key()))
475 .await
476 .filter(|gateway| gateways.contains(gateway))
477 && let Ok(Some(routing_info)) = self.routing_info(&gateway).await
478 {
479 return Ok((gateway, routing_info));
480 }
481
482 for gateway in gateways {
483 if let Ok(Some(routing_info)) = self.routing_info(&gateway).await {
484 return Ok((gateway, routing_info));
485 }
486 }
487
488 Err(SelectGatewayError::GatewaysUnresponsive)
489 }
490
491 pub async fn list_gateways(
494 &self,
495 peer: Option<PeerId>,
496 ) -> Result<Vec<SafeUrl>, ListGatewaysError> {
497 if let Some(peer) = peer {
498 self.module_api
499 .gateways_from_peer(peer)
500 .await
501 .map_err(|_| ListGatewaysError::FailedToListGateways)
502 } else {
503 self.module_api
504 .gateways()
505 .await
506 .map_err(|_| ListGatewaysError::FailedToListGateways)
507 }
508 }
509
510 pub async fn routing_info(
513 &self,
514 gateway: &SafeUrl,
515 ) -> Result<Option<RoutingInfo>, RoutingInfoError> {
516 self.gateway_conn
517 .routing_info(gateway.clone(), &self.federation_id)
518 .await
519 .map_err(|_| RoutingInfoError::FailedToRequestRoutingInfo)
520 }
521
522 #[allow(clippy::too_many_lines)]
539 pub async fn send(
540 &self,
541 invoice: Bolt11Invoice,
542 gateway: Option<SafeUrl>,
543 custom_meta: Value,
544 ) -> Result<OperationId, SendPaymentError> {
545 let amount = invoice
546 .amount_milli_satoshis()
547 .ok_or(SendPaymentError::InvoiceMissingAmount)?;
548
549 if invoice.is_expired() {
550 return Err(SendPaymentError::InvoiceExpired);
551 }
552
553 if self.cfg.network != invoice.currency().into() {
554 return Err(SendPaymentError::WrongCurrency {
555 invoice_currency: invoice.currency(),
556 federation_currency: self.cfg.network.into(),
557 });
558 }
559
560 let operation_id = OperationId::from_encodable(&(invoice.clone(), 0u64));
564
565 if self.client_ctx.operation_exists(operation_id).await {
566 return Err(SendPaymentError::DuplicatePaymentAttempt(operation_id));
567 }
568
569 let (ephemeral_tweak, ephemeral_pk) = tweak::generate(self.keypair.public_key());
570
571 let refund_keypair = SecretKey::from_slice(&ephemeral_tweak)
572 .expect("32 bytes, within curve order")
573 .keypair(secp256k1::SECP256K1);
574
575 let (gateway_api, routing_info) = match gateway {
576 Some(gateway_api) => (
577 gateway_api.clone(),
578 self.routing_info(&gateway_api)
579 .await
580 .map_err(|e| SendPaymentError::FailedToConnectToGateway(e.to_string()))?
581 .ok_or(SendPaymentError::FederationNotSupported)?,
582 ),
583 None => self
584 .select_gateway(Some(invoice.clone()))
585 .await
586 .map_err(SendPaymentError::SelectGateway)?,
587 };
588
589 let (send_fee, expiration_delta) = routing_info.send_parameters(&invoice);
590
591 if !send_fee.is_within(&PaymentFee::SEND_FEE_LIMIT) {
592 return Err(SendPaymentError::GatewayFeeExceedsLimit);
593 }
594
595 if EXPIRATION_DELTA_LIMIT < expiration_delta {
596 return Err(SendPaymentError::GatewayExpirationExceedsLimit);
597 }
598
599 let consensus_block_count = self
600 .module_api
601 .consensus_block_count()
602 .await
603 .map_err(|e| SendPaymentError::FailedToRequestBlockCount(e.to_string()))?;
604
605 let contract = OutgoingContract {
606 payment_image: PaymentImage::Hash(*invoice.payment_hash()),
607 amount: send_fee.add_to(amount),
608 expiration: consensus_block_count + expiration_delta + CONTRACT_CONFIRMATION_BUFFER,
609 claim_pk: routing_info.module_public_key,
610 refund_pk: refund_keypair.public_key(),
611 ephemeral_pk,
612 };
613
614 let contract_clone = contract.clone();
615 let gateway_api_clone = gateway_api.clone();
616 let invoice_clone = invoice.clone();
617
618 let client_output = ClientOutput::<LightningOutput> {
619 output: LightningOutput::V0(LightningOutputV0::Outgoing(contract.clone())),
620 amounts: Amounts::new_bitcoin(contract.amount),
621 };
622
623 let client_output_sm = ClientOutputSM::<LightningClientStateMachines> {
624 state_machines: Arc::new(move |range: OutPointRange| {
625 vec![LightningClientStateMachines::Send(SendStateMachine {
626 common: SendSMCommon {
627 operation_id,
628 outpoint: range.into_iter().next().unwrap(),
629 contract: contract_clone.clone(),
630 gateway_api: Some(gateway_api_clone.clone()),
631 invoice: Some(LightningInvoice::Bolt11(invoice_clone.clone())),
632 refund_keypair,
633 },
634 state: SendSMState::Funding,
635 })]
636 }),
637 };
638
639 let client_output = self.client_ctx.make_client_outputs(ClientOutputBundle::new(
640 vec![client_output],
641 vec![client_output_sm],
642 ));
643
644 let transaction = TransactionBuilder::new().with_outputs(client_output);
645
646 self.client_ctx
647 .finalize_and_submit_transaction(
648 operation_id,
649 LightningCommonInit::KIND.as_str(),
650 move |change_outpoint_range| {
651 LightningOperationMeta::Send(SendOperationMeta {
652 change_outpoint_range,
653 gateway: gateway_api.clone(),
654 contract: contract.clone(),
655 invoice: LightningInvoice::Bolt11(invoice.clone()),
656 custom_meta: custom_meta.clone(),
657 })
658 },
659 transaction,
660 )
661 .await
662 .map_err(|e| SendPaymentError::FailedToFundPayment(e.to_string()))?;
663
664 let mut dbtx = self.client_ctx.module_db().begin_transaction().await;
665
666 self.client_ctx
667 .log_event(
668 &mut dbtx,
669 SendPaymentEvent {
670 operation_id,
671 amount: Amount::from_msats(amount),
672 fee: send_fee.fee(amount),
673 },
674 )
675 .await;
676
677 dbtx.commit_tx().await;
678
679 Ok(operation_id)
680 }
681
682 pub async fn get_invoice_send_status(
688 &self,
689 invoice: &Bolt11Invoice,
690 ) -> anyhow::Result<InvoiceSendStatus> {
691 let mut last_op = None;
696
697 for attempt in 0_u64.. {
698 let operation_id = OperationId::from_encodable(&(invoice.clone(), attempt));
699
700 if !self.client_ctx.operation_exists(operation_id).await {
701 break;
702 }
703
704 last_op = Some(operation_id);
705 }
706
707 let Some(operation_id) = last_op else {
708 return Ok(InvoiceSendStatus::NotAttempted);
709 };
710
711 if self.client_ctx.has_active_states(operation_id).await {
712 return Ok(InvoiceSendStatus::InFlight(operation_id));
713 }
714
715 let mut stream = self
718 .subscribe_send_operation_state_updates(operation_id)
719 .await?
720 .into_stream();
721
722 while let Some(state) = stream.next().await {
723 if let SendOperationState::Success(_) = state {
724 return Ok(InvoiceSendStatus::Succeeded(operation_id));
725 }
726 }
727
728 Ok(InvoiceSendStatus::Failed(operation_id))
729 }
730
731 pub async fn subscribe_send_operation_state_updates(
733 &self,
734 operation_id: OperationId,
735 ) -> anyhow::Result<UpdateStreamOrOutcome<SendOperationState>> {
736 let operation = self.client_ctx.get_operation(operation_id).await?;
737 let mut stream = self.notifier.subscribe(operation_id).await;
738 let client_ctx = self.client_ctx.clone();
739 let module_api = self.module_api.clone();
740
741 Ok(self.client_ctx.outcome_or_updates(&operation, operation_id, |state| match state {
742 SendOperationState::Funding
743 | SendOperationState::Funded
744 | SendOperationState::Refunding => false,
745 SendOperationState::Success(_)
746 | SendOperationState::Refunded
747 | SendOperationState::Failure => true,
748 }, move || {
749 stream! {
750 loop {
751 if let Some(LightningClientStateMachines::Send(state)) = stream.next().await {
752 match state.state {
753 SendSMState::Funding => yield SendOperationState::Funding,
754 SendSMState::Funded => yield SendOperationState::Funded,
755 SendSMState::Success(preimage) => {
756 assert!(state.common.contract.verify_preimage(&preimage));
758
759 yield SendOperationState::Success(preimage);
760 return;
761 },
762 SendSMState::Refunding(out_points) => {
763 yield SendOperationState::Refunding;
764
765 if client_ctx.await_primary_module_outputs(operation_id, out_points.clone()).await.is_ok() {
766 yield SendOperationState::Refunded;
767 return;
768 }
769
770 if let Some(preimage) = module_api.await_preimage(
774 state.common.outpoint,
775 0
776 ).await
777 && state.common.contract.verify_preimage(&preimage) {
778 yield SendOperationState::Success(preimage);
779 return;
780 }
781
782 yield SendOperationState::Failure;
783 return;
784 },
785 SendSMState::Rejected(..) => {
786 yield SendOperationState::Failure;
787 return;
788 },
789 }
790 }
791 }
792 }
793 }))
794 }
795
796 pub async fn await_final_send_operation_state(
798 &self,
799 operation_id: OperationId,
800 ) -> anyhow::Result<FinalSendOperationState> {
801 let mut stream = self
802 .subscribe_send_operation_state_updates(operation_id)
803 .await?
804 .into_stream();
805
806 let mut final_state = None;
807
808 while let Some(state) = stream.next().await {
809 match state {
810 SendOperationState::Success(preimage) => {
811 final_state = Some(FinalSendOperationState::Success(preimage));
812 }
813 SendOperationState::Refunded => {
814 final_state = Some(FinalSendOperationState::Refunded);
815 }
816 SendOperationState::Failure => final_state = Some(FinalSendOperationState::Failure),
817 _ => {}
818 }
819 }
820
821 Ok(final_state.expect("Stream contains one final state"))
822 }
823
824 pub async fn receive(
836 &self,
837 amount: Amount,
838 expiry_secs: u32,
839 description: Bolt11InvoiceDescription,
840 gateway: Option<SafeUrl>,
841 custom_meta: Value,
842 ) -> Result<(Bolt11Invoice, OperationId), ReceiveError> {
843 let (gateway, contract, invoice) = self
844 .create_contract_and_fetch_invoice(
845 self.keypair.public_key(),
846 amount,
847 expiry_secs,
848 description,
849 gateway,
850 )
851 .await?;
852
853 let operation_id = self
854 .receive_incoming_contract(
855 self.keypair.secret_key(),
856 contract.clone(),
857 LightningOperationMeta::Receive(ReceiveOperationMeta {
858 gateway,
859 contract,
860 invoice: LightningInvoice::Bolt11(invoice.clone()),
861 custom_meta,
862 }),
863 )
864 .await
865 .expect("The contract has been generated with our public key");
866
867 Ok((invoice, operation_id))
868 }
869
870 pub async fn receive_fee_quote(&self, amount: Amount) -> anyhow::Result<FeeQuote> {
885 self.client_ctx
886 .fee_quote(
887 OperationId::new_random(),
888 FeeQuoteRequest {
889 input_amount: Amounts::new_bitcoin(amount),
890 output_amount: Amounts::ZERO,
891 input_fee: Amounts::new_bitcoin(self.cfg.fee_consensus.fee(amount)),
892 output_fee: Amounts::ZERO,
893 },
894 )
895 .await
896 }
897
898 pub async fn send_fee_quote(&self, amount: Amount) -> anyhow::Result<FeeQuote> {
914 self.client_ctx
915 .fee_quote(
916 OperationId::new_random(),
917 FeeQuoteRequest {
918 input_amount: Amounts::ZERO,
919 output_amount: Amounts::new_bitcoin(amount),
920 input_fee: Amounts::ZERO,
921 output_fee: Amounts::new_bitcoin(self.cfg.fee_consensus.fee(amount)),
922 },
923 )
924 .await
925 }
926
927 pub async fn spendable_amount(
960 &self,
961 balance: Amount,
962 gateway: Option<SafeUrl>,
963 ) -> anyhow::Result<Amount> {
964 let routing_info = match gateway {
965 Some(gateway) => self
966 .routing_info(&gateway)
967 .await?
968 .ok_or_else(|| anyhow::anyhow!("Federation not supported by gateway"))?,
969 None => self.select_gateway(None).await?.1,
970 };
971
972 let send_fee = routing_info.send_fee_default;
977
978 anyhow::ensure!(
979 send_fee.is_within(&PaymentFee::SEND_FEE_LIMIT),
980 "Gateway's default send fee exceeds the limit"
981 );
982
983 max_affordable_send_amount(
984 balance,
985 Amount::from_msats(1),
986 balance,
987 |invoice_amount: Amount| send_fee.add_to(invoice_amount.msats),
988 |contract_amount: Amount| self.send_fee_quote(contract_amount),
989 )
990 .await
991 .ok_or_else(|| anyhow::anyhow!("Balance is too low to send any amount after fees"))
992 }
993
994 async fn create_contract_and_fetch_invoice(
998 &self,
999 recipient_static_pk: PublicKey,
1000 amount: Amount,
1001 expiry_secs: u32,
1002 description: Bolt11InvoiceDescription,
1003 gateway: Option<SafeUrl>,
1004 ) -> Result<(SafeUrl, IncomingContract, Bolt11Invoice), ReceiveError> {
1005 let (ephemeral_tweak, ephemeral_pk) = tweak::generate(recipient_static_pk);
1006
1007 let encryption_seed = ephemeral_tweak
1008 .consensus_hash::<sha256::Hash>()
1009 .to_byte_array();
1010
1011 let preimage = encryption_seed
1012 .consensus_hash::<sha256::Hash>()
1013 .to_byte_array();
1014
1015 let (gateway, routing_info) = match gateway {
1016 Some(gateway) => (
1017 gateway.clone(),
1018 self.routing_info(&gateway)
1019 .await
1020 .map_err(|e| ReceiveError::FailedToConnectToGateway(e.to_string()))?
1021 .ok_or(ReceiveError::FederationNotSupported)?,
1022 ),
1023 None => self
1024 .select_gateway(None)
1025 .await
1026 .map_err(ReceiveError::SelectGateway)?,
1027 };
1028
1029 if !routing_info
1030 .receive_fee
1031 .is_within(&PaymentFee::RECEIVE_FEE_LIMIT)
1032 {
1033 return Err(ReceiveError::GatewayFeeExceedsLimit);
1034 }
1035
1036 let contract_amount = routing_info.receive_fee.subtract_from(amount.msats);
1037
1038 if contract_amount < MINIMUM_INCOMING_CONTRACT_AMOUNT {
1039 return Err(ReceiveError::AmountTooSmall);
1040 }
1041
1042 let expiration = duration_since_epoch()
1043 .as_secs()
1044 .saturating_add(u64::from(expiry_secs));
1045
1046 let claim_pk = recipient_static_pk
1047 .mul_tweak(
1048 secp256k1::SECP256K1,
1049 &Scalar::from_be_bytes(ephemeral_tweak).expect("Within curve order"),
1050 )
1051 .expect("Tweak is valid");
1052
1053 let contract = IncomingContract::new(
1054 self.cfg.tpe_agg_pk,
1055 encryption_seed,
1056 preimage,
1057 PaymentImage::Hash(preimage.consensus_hash()),
1058 contract_amount,
1059 expiration,
1060 claim_pk,
1061 routing_info.module_public_key,
1062 ephemeral_pk,
1063 );
1064
1065 let invoice = self
1066 .gateway_conn
1067 .bolt11_invoice(
1068 gateway.clone(),
1069 self.federation_id,
1070 contract.clone(),
1071 amount,
1072 description,
1073 expiry_secs,
1074 )
1075 .await
1076 .map_err(|e| ReceiveError::FailedToConnectToGateway(e.to_string()))?;
1077
1078 if invoice.payment_hash() != &preimage.consensus_hash() {
1079 return Err(ReceiveError::InvalidInvoice);
1080 }
1081
1082 if invoice.amount_milli_satoshis() != Some(amount.msats) {
1083 return Err(ReceiveError::IncorrectInvoiceAmount);
1084 }
1085
1086 Ok((gateway, contract, invoice))
1087 }
1088
1089 async fn receive_incoming_contract(
1092 &self,
1093 sk: SecretKey,
1094 contract: IncomingContract,
1095 operation_meta: LightningOperationMeta,
1096 ) -> Option<OperationId> {
1097 let operation_id = OperationId::from_encodable(&contract.clone());
1098
1099 let (claim_keypair, agg_decryption_key) = self.recover_contract_keys(sk, &contract)?;
1100
1101 let receive_sm = LightningClientStateMachines::Receive(ReceiveStateMachine {
1102 common: ReceiveSMCommon {
1103 operation_id,
1104 contract: contract.clone(),
1105 claim_keypair,
1106 agg_decryption_key,
1107 },
1108 state: ReceiveSMState::Pending,
1109 });
1110
1111 self.client_ctx
1114 .manual_operation_start(
1115 operation_id,
1116 LightningCommonInit::KIND.as_str(),
1117 operation_meta,
1118 vec![self.client_ctx.make_dyn_state(receive_sm)],
1119 )
1120 .await
1121 .ok();
1122
1123 Some(operation_id)
1124 }
1125
1126 fn recover_contract_keys(
1127 &self,
1128 sk: SecretKey,
1129 contract: &IncomingContract,
1130 ) -> Option<(Keypair, AggregateDecryptionKey)> {
1131 let tweak = ecdh::SharedSecret::new(&contract.commitment.ephemeral_pk, &sk);
1132
1133 let encryption_seed = tweak
1134 .secret_bytes()
1135 .consensus_hash::<sha256::Hash>()
1136 .to_byte_array();
1137
1138 let claim_keypair = sk
1139 .mul_tweak(&Scalar::from_be_bytes(tweak.secret_bytes()).expect("Within curve order"))
1140 .expect("Tweak is valid")
1141 .keypair(secp256k1::SECP256K1);
1142
1143 if claim_keypair.public_key() != contract.commitment.claim_pk {
1144 return None; }
1146
1147 let agg_decryption_key = derive_agg_dk(&self.cfg.tpe_agg_pk, &encryption_seed);
1148
1149 if !contract.verify_agg_decryption_key(&self.cfg.tpe_agg_pk, &agg_decryption_key) {
1150 return None; }
1152
1153 contract.decrypt_preimage(&agg_decryption_key)?;
1154
1155 Some((claim_keypair, agg_decryption_key))
1156 }
1157
1158 pub async fn subscribe_receive_operation_state_updates(
1160 &self,
1161 operation_id: OperationId,
1162 ) -> anyhow::Result<UpdateStreamOrOutcome<ReceiveOperationState>> {
1163 let operation = self.client_ctx.get_operation(operation_id).await?;
1164 let mut stream = self.notifier.subscribe(operation_id).await;
1165 let client_ctx = self.client_ctx.clone();
1166
1167 Ok(self.client_ctx.outcome_or_updates(&operation, operation_id, |state| match state {
1168 ReceiveOperationState::Pending | ReceiveOperationState::Claiming => false,
1169 ReceiveOperationState::Expired
1170 | ReceiveOperationState::Claimed
1171 | ReceiveOperationState::Failure => true,
1172 }, move || {
1173 stream! {
1174 loop {
1175 if let Some(LightningClientStateMachines::Receive(state)) = stream.next().await {
1176 match state.state {
1177 ReceiveSMState::Pending => yield ReceiveOperationState::Pending,
1178 ReceiveSMState::Claiming(out_points) => {
1179 yield ReceiveOperationState::Claiming;
1180
1181 if client_ctx.await_primary_module_outputs(operation_id, out_points).await.is_ok() {
1182 yield ReceiveOperationState::Claimed;
1183 } else {
1184 yield ReceiveOperationState::Failure;
1185 }
1186 return;
1187 },
1188 ReceiveSMState::Expired => {
1189 yield ReceiveOperationState::Expired;
1190 return;
1191 }
1192 }
1193 }
1194 }
1195 }
1196 }))
1197 }
1198
1199 pub async fn await_final_receive_operation_state(
1201 &self,
1202 operation_id: OperationId,
1203 ) -> anyhow::Result<FinalReceiveOperationState> {
1204 let mut stream = self
1205 .subscribe_receive_operation_state_updates(operation_id)
1206 .await?
1207 .into_stream();
1208
1209 let mut final_state = None;
1210
1211 while let Some(state) = stream.next().await {
1212 match state {
1213 ReceiveOperationState::Expired => {
1214 final_state = Some(FinalReceiveOperationState::Expired);
1215 }
1216 ReceiveOperationState::Claimed => {
1217 final_state = Some(FinalReceiveOperationState::Claimed);
1218 }
1219 ReceiveOperationState::Failure => {
1220 final_state = Some(FinalReceiveOperationState::Failure);
1221 }
1222 _ => {}
1223 }
1224 }
1225
1226 Ok(final_state.expect("Stream contains one final state"))
1227 }
1228
1229 pub async fn generate_lnurl(
1232 &self,
1233 recurringd: SafeUrl,
1234 gateway: Option<SafeUrl>,
1235 ) -> Result<String, GenerateLnurlError> {
1236 let gateways = if let Some(gateway) = gateway {
1237 vec![gateway]
1238 } else {
1239 let gateways = self
1240 .module_api
1241 .gateways()
1242 .await
1243 .map_err(|e| GenerateLnurlError::FailedToRequestGateways(e.to_string()))?;
1244
1245 if gateways.is_empty() {
1246 return Err(GenerateLnurlError::NoGatewaysAvailable);
1247 }
1248
1249 gateways
1250 };
1251
1252 let payload = fedimint_core::base32::encode_prefixed(
1253 fedimint_core::base32::FEDIMINT_PREFIX,
1254 &lnurl::LnurlRequest {
1255 federation_id: self.federation_id,
1256 recipient_pk: self.lnurl_keypair.public_key(),
1257 aggregate_pk: self.cfg.tpe_agg_pk,
1258 gateways,
1259 },
1260 );
1261
1262 Ok(fedimint_lnurl::encode_lnurl(&format!(
1263 "{recurringd}pay/{payload}"
1264 )))
1265 }
1266
1267 fn spawn_receive_lnurl_task(
1268 &self,
1269 custom_meta_fn: Arc<dyn Fn() -> Value + Send + Sync>,
1270 task_group: &TaskGroup,
1271 client_span: &tracing::Span,
1272 ) {
1273 let module = self.clone();
1274 let api = self.module_api.clone();
1275
1276 task_group.spawn_cancellable_with_span(
1277 client_span.clone(),
1278 "receive_lnurl_task",
1279 async move {
1280 api.wait_for_initialized_connections().await;
1281 loop {
1282 module.receive_lnurl(custom_meta_fn()).await;
1283 }
1284 },
1285 );
1286 }
1287
1288 async fn receive_lnurl(&self, custom_meta: Value) {
1289 let stream_index = self
1298 .client_ctx
1299 .module_db()
1300 .begin_transaction_nc()
1301 .await
1302 .get_value(&IncomingContractStreamIndexKey)
1303 .await
1304 .unwrap_or(0);
1305
1306 let (contracts, next_index) = self
1307 .module_api
1308 .await_incoming_contracts(stream_index, 128)
1309 .await;
1310
1311 for contract in &contracts {
1312 if let Some(operation_id) = self
1313 .receive_incoming_contract(
1314 self.lnurl_keypair.secret_key(),
1315 contract.clone(),
1316 LightningOperationMeta::LnurlReceive(LnurlReceiveOperationMeta {
1317 contract: contract.clone(),
1318 custom_meta: custom_meta.clone(),
1319 }),
1320 )
1321 .await
1322 {
1323 self.await_final_receive_operation_state(operation_id)
1324 .await
1325 .ok();
1326 }
1327 }
1328
1329 let mut dbtx = self.client_ctx.module_db().begin_transaction().await;
1338
1339 dbtx.insert_entry(&IncomingContractStreamIndexKey, &next_index)
1340 .await;
1341
1342 dbtx.commit_tx().await;
1343 }
1344}
1345
1346#[derive(Error, Debug, Clone, Eq, PartialEq)]
1347pub enum SelectGatewayError {
1348 #[error("Failed to request gateways")]
1349 FailedToRequestGateways(String),
1350 #[error("No gateways are available")]
1351 NoGatewaysAvailable,
1352 #[error("All gateways failed to respond")]
1353 GatewaysUnresponsive,
1354}
1355
1356#[derive(Debug, Clone, Eq, PartialEq)]
1359pub enum InvoiceSendStatus {
1360 NotAttempted,
1362 InFlight(OperationId),
1364 Succeeded(OperationId),
1367 Failed(OperationId),
1369}
1370
1371#[derive(Error, Debug, Clone, Eq, PartialEq)]
1372pub enum SendPaymentError {
1373 #[error("Invoice is missing an amount")]
1374 InvoiceMissingAmount,
1375 #[error("Invoice has expired")]
1376 InvoiceExpired,
1377 #[error("Payment attempt is duplicate")]
1378 DuplicatePaymentAttempt(OperationId),
1379 #[error(transparent)]
1380 SelectGateway(SelectGatewayError),
1381 #[error("Failed to connect to gateway")]
1382 FailedToConnectToGateway(String),
1383 #[error("Gateway does not support this federation")]
1384 FederationNotSupported,
1385 #[error("Gateway fee exceeds the allowed limit")]
1386 GatewayFeeExceedsLimit,
1387 #[error("Gateway expiration time exceeds the allowed limit")]
1388 GatewayExpirationExceedsLimit,
1389 #[error("Failed to request block count")]
1390 FailedToRequestBlockCount(String),
1391 #[error("Failed to fund the payment")]
1392 FailedToFundPayment(String),
1393 #[error("Invoice is for a different currency")]
1394 WrongCurrency {
1395 invoice_currency: Currency,
1396 federation_currency: Currency,
1397 },
1398}
1399
1400#[derive(Error, Debug, Clone, Eq, PartialEq)]
1401pub enum ReceiveError {
1402 #[error(transparent)]
1403 SelectGateway(SelectGatewayError),
1404 #[error("Failed to connect to gateway")]
1405 FailedToConnectToGateway(String),
1406 #[error("Gateway does not support this federation")]
1407 FederationNotSupported,
1408 #[error("Gateway fee exceeds the allowed limit")]
1409 GatewayFeeExceedsLimit,
1410 #[error("Amount is too small to cover fees")]
1411 AmountTooSmall,
1412 #[error("Gateway returned an invalid invoice")]
1413 InvalidInvoice,
1414 #[error("Gateway returned an invoice with incorrect amount")]
1415 IncorrectInvoiceAmount,
1416}
1417
1418#[derive(Error, Debug, Clone, Eq, PartialEq)]
1419pub enum GenerateLnurlError {
1420 #[error("No gateways are available")]
1421 NoGatewaysAvailable,
1422 #[error("Failed to request gateways")]
1423 FailedToRequestGateways(String),
1424}
1425
1426#[derive(Error, Debug, Clone, Eq, PartialEq)]
1427pub enum ListGatewaysError {
1428 #[error("Failed to request gateways")]
1429 FailedToListGateways,
1430}
1431
1432#[derive(Error, Debug, Clone, Eq, PartialEq)]
1433pub enum RoutingInfoError {
1434 #[error("Failed to request routing info")]
1435 FailedToRequestRoutingInfo,
1436}
1437
1438#[derive(Debug, Clone, Eq, PartialEq, Hash, Decodable, Encodable)]
1439pub enum LightningClientStateMachines {
1440 Send(SendStateMachine),
1441 Receive(ReceiveStateMachine),
1442}
1443
1444impl IntoDynInstance for LightningClientStateMachines {
1445 type DynType = DynState;
1446
1447 fn into_dyn(self, instance_id: ModuleInstanceId) -> Self::DynType {
1448 DynState::from_typed(instance_id, self)
1449 }
1450}
1451
1452impl State for LightningClientStateMachines {
1453 type ModuleContext = LightningClientContext;
1454
1455 fn transitions(
1456 &self,
1457 context: &Self::ModuleContext,
1458 global_context: &DynGlobalClientContext,
1459 ) -> Vec<StateTransition<Self>> {
1460 match self {
1461 LightningClientStateMachines::Send(state) => {
1462 sm_enum_variant_translation!(
1463 state.transitions(context, global_context),
1464 LightningClientStateMachines::Send
1465 )
1466 }
1467 LightningClientStateMachines::Receive(state) => {
1468 sm_enum_variant_translation!(
1469 state.transitions(context, global_context),
1470 LightningClientStateMachines::Receive
1471 )
1472 }
1473 }
1474 }
1475
1476 fn operation_id(&self) -> OperationId {
1477 match self {
1478 LightningClientStateMachines::Send(state) => state.operation_id(),
1479 LightningClientStateMachines::Receive(state) => state.operation_id(),
1480 }
1481 }
1482}