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, MAX_INVOICE_EXPIRY_SECS, PaymentFee, RealGatewayConnection, RoutingInfo,
52};
53use fedimint_lnv2_common::{
54 Bolt11InvoiceDescription, GatewayApi, KIND, LightningCommonInit, LightningInvoice,
55 LightningModuleTypes, LightningOutput, LightningOutputV0, lnurl, tweak,
56};
57use fedimint_logging::LOG_CLIENT_MODULE_LNV2;
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 Uneconomical,
210}
211
212#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
214pub enum FinalReceiveOperationState {
215 Expired,
217 Claimed,
219 Failure,
221 Uneconomical,
224}
225
226pub type ReceiveResult = Result<(Bolt11Invoice, OperationId), ReceiveError>;
227
228#[derive(Clone)]
229pub struct LightningClientInit {
230 pub gateway_conn: Option<Arc<dyn GatewayConnection + Send + Sync>>,
231 pub custom_meta_fn: Arc<dyn Fn() -> Value + Send + Sync>,
232}
233
234impl std::fmt::Debug for LightningClientInit {
235 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
236 f.debug_struct("LightningClientInit")
237 .field("gateway_conn", &self.gateway_conn)
238 .field("custom_meta_fn", &"<function>")
239 .finish()
240 }
241}
242
243impl Default for LightningClientInit {
244 fn default() -> Self {
245 LightningClientInit {
246 gateway_conn: None,
247 custom_meta_fn: Arc::new(|| Value::Null),
248 }
249 }
250}
251
252impl ModuleInit for LightningClientInit {
253 type Common = LightningCommonInit;
254
255 async fn dump_database(
256 &self,
257 _dbtx: &mut DatabaseTransaction<'_>,
258 _prefix_names: Vec<String>,
259 ) -> Box<dyn Iterator<Item = (String, Box<dyn erased_serde::Serialize + Send>)> + '_> {
260 Box::new(BTreeMap::new().into_iter())
261 }
262}
263
264#[apply(async_trait_maybe_send!)]
265impl ClientModuleInit for LightningClientInit {
266 type Module = LightningClientModule;
267
268 fn supported_api_versions(&self) -> MultiApiVersion {
269 MultiApiVersion::try_from_iter([ApiVersion { major: 0, minor: 0 }])
270 .expect("no version conflicts")
271 }
272
273 async fn init(&self, args: &ClientModuleInitArgs<Self>) -> anyhow::Result<Self::Module> {
274 let gateway_conn = if let Some(gateway_conn) = self.gateway_conn.clone() {
275 gateway_conn
276 } else {
277 let api = GatewayApi::new(None, args.connector_registry.clone());
278 Arc::new(RealGatewayConnection { api })
279 };
280 Ok(LightningClientModule::new(
281 *args.federation_id(),
282 args.cfg().clone(),
283 args.notifier().clone(),
284 args.context(),
285 args.module_api().clone(),
286 args.module_root_secret(),
287 gateway_conn,
288 self.custom_meta_fn.clone(),
289 args.admin_auth().cloned(),
290 args.task_group(),
291 args.client_span(),
292 ))
293 }
294
295 fn used_db_prefixes(&self) -> Option<BTreeSet<u8>> {
296 Some(
297 DbKeyPrefix::iter()
298 .map(|p| p as u8)
299 .chain(
300 DbKeyPrefix::ExternalReservedStart as u8
301 ..=DbKeyPrefix::CoreInternalReservedEnd as u8,
302 )
303 .collect(),
304 )
305 }
306}
307
308#[derive(Debug, Clone)]
309pub struct LightningClientContext {
310 federation_id: FederationId,
311 gateway_conn: Arc<dyn GatewayConnection + Send + Sync>,
312 pub(crate) client_ctx: ClientContext<LightningClientModule>,
313}
314
315impl Context for LightningClientContext {
316 const KIND: Option<ModuleKind> = Some(KIND);
317}
318
319#[derive(Debug, Clone)]
320pub struct LightningClientModule {
321 federation_id: FederationId,
322 cfg: LightningClientConfig,
323 notifier: ModuleNotifier<LightningClientStateMachines>,
324 client_ctx: ClientContext<Self>,
325 module_api: DynModuleApi,
326 keypair: Keypair,
327 lnurl_keypair: Keypair,
328 gateway_conn: Arc<dyn GatewayConnection + Send + Sync>,
329 #[allow(unused)] admin_auth: Option<ApiAuth>,
331}
332
333#[apply(async_trait_maybe_send!)]
334impl ClientModule for LightningClientModule {
335 type Init = LightningClientInit;
336 type Common = LightningModuleTypes;
337 type Backup = NoModuleBackup;
338 type ModuleStateMachineContext = LightningClientContext;
339 type States = LightningClientStateMachines;
340
341 fn context(&self) -> Self::ModuleStateMachineContext {
342 LightningClientContext {
343 federation_id: self.federation_id,
344 gateway_conn: self.gateway_conn.clone(),
345 client_ctx: self.client_ctx.clone(),
346 }
347 }
348
349 fn input_fee(
350 &self,
351 amounts: &Amounts,
352 _input: &<Self::Common as ModuleCommon>::Input,
353 ) -> Option<Amounts> {
354 Some(Amounts::new_bitcoin(
355 self.cfg.fee_consensus.fee(amounts.expect_only_bitcoin()),
356 ))
357 }
358
359 fn output_fee(
360 &self,
361 amounts: &Amounts,
362 _output: &<Self::Common as ModuleCommon>::Output,
363 ) -> Option<Amounts> {
364 Some(Amounts::new_bitcoin(
365 self.cfg.fee_consensus.fee(amounts.expect_only_bitcoin()),
366 ))
367 }
368
369 #[cfg(feature = "cli")]
370 async fn handle_cli_command(
371 &self,
372 args: &[std::ffi::OsString],
373 ) -> anyhow::Result<serde_json::Value> {
374 cli::handle_cli_command(self, args).await
375 }
376}
377
378impl LightningClientModule {
379 #[allow(clippy::too_many_arguments)]
380 fn new(
381 federation_id: FederationId,
382 cfg: LightningClientConfig,
383 notifier: ModuleNotifier<LightningClientStateMachines>,
384 client_ctx: ClientContext<Self>,
385 module_api: DynModuleApi,
386 module_root_secret: &DerivableSecret,
387 gateway_conn: Arc<dyn GatewayConnection + Send + Sync>,
388 custom_meta_fn: Arc<dyn Fn() -> Value + Send + Sync>,
389 admin_auth: Option<ApiAuth>,
390 task_group: &TaskGroup,
391 client_span: &tracing::Span,
392 ) -> Self {
393 let module = Self {
394 federation_id,
395 cfg,
396 notifier,
397 client_ctx,
398 module_api,
399 keypair: module_root_secret
400 .child_key(ChildId(0))
401 .to_secp_key(SECP256K1),
402 lnurl_keypair: module_root_secret
403 .child_key(ChildId(1))
404 .to_secp_key(SECP256K1),
405 gateway_conn,
406 admin_auth,
407 };
408
409 module.spawn_receive_lnurl_task(custom_meta_fn, task_group, client_span);
410
411 module.spawn_gateway_map_update_task(task_group, client_span);
412
413 module
414 }
415
416 fn spawn_gateway_map_update_task(&self, task_group: &TaskGroup, client_span: &tracing::Span) {
417 let module = self.clone();
418 let api = self.module_api.clone();
419
420 task_group.spawn_cancellable_with_span(
421 client_span.clone(),
422 "gateway_map_update_task",
423 async move {
424 api.wait_for_initialized_connections().await;
425 module.update_gateway_map().await;
426 },
427 );
428 }
429
430 async fn update_gateway_map(&self) {
431 if let Ok(gateways) = self.module_api.gateways().await {
438 let mut dbtx = self.client_ctx.module_db().begin_transaction().await;
439
440 for gateway in gateways {
441 if let Ok(Some(routing_info)) = self
442 .gateway_conn
443 .routing_info(gateway.clone(), &self.federation_id)
444 .await
445 {
446 dbtx.insert_entry(&GatewayKey(routing_info.lightning_public_key), &gateway)
447 .await;
448 }
449 }
450
451 if let Err(e) = dbtx.commit_tx_result().await {
452 warn!("Failed to commit the updated gateway mapping to the database: {e}");
453 }
454 }
455 }
456
457 pub async fn select_gateway(
462 &self,
463 invoice: Option<Bolt11Invoice>,
464 ) -> Result<(SafeUrl, RoutingInfo), SelectGatewayError> {
465 let gateways = self
466 .module_api
467 .gateways()
468 .await
469 .map_err(|e| SelectGatewayError::FailedToRequestGateways(e.to_string()))?;
470
471 if gateways.is_empty() {
472 return Err(SelectGatewayError::NoGatewaysAvailable);
473 }
474
475 if let Some(invoice) = invoice
476 && let Some(gateway) = self
477 .client_ctx
478 .module_db()
479 .begin_transaction_nc()
480 .await
481 .get_value(&GatewayKey(invoice.recover_payee_pub_key()))
482 .await
483 .filter(|gateway| gateways.contains(gateway))
484 && let Ok(Some(routing_info)) = self.routing_info(&gateway).await
485 {
486 return Ok((gateway, routing_info));
487 }
488
489 for gateway in gateways {
490 if let Ok(Some(routing_info)) = self.routing_info(&gateway).await {
491 return Ok((gateway, routing_info));
492 }
493 }
494
495 Err(SelectGatewayError::GatewaysUnresponsive)
496 }
497
498 pub async fn list_gateways(
501 &self,
502 peer: Option<PeerId>,
503 ) -> Result<Vec<SafeUrl>, ListGatewaysError> {
504 if let Some(peer) = peer {
505 self.module_api
506 .gateways_from_peer(peer)
507 .await
508 .map_err(|_| ListGatewaysError::FailedToListGateways)
509 } else {
510 self.module_api
511 .gateways()
512 .await
513 .map_err(|_| ListGatewaysError::FailedToListGateways)
514 }
515 }
516
517 pub async fn routing_info(
520 &self,
521 gateway: &SafeUrl,
522 ) -> Result<Option<RoutingInfo>, RoutingInfoError> {
523 self.gateway_conn
524 .routing_info(gateway.clone(), &self.federation_id)
525 .await
526 .map_err(|_| RoutingInfoError::FailedToRequestRoutingInfo)
527 }
528
529 #[allow(clippy::too_many_lines)]
546 pub async fn send(
547 &self,
548 invoice: Bolt11Invoice,
549 gateway: Option<SafeUrl>,
550 custom_meta: Value,
551 ) -> Result<OperationId, SendPaymentError> {
552 let amount = invoice
553 .amount_milli_satoshis()
554 .ok_or(SendPaymentError::InvoiceMissingAmount)?;
555
556 if invoice.is_expired() {
557 return Err(SendPaymentError::InvoiceExpired);
558 }
559
560 if self.cfg.network != invoice.currency().into() {
561 return Err(SendPaymentError::WrongCurrency {
562 invoice_currency: invoice.currency(),
563 federation_currency: self.cfg.network.into(),
564 });
565 }
566
567 let operation_id = OperationId::from_encodable(&(invoice.clone(), 0u64));
571
572 if self.client_ctx.operation_exists(operation_id).await {
573 return Err(SendPaymentError::DuplicatePaymentAttempt(operation_id));
574 }
575
576 let (ephemeral_tweak, ephemeral_pk) = tweak::generate(self.keypair.public_key());
577
578 let refund_keypair = SecretKey::from_slice(&ephemeral_tweak)
579 .expect("32 bytes, within curve order")
580 .keypair(secp256k1::SECP256K1);
581
582 let (gateway_api, routing_info) = match gateway {
583 Some(gateway_api) => (
584 gateway_api.clone(),
585 self.routing_info(&gateway_api)
586 .await
587 .map_err(|e| SendPaymentError::FailedToConnectToGateway(e.to_string()))?
588 .ok_or(SendPaymentError::FederationNotSupported)?,
589 ),
590 None => self
591 .select_gateway(Some(invoice.clone()))
592 .await
593 .map_err(SendPaymentError::SelectGateway)?,
594 };
595
596 let (send_fee, expiration_delta) = routing_info.send_parameters(&invoice);
597
598 if !send_fee.is_within(&PaymentFee::SEND_FEE_LIMIT) {
599 return Err(SendPaymentError::GatewayFeeExceedsLimit);
600 }
601
602 if EXPIRATION_DELTA_LIMIT < expiration_delta {
603 return Err(SendPaymentError::GatewayExpirationExceedsLimit);
604 }
605
606 let consensus_block_count = self
607 .module_api
608 .consensus_block_count()
609 .await
610 .map_err(|e| SendPaymentError::FailedToRequestBlockCount(e.to_string()))?;
611
612 let contract = OutgoingContract {
613 payment_image: PaymentImage::Hash(*invoice.payment_hash()),
614 amount: send_fee.add_to(amount),
615 expiration: consensus_block_count + expiration_delta + CONTRACT_CONFIRMATION_BUFFER,
616 claim_pk: routing_info.module_public_key,
617 refund_pk: refund_keypair.public_key(),
618 ephemeral_pk,
619 };
620
621 let contract_clone = contract.clone();
622 let gateway_api_clone = gateway_api.clone();
623 let invoice_clone = invoice.clone();
624
625 let client_output = ClientOutput::<LightningOutput> {
626 output: LightningOutput::V0(LightningOutputV0::Outgoing(contract.clone())),
627 amounts: Amounts::new_bitcoin(contract.amount),
628 };
629
630 let client_output_sm = ClientOutputSM::<LightningClientStateMachines> {
631 state_machines: Arc::new(move |range: OutPointRange| {
632 vec![LightningClientStateMachines::Send(SendStateMachine {
633 common: SendSMCommon {
634 operation_id,
635 outpoint: range.into_iter().next().unwrap(),
636 contract: contract_clone.clone(),
637 gateway_api: Some(gateway_api_clone.clone()),
638 invoice: Some(LightningInvoice::Bolt11(invoice_clone.clone())),
639 refund_keypair,
640 },
641 state: SendSMState::Funding,
642 })]
643 }),
644 };
645
646 let client_output = self.client_ctx.make_client_outputs(ClientOutputBundle::new(
647 vec![client_output],
648 vec![client_output_sm],
649 ));
650
651 let transaction = TransactionBuilder::new().with_outputs(client_output);
652
653 self.client_ctx
654 .finalize_and_submit_transaction(
655 operation_id,
656 LightningCommonInit::KIND.as_str(),
657 move |change_outpoint_range| {
658 LightningOperationMeta::Send(SendOperationMeta {
659 change_outpoint_range,
660 gateway: gateway_api.clone(),
661 contract: contract.clone(),
662 invoice: LightningInvoice::Bolt11(invoice.clone()),
663 custom_meta: custom_meta.clone(),
664 })
665 },
666 transaction,
667 )
668 .await
669 .map_err(|e| SendPaymentError::FailedToFundPayment(e.to_string()))?;
670
671 let mut dbtx = self.client_ctx.module_db().begin_transaction().await;
672
673 self.client_ctx
674 .log_event(
675 &mut dbtx,
676 SendPaymentEvent {
677 operation_id,
678 amount: Amount::from_msats(amount),
679 fee: send_fee.fee(amount),
680 },
681 )
682 .await;
683
684 dbtx.commit_tx().await;
685
686 Ok(operation_id)
687 }
688
689 pub async fn get_invoice_send_status(
695 &self,
696 invoice: &Bolt11Invoice,
697 ) -> anyhow::Result<InvoiceSendStatus> {
698 let mut last_op = None;
703
704 for attempt in 0_u64.. {
705 let operation_id = OperationId::from_encodable(&(invoice.clone(), attempt));
706
707 if !self.client_ctx.operation_exists(operation_id).await {
708 break;
709 }
710
711 last_op = Some(operation_id);
712 }
713
714 let Some(operation_id) = last_op else {
715 return Ok(InvoiceSendStatus::NotAttempted);
716 };
717
718 if self.client_ctx.has_active_states(operation_id).await {
719 return Ok(InvoiceSendStatus::InFlight(operation_id));
720 }
721
722 let mut stream = self
725 .subscribe_send_operation_state_updates(operation_id)
726 .await?
727 .into_stream();
728
729 while let Some(state) = stream.next().await {
730 if let SendOperationState::Success(_) = state {
731 return Ok(InvoiceSendStatus::Succeeded(operation_id));
732 }
733 }
734
735 Ok(InvoiceSendStatus::Failed(operation_id))
736 }
737
738 pub async fn subscribe_send_operation_state_updates(
740 &self,
741 operation_id: OperationId,
742 ) -> anyhow::Result<UpdateStreamOrOutcome<SendOperationState>> {
743 let operation = self.client_ctx.get_operation(operation_id).await?;
744 let mut stream = self.notifier.subscribe(operation_id).await;
745 let client_ctx = self.client_ctx.clone();
746 let module_api = self.module_api.clone();
747
748 Ok(self.client_ctx.outcome_or_updates(&operation, operation_id, |state| match state {
749 SendOperationState::Funding
750 | SendOperationState::Funded
751 | SendOperationState::Refunding => false,
752 SendOperationState::Success(_)
753 | SendOperationState::Refunded
754 | SendOperationState::Failure => true,
755 }, move || {
756 stream! {
757 loop {
758 if let Some(LightningClientStateMachines::Send(state)) = stream.next().await {
759 match state.state {
760 SendSMState::Funding => yield SendOperationState::Funding,
761 SendSMState::Funded => yield SendOperationState::Funded,
762 SendSMState::Success(preimage) => {
763 assert!(state.common.contract.verify_preimage(&preimage));
765
766 yield SendOperationState::Success(preimage);
767 return;
768 },
769 SendSMState::Refunding(out_points) => {
770 yield SendOperationState::Refunding;
771
772 if client_ctx.await_primary_module_outputs(operation_id, out_points.clone()).await.is_ok() {
773 yield SendOperationState::Refunded;
774 return;
775 }
776
777 if let Some(preimage) = module_api.await_preimage(
781 state.common.outpoint,
782 0
783 ).await
784 && state.common.contract.verify_preimage(&preimage) {
785 yield SendOperationState::Success(preimage);
786 return;
787 }
788
789 yield SendOperationState::Failure;
790 return;
791 },
792 SendSMState::Rejected(..) => {
793 yield SendOperationState::Failure;
794 return;
795 },
796 }
797 }
798 }
799 }
800 }))
801 }
802
803 pub async fn await_final_send_operation_state(
805 &self,
806 operation_id: OperationId,
807 ) -> anyhow::Result<FinalSendOperationState> {
808 let mut stream = self
809 .subscribe_send_operation_state_updates(operation_id)
810 .await?
811 .into_stream();
812
813 let mut final_state = None;
814
815 while let Some(state) = stream.next().await {
816 match state {
817 SendOperationState::Success(preimage) => {
818 final_state = Some(FinalSendOperationState::Success(preimage));
819 }
820 SendOperationState::Refunded => {
821 final_state = Some(FinalSendOperationState::Refunded);
822 }
823 SendOperationState::Failure => final_state = Some(FinalSendOperationState::Failure),
824 _ => {}
825 }
826 }
827
828 Ok(final_state.expect("Stream contains one final state"))
829 }
830
831 pub async fn receive(
843 &self,
844 amount: Amount,
845 expiry_secs: u32,
846 description: Bolt11InvoiceDescription,
847 gateway: Option<SafeUrl>,
848 custom_meta: Value,
849 ) -> Result<(Bolt11Invoice, OperationId), ReceiveError> {
850 let (gateway, contract, invoice) = self
851 .create_contract_and_fetch_invoice(
852 self.keypair.public_key(),
853 amount,
854 expiry_secs,
855 description,
856 gateway,
857 )
858 .await?;
859
860 let operation_id = self
861 .receive_incoming_contract(
862 self.keypair.secret_key(),
863 contract.clone(),
864 LightningOperationMeta::Receive(ReceiveOperationMeta {
865 gateway,
866 contract,
867 invoice: LightningInvoice::Bolt11(invoice.clone()),
868 custom_meta,
869 }),
870 )
871 .await
872 .expect("The contract has been generated with our public key");
873
874 Ok((invoice, operation_id))
875 }
876
877 pub async fn receive_fee_quote(&self, amount: Amount) -> anyhow::Result<FeeQuote> {
892 self.client_ctx
893 .fee_quote(
894 OperationId::new_random(),
895 FeeQuoteRequest {
896 input_amount: Amounts::new_bitcoin(amount),
897 output_amount: Amounts::ZERO,
898 input_fee: Amounts::new_bitcoin(self.cfg.fee_consensus.fee(amount)),
899 output_fee: Amounts::ZERO,
900 },
901 )
902 .await
903 }
904
905 async fn is_worth_claiming(&self, amount: Amount) -> bool {
916 match self.receive_fee_quote(amount).await {
917 Ok(quote) => quote.total().get_bitcoin() < amount,
918 Err(_) => false,
919 }
920 }
921
922 pub async fn send_fee_quote(&self, amount: Amount) -> anyhow::Result<FeeQuote> {
938 self.client_ctx
939 .fee_quote(
940 OperationId::new_random(),
941 FeeQuoteRequest {
942 input_amount: Amounts::ZERO,
943 output_amount: Amounts::new_bitcoin(amount),
944 input_fee: Amounts::ZERO,
945 output_fee: Amounts::new_bitcoin(self.cfg.fee_consensus.fee(amount)),
946 },
947 )
948 .await
949 }
950
951 pub async fn spendable_amount(
984 &self,
985 balance: Amount,
986 gateway: Option<SafeUrl>,
987 ) -> anyhow::Result<Amount> {
988 let routing_info = match gateway {
989 Some(gateway) => self
990 .routing_info(&gateway)
991 .await?
992 .ok_or_else(|| anyhow::anyhow!("Federation not supported by gateway"))?,
993 None => self.select_gateway(None).await?.1,
994 };
995
996 let send_fee = routing_info.send_fee_default;
1001
1002 anyhow::ensure!(
1003 send_fee.is_within(&PaymentFee::SEND_FEE_LIMIT),
1004 "Gateway's default send fee exceeds the limit"
1005 );
1006
1007 max_affordable_send_amount(
1008 balance,
1009 Amount::from_msats(1),
1010 balance,
1011 |invoice_amount: Amount| send_fee.add_to(invoice_amount.msats),
1012 |contract_amount: Amount| self.send_fee_quote(contract_amount),
1013 )
1014 .await
1015 .ok_or_else(|| anyhow::anyhow!("Balance is too low to send any amount after fees"))
1016 }
1017
1018 async fn create_contract_and_fetch_invoice(
1022 &self,
1023 recipient_static_pk: PublicKey,
1024 amount: Amount,
1025 expiry_secs: u32,
1026 description: Bolt11InvoiceDescription,
1027 gateway: Option<SafeUrl>,
1028 ) -> Result<(SafeUrl, IncomingContract, Bolt11Invoice), ReceiveError> {
1029 if expiry_secs > MAX_INVOICE_EXPIRY_SECS {
1030 return Err(ReceiveError::InvoiceExpiryTooLong);
1031 }
1032
1033 let (ephemeral_tweak, ephemeral_pk) = tweak::generate(recipient_static_pk);
1034
1035 let encryption_seed = ephemeral_tweak
1036 .consensus_hash::<sha256::Hash>()
1037 .to_byte_array();
1038
1039 let preimage = encryption_seed
1040 .consensus_hash::<sha256::Hash>()
1041 .to_byte_array();
1042
1043 let (gateway, routing_info) = match gateway {
1044 Some(gateway) => (
1045 gateway.clone(),
1046 self.routing_info(&gateway)
1047 .await
1048 .map_err(|e| ReceiveError::FailedToConnectToGateway(e.to_string()))?
1049 .ok_or(ReceiveError::FederationNotSupported)?,
1050 ),
1051 None => self
1052 .select_gateway(None)
1053 .await
1054 .map_err(ReceiveError::SelectGateway)?,
1055 };
1056
1057 if !routing_info
1058 .receive_fee
1059 .is_within(&PaymentFee::RECEIVE_FEE_LIMIT)
1060 {
1061 return Err(ReceiveError::GatewayFeeExceedsLimit);
1062 }
1063
1064 let contract_amount = routing_info.receive_fee.subtract_from(amount.msats);
1065
1066 if !self.is_worth_claiming(contract_amount).await {
1070 return Err(ReceiveError::AmountTooSmall);
1071 }
1072
1073 let expiration = duration_since_epoch()
1074 .as_secs()
1075 .saturating_add(u64::from(expiry_secs));
1076
1077 let claim_pk = recipient_static_pk
1078 .mul_tweak(
1079 secp256k1::SECP256K1,
1080 &Scalar::from_be_bytes(ephemeral_tweak).expect("Within curve order"),
1081 )
1082 .expect("Tweak is valid");
1083
1084 let contract = IncomingContract::new(
1085 self.cfg.tpe_agg_pk,
1086 encryption_seed,
1087 preimage,
1088 PaymentImage::Hash(preimage.consensus_hash()),
1089 contract_amount,
1090 expiration,
1091 claim_pk,
1092 routing_info.module_public_key,
1093 ephemeral_pk,
1094 );
1095
1096 let invoice = self
1097 .gateway_conn
1098 .bolt11_invoice(
1099 gateway.clone(),
1100 self.federation_id,
1101 contract.clone(),
1102 amount,
1103 description,
1104 expiry_secs,
1105 )
1106 .await
1107 .map_err(|e| ReceiveError::FailedToConnectToGateway(e.to_string()))?;
1108
1109 if invoice.payment_hash() != &preimage.consensus_hash() {
1110 return Err(ReceiveError::InvalidInvoice);
1111 }
1112
1113 if invoice.amount_milli_satoshis() != Some(amount.msats) {
1114 return Err(ReceiveError::IncorrectInvoiceAmount);
1115 }
1116
1117 Ok((gateway, contract, invoice))
1118 }
1119
1120 async fn receive_incoming_contract(
1123 &self,
1124 sk: SecretKey,
1125 contract: IncomingContract,
1126 operation_meta: LightningOperationMeta,
1127 ) -> Option<OperationId> {
1128 let operation_id = OperationId::from_encodable(&contract.clone());
1129
1130 let (claim_keypair, agg_decryption_key) = self.recover_contract_keys(sk, &contract)?;
1131
1132 let receive_sm = LightningClientStateMachines::Receive(ReceiveStateMachine {
1133 common: ReceiveSMCommon {
1134 operation_id,
1135 contract: contract.clone(),
1136 claim_keypair,
1137 agg_decryption_key,
1138 },
1139 state: ReceiveSMState::Pending,
1140 });
1141
1142 self.client_ctx
1145 .manual_operation_start(
1146 operation_id,
1147 LightningCommonInit::KIND.as_str(),
1148 operation_meta,
1149 vec![self.client_ctx.make_dyn_state(receive_sm)],
1150 )
1151 .await
1152 .ok();
1153
1154 Some(operation_id)
1155 }
1156
1157 fn recover_contract_keys(
1158 &self,
1159 sk: SecretKey,
1160 contract: &IncomingContract,
1161 ) -> Option<(Keypair, AggregateDecryptionKey)> {
1162 let tweak = ecdh::SharedSecret::new(&contract.commitment.ephemeral_pk, &sk);
1163
1164 let encryption_seed = tweak
1165 .secret_bytes()
1166 .consensus_hash::<sha256::Hash>()
1167 .to_byte_array();
1168
1169 let claim_keypair = sk
1170 .mul_tweak(&Scalar::from_be_bytes(tweak.secret_bytes()).expect("Within curve order"))
1171 .expect("Tweak is valid")
1172 .keypair(secp256k1::SECP256K1);
1173
1174 if claim_keypair.public_key() != contract.commitment.claim_pk {
1175 return None; }
1177
1178 let agg_decryption_key = derive_agg_dk(&self.cfg.tpe_agg_pk, &encryption_seed);
1179
1180 if !contract.verify_agg_decryption_key(&self.cfg.tpe_agg_pk, &agg_decryption_key) {
1181 return None; }
1183
1184 contract.decrypt_preimage(&agg_decryption_key)?;
1185
1186 Some((claim_keypair, agg_decryption_key))
1187 }
1188
1189 pub async fn subscribe_receive_operation_state_updates(
1191 &self,
1192 operation_id: OperationId,
1193 ) -> anyhow::Result<UpdateStreamOrOutcome<ReceiveOperationState>> {
1194 let operation = self.client_ctx.get_operation(operation_id).await?;
1195 let mut stream = self.notifier.subscribe(operation_id).await;
1196 let client_ctx = self.client_ctx.clone();
1197
1198 Ok(self.client_ctx.outcome_or_updates(&operation, operation_id, |state| match state {
1199 ReceiveOperationState::Pending | ReceiveOperationState::Claiming => false,
1200 ReceiveOperationState::Expired
1201 | ReceiveOperationState::Claimed
1202 | ReceiveOperationState::Failure
1203 | ReceiveOperationState::Uneconomical => true,
1204 }, move || {
1205 stream! {
1206 loop {
1207 if let Some(LightningClientStateMachines::Receive(state)) = stream.next().await {
1208 match state.state {
1209 ReceiveSMState::Pending => yield ReceiveOperationState::Pending,
1210 ReceiveSMState::Claiming(out_points) => {
1211 yield ReceiveOperationState::Claiming;
1212
1213 if client_ctx.await_primary_module_outputs(operation_id, out_points).await.is_ok() {
1214 yield ReceiveOperationState::Claimed;
1215 } else {
1216 yield ReceiveOperationState::Failure;
1217 }
1218 return;
1219 },
1220 ReceiveSMState::Expired => {
1221 yield ReceiveOperationState::Expired;
1222 return;
1223 }
1224 ReceiveSMState::Uneconomical => {
1225 yield ReceiveOperationState::Uneconomical;
1226 return;
1227 }
1228 }
1229 }
1230 }
1231 }
1232 }))
1233 }
1234
1235 pub async fn await_final_receive_operation_state(
1237 &self,
1238 operation_id: OperationId,
1239 ) -> anyhow::Result<FinalReceiveOperationState> {
1240 let mut stream = self
1241 .subscribe_receive_operation_state_updates(operation_id)
1242 .await?
1243 .into_stream();
1244
1245 let mut final_state = None;
1246
1247 while let Some(state) = stream.next().await {
1248 match state {
1249 ReceiveOperationState::Expired => {
1250 final_state = Some(FinalReceiveOperationState::Expired);
1251 }
1252 ReceiveOperationState::Claimed => {
1253 final_state = Some(FinalReceiveOperationState::Claimed);
1254 }
1255 ReceiveOperationState::Failure => {
1256 final_state = Some(FinalReceiveOperationState::Failure);
1257 }
1258 ReceiveOperationState::Uneconomical => {
1259 final_state = Some(FinalReceiveOperationState::Uneconomical);
1260 }
1261 _ => {}
1262 }
1263 }
1264
1265 Ok(final_state.expect("Stream contains one final state"))
1266 }
1267
1268 pub async fn generate_lnurl(
1271 &self,
1272 recurringd: SafeUrl,
1273 gateway: Option<SafeUrl>,
1274 ) -> Result<String, GenerateLnurlError> {
1275 let gateways = if let Some(gateway) = gateway {
1276 vec![gateway]
1277 } else {
1278 let gateways = self
1279 .module_api
1280 .gateways()
1281 .await
1282 .map_err(|e| GenerateLnurlError::FailedToRequestGateways(e.to_string()))?;
1283
1284 if gateways.is_empty() {
1285 return Err(GenerateLnurlError::NoGatewaysAvailable);
1286 }
1287
1288 gateways
1289 };
1290
1291 let payload = fedimint_core::base32::encode_prefixed(
1292 fedimint_core::base32::FEDIMINT_PREFIX,
1293 &lnurl::LnurlRequest {
1294 federation_id: self.federation_id,
1295 recipient_pk: self.lnurl_keypair.public_key(),
1296 aggregate_pk: self.cfg.tpe_agg_pk,
1297 gateways,
1298 },
1299 );
1300
1301 Ok(fedimint_lnurl::encode_lnurl(&format!(
1302 "{recurringd}pay/{payload}"
1303 )))
1304 }
1305
1306 fn spawn_receive_lnurl_task(
1307 &self,
1308 custom_meta_fn: Arc<dyn Fn() -> Value + Send + Sync>,
1309 task_group: &TaskGroup,
1310 client_span: &tracing::Span,
1311 ) {
1312 let module = self.clone();
1313 let api = self.module_api.clone();
1314
1315 task_group.spawn_cancellable_with_span(
1316 client_span.clone(),
1317 "receive_lnurl_task",
1318 async move {
1319 api.wait_for_initialized_connections().await;
1320 loop {
1321 module.receive_lnurl(custom_meta_fn()).await;
1322 }
1323 },
1324 );
1325 }
1326
1327 async fn receive_lnurl(&self, custom_meta: Value) {
1328 let stream_index = self
1339 .client_ctx
1340 .module_db()
1341 .begin_transaction_nc()
1342 .await
1343 .get_value(&IncomingContractStreamIndexKey)
1344 .await
1345 .unwrap_or(0);
1346
1347 let (contracts, next_index) = self
1348 .module_api
1349 .await_incoming_contracts(stream_index, 128)
1350 .await;
1351
1352 for contract in &contracts {
1353 if self
1359 .recover_contract_keys(self.lnurl_keypair.secret_key(), contract)
1360 .is_none()
1361 {
1362 continue;
1363 }
1364
1365 if !self.is_worth_claiming(contract.commitment.amount).await {
1366 warn!(
1367 target: LOG_CLIENT_MODULE_LNV2,
1368 amount = %contract.commitment.amount,
1369 "Ignoring incoming contract, its amount does not cover the claim fee"
1370 );
1371
1372 continue;
1373 }
1374
1375 if let Some(operation_id) = self
1376 .receive_incoming_contract(
1377 self.lnurl_keypair.secret_key(),
1378 contract.clone(),
1379 LightningOperationMeta::LnurlReceive(LnurlReceiveOperationMeta {
1380 contract: contract.clone(),
1381 custom_meta: custom_meta.clone(),
1382 }),
1383 )
1384 .await
1385 {
1386 self.await_final_receive_operation_state(operation_id)
1387 .await
1388 .ok();
1389 }
1390 }
1391
1392 let mut dbtx = self.client_ctx.module_db().begin_transaction().await;
1401
1402 dbtx.insert_entry(&IncomingContractStreamIndexKey, &next_index)
1403 .await;
1404
1405 dbtx.commit_tx().await;
1406 }
1407}
1408
1409#[derive(Error, Debug, Clone, Eq, PartialEq)]
1410pub enum SelectGatewayError {
1411 #[error("Failed to request gateways")]
1412 FailedToRequestGateways(String),
1413 #[error("No gateways are available")]
1414 NoGatewaysAvailable,
1415 #[error("All gateways failed to respond")]
1416 GatewaysUnresponsive,
1417}
1418
1419#[derive(Debug, Clone, Eq, PartialEq)]
1422pub enum InvoiceSendStatus {
1423 NotAttempted,
1425 InFlight(OperationId),
1427 Succeeded(OperationId),
1430 Failed(OperationId),
1432}
1433
1434#[derive(Error, Debug, Clone, Eq, PartialEq)]
1435pub enum SendPaymentError {
1436 #[error("Invoice is missing an amount")]
1437 InvoiceMissingAmount,
1438 #[error("Invoice has expired")]
1439 InvoiceExpired,
1440 #[error("Payment attempt is duplicate")]
1441 DuplicatePaymentAttempt(OperationId),
1442 #[error(transparent)]
1443 SelectGateway(SelectGatewayError),
1444 #[error("Failed to connect to gateway")]
1445 FailedToConnectToGateway(String),
1446 #[error("Gateway does not support this federation")]
1447 FederationNotSupported,
1448 #[error("Gateway fee exceeds the allowed limit")]
1449 GatewayFeeExceedsLimit,
1450 #[error("Gateway expiration time exceeds the allowed limit")]
1451 GatewayExpirationExceedsLimit,
1452 #[error("Failed to request block count")]
1453 FailedToRequestBlockCount(String),
1454 #[error("Failed to fund the payment")]
1455 FailedToFundPayment(String),
1456 #[error("Invoice is for a different currency")]
1457 WrongCurrency {
1458 invoice_currency: Currency,
1459 federation_currency: Currency,
1460 },
1461}
1462
1463#[derive(Error, Debug, Clone, Eq, PartialEq)]
1464pub enum ReceiveError {
1465 #[error(transparent)]
1466 SelectGateway(SelectGatewayError),
1467 #[error("Failed to connect to gateway")]
1468 FailedToConnectToGateway(String),
1469 #[error("Gateway does not support this federation")]
1470 FederationNotSupported,
1471 #[error("Gateway fee exceeds the allowed limit")]
1472 GatewayFeeExceedsLimit,
1473 #[error("Amount is too small to cover fees")]
1474 AmountTooSmall,
1475 #[error("Gateway returned an invalid invoice")]
1476 InvalidInvoice,
1477 #[error("Gateway returned an invoice with incorrect amount")]
1478 IncorrectInvoiceAmount,
1479 #[error("Requested invoice expiry exceeds the maximum of one day")]
1480 InvoiceExpiryTooLong,
1481}
1482
1483#[derive(Error, Debug, Clone, Eq, PartialEq)]
1484pub enum GenerateLnurlError {
1485 #[error("No gateways are available")]
1486 NoGatewaysAvailable,
1487 #[error("Failed to request gateways")]
1488 FailedToRequestGateways(String),
1489}
1490
1491#[derive(Error, Debug, Clone, Eq, PartialEq)]
1492pub enum ListGatewaysError {
1493 #[error("Failed to request gateways")]
1494 FailedToListGateways,
1495}
1496
1497#[derive(Error, Debug, Clone, Eq, PartialEq)]
1498pub enum RoutingInfoError {
1499 #[error("Failed to request routing info")]
1500 FailedToRequestRoutingInfo,
1501}
1502
1503#[derive(Debug, Clone, Eq, PartialEq, Hash, Decodable, Encodable)]
1504pub enum LightningClientStateMachines {
1505 Send(SendStateMachine),
1506 Receive(ReceiveStateMachine),
1507}
1508
1509impl IntoDynInstance for LightningClientStateMachines {
1510 type DynType = DynState;
1511
1512 fn into_dyn(self, instance_id: ModuleInstanceId) -> Self::DynType {
1513 DynState::from_typed(instance_id, self)
1514 }
1515}
1516
1517impl State for LightningClientStateMachines {
1518 type ModuleContext = LightningClientContext;
1519
1520 fn transitions(
1521 &self,
1522 context: &Self::ModuleContext,
1523 global_context: &DynGlobalClientContext,
1524 ) -> Vec<StateTransition<Self>> {
1525 match self {
1526 LightningClientStateMachines::Send(state) => {
1527 sm_enum_variant_translation!(
1528 state.transitions(context, global_context),
1529 LightningClientStateMachines::Send
1530 )
1531 }
1532 LightningClientStateMachines::Receive(state) => {
1533 sm_enum_variant_translation!(
1534 state.transitions(context, global_context),
1535 LightningClientStateMachines::Receive
1536 )
1537 }
1538 }
1539 }
1540
1541 fn operation_id(&self) -> OperationId {
1542 match self {
1543 LightningClientStateMachines::Send(state) => state.operation_id(),
1544 LightningClientStateMachines::Receive(state) => state.operation_id(),
1545 }
1546 }
1547}