Skip to main content

fedimint_lnv2_client/
lib.rs

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
73/// Number of blocks until outgoing lightning contracts times out and user
74/// client can refund it unilaterally
75const EXPIRATION_DELTA_LIMIT: u64 = 1440;
76
77/// A two hour buffer in case either the client or gateway go offline
78const 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    /// Calculate the absolute fee paid to the gateway on success.
99    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    /// Calculate the absolute fee paid to the gateway on success.
118    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/// The state of an operation sending a payment over lightning.
136///
137/// ```mermaid
138/// graph LR
139/// classDef virtual fill:#fff,stroke-dasharray: 5 5
140///
141///     Funding -- funding transaction is rejected --> Rejected
142///     Funding -- funding transaction is accepted --> Funded
143///     Funded -- payment is confirmed  --> Success
144///     Funded -- payment attempt expires --> Refunding
145///     Funded -- gateway cancels payment attempt --> Refunding
146///     Refunding -- payment is confirmed --> Success
147///     Refunding -- ecash is minted --> Refunded
148///     Refunding -- minting ecash fails --> Failure
149/// ```
150/// The transition from Refunding to Success is only possible if the gateway
151/// misbehaves.
152#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
153pub enum SendOperationState {
154    /// We are funding the contract to incentivize the gateway.
155    Funding,
156    /// We are waiting for the gateway to complete the payment.
157    Funded,
158    /// The payment was successful.
159    Success([u8; 32]),
160    /// The payment has failed and we are refunding the contract.
161    Refunding,
162    /// The payment has been refunded.
163    Refunded,
164    /// Either a programming error has occurred or the federation is malicious.
165    Failure,
166}
167
168/// The final state of an operation sending a payment over lightning.
169#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
170pub enum FinalSendOperationState {
171    /// The payment was successful. Carries the payment preimage proving
172    /// the gateway settled the invoice, serialized as a lowercase hex string.
173    Success(#[serde(with = "fedimint_core::hex::serde")] [u8; 32]),
174    /// The payment has been refunded.
175    Refunded,
176    /// Either a programming error has occurred or the federation is malicious.
177    Failure,
178}
179
180pub type SendResult = Result<OperationId, SendPaymentError>;
181
182#[cfg_attr(doc, aquamarine::aquamarine)]
183/// The state of an operation receiving a payment over lightning.
184///
185/// ```mermaid
186/// graph LR
187/// classDef virtual fill:#fff,stroke-dasharray: 5 5
188///
189///     Pending -- payment is confirmed --> Claiming
190///     Pending -- invoice expires --> Expired
191///     Claiming -- ecash is minted --> Claimed
192///     Claiming -- minting ecash fails --> Failure
193/// ```
194#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
195pub enum ReceiveOperationState {
196    /// We are waiting for the payment.
197    Pending,
198    /// The payment request has expired.
199    Expired,
200    /// The payment has been confirmed and we are issuing the ecash.
201    Claiming,
202    /// The payment has been successful.
203    Claimed,
204    /// Either a programming error has occurred or the federation is malicious.
205    Failure,
206    /// The contract is worth less than the federation charges to claim it, so
207    /// it was left unclaimed. Reachable without any payment request of ours:
208    /// anyone can address an incoming contract to a published lnurl key.
209    Uneconomical,
210}
211
212/// The final state of an operation receiving a payment over lightning.
213#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
214pub enum FinalReceiveOperationState {
215    /// The payment request has expired.
216    Expired,
217    /// The payment has been successful.
218    Claimed,
219    /// Either a programming error has occurred or the federation is malicious.
220    Failure,
221    /// The contract is worth less than the federation charges to claim it, so
222    /// it was left unclaimed.
223    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)] // The field is only used by the cli feature
330    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        // Update the mapping from lightning node public keys to gateway api
432        // endpoints maintained in the module database. When paying an invoice this
433        // enables the client to select the gateway that has created the invoice,
434        // if possible, such that the payment does not go over lightning, reducing
435        // fees and latency.
436
437        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    /// Selects an available gateway by querying the federation's registered
458    /// gateways, checking if one of them match the invoice's payee public
459    /// key, then queries the gateway for `RoutingInfo` to determine if it is
460    /// online.
461    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    /// Sends a request to each peer for their registered gateway list and
499    /// returns a `Vec<SafeUrl` of all registered gateways to the client.
500    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    /// Requests the `RoutingInfo`, including fee information, from the gateway
518    /// available at the `SafeUrl`.
519    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    /// Pay an invoice. For testing you can optionally specify a gateway to
530    /// route with, otherwise a gateway will be selected automatically. If the
531    /// invoice was created by a gateway connected to our federation, the same
532    /// gateway will be selected to allow for a direct ecash swap. Otherwise we
533    /// select a random online gateway.
534    ///
535    /// The fee for this payment may depend on the selected gateway but
536    /// will be limited to one and a half percent plus one hundred satoshis.
537    /// This fee accounts for the fee charged by the gateway as well as
538    /// the additional fee required to reliably route this payment over
539    /// lightning if necessary. Since the gateway has been vetted by at least
540    /// one guardian we trust it to set a reasonable fee and only enforce a
541    /// rather high limit.
542    ///
543    /// The absolute fee for a payment can be calculated from the operation meta
544    /// to be shown to the user in the transaction history.
545    #[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        // The attempt index is fixed at `0` so the operation id matches the one
568        // older clients derived for the first payment attempt, ensuring an
569        // already-paid or in-flight invoice is still detected after an upgrade.
570        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    /// The status of a previous [`Self::send`] for this invoice.
690    ///
691    /// Callers deciding whether an invoice is safe to pay through another
692    /// mechanism must treat anything but [`InvoiceSendStatus::NotAttempted`]
693    /// as a potential double payment.
694    pub async fn get_invoice_send_status(
695        &self,
696        invoice: &Bolt11Invoice,
697    ) -> anyhow::Result<InvoiceSendStatus> {
698        // Send only creates attempt index 0 nowadays, but older clients
699        // allocated a fresh index per retry, so scan for the latest attempt.
700        // No new attempt was ever allocated after a success, so the latest
701        // attempt alone determines the status.
702        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        // The operation is finished; replaying its (already terminated) update
723        // stream yields the cached outcome without blocking.
724        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    /// Subscribe to all state updates of the send operation.
739    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                                // the preimage has been verified by the state machine previously
764                                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                                // The gateway may have incorrectly claimed the outgoing contract thereby causing
778                                // our refund transaction to be rejected. Therefore, we check one last time if
779                                // the preimage is available before we enter the failure state.
780                                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    /// Await the final state of the send operation.
804    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    /// Request an invoice. For testing you can optionally specify a gateway to
832    /// generate the invoice, otherwise a random online gateway will be selected
833    /// automatically.
834    ///
835    /// The total fee for this payment may depend on the chosen gateway but
836    /// will be limited to half of one percent plus fifty satoshis. Since the
837    /// selected gateway has been vetted by at least one guardian we trust it to
838    /// set a reasonable fee and only enforce a rather high limit.
839    ///
840    /// The absolute fee for a payment can be calculated from the operation meta
841    /// to be shown to the user in the transaction history.
842    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    /// Computes the federation fee a `receive` of `amount` would incur, without
878    /// submitting anything.
879    ///
880    /// When the incoming contract is claimed, the client submits a transaction
881    /// with a single Lightning input worth the contract amount; the primary
882    /// module balances it by minting the change credited to the wallet. This
883    /// quotes the fee of that transaction — the Lightning input fee, the mint
884    /// output fees, and any sub-denomination dust — via the shared,
885    /// module-agnostic fee quote.
886    ///
887    /// The gateway's off-chain Lightning fee is deliberately excluded: this is
888    /// only the fee of the on-federation transaction. For that reason the quote
889    /// is taken on `amount` directly (rather than the gateway-reduced contract
890    /// amount), and no gateway round-trip is needed.
891    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    /// Whether an incoming contract worth `amount` is worth claiming, i.e.
906    /// whether claiming it would leave the wallet better off than leaving it.
907    ///
908    /// This is the computed counterpart of a fixed dust limit: it prices the
909    /// actual claim - the lightning input fee at this federation's fee
910    /// consensus, the change the primary module has to mint, and the
911    /// sub-denomination remainder that cannot be minted at all - rather than
912    /// assuming a floor. A quote that fails outright is the same verdict: the
913    /// claim spends the contract as the transaction's only input, so a fee
914    /// larger than the contract leaves a transaction that cannot be balanced.
915    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    /// Computes the federation fee a `send` funding an outgoing contract worth
923    /// `amount` would incur, without submitting anything.
924    ///
925    /// When a payment is sent, the client submits a transaction with a single
926    /// Lightning output (the outgoing contract) worth `amount`; the primary
927    /// module balances it by spending ecash to fund the contract and minting
928    /// any change. This quotes the fee of that transaction — the Lightning
929    /// output fee, the mint input fees on the funding notes, any mint change
930    /// output fees, and sub-denomination dust — via the shared, module-agnostic
931    /// fee quote.
932    ///
933    /// The gateway's off-chain Lightning fee is deliberately excluded: it is
934    /// part of the contract `amount` the gateway claims, not the on-federation
935    /// transaction fee. So `amount` is the full outgoing contract value
936    /// (`send_fee.add_to(invoice_amount)`).
937    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    /// Computes the largest invoice amount the client can pay in full out of
952    /// `balance`, i.e. the amount to request an invoice for in order to spend
953    /// (close to) the entire balance.
954    ///
955    /// Paying an invoice deducts two kinds of fee from the balance:
956    /// - the *gateway* fee, which is added on top of the invoice amount to form
957    ///   the outgoing contract (`send_fee.add_to(invoice_amount)`), and
958    /// - the *federation* fee of funding that contract — the Lightning output
959    ///   fee, the mint input fees on the funding notes, the mint output fees on
960    ///   any change, and sub-denomination dust — as quoted by
961    ///   [`Self::send_fee_quote`].
962    ///
963    /// `balance` is the client's current Bitcoin balance (e.g. from
964    /// `Client::get_balance_for_btc`). `gateway` optionally pins the gateway
965    /// whose fee schedule to use; if `None` one is selected automatically, the
966    /// same way [`Self::send`] does when no gateway is given. The gateway's
967    /// *default* send fee is used — the higher of its two send fees, applied to
968    /// a Lightning swap rather than a direct fedimint-to-fedimint swap — so the
969    /// returned amount stays payable even when the eventual invoice is routed
970    /// over Lightning.
971    ///
972    /// The maximum payable amount is found by binary search over the real fee
973    /// quote (see [`max_affordable_send_amount`]) rather than a closed form,
974    /// because the federation fee is stepwise in the amount. The quote is
975    /// point-in-time and moves with the balance, exactly like
976    /// [`Self::send_fee_quote`]; the eventual [`Self::send`] remains the source
977    /// of truth and may still fail if balance or gateway state changes in
978    /// between.
979    ///
980    /// Returns an error if the balance cannot cover even the smallest payable
981    /// amount plus fees. Any LNURL `minSendable`/`maxSendable` bounds are the
982    /// caller's responsibility to apply.
983    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        // The default (Lightning-swap) send fee is the higher of the gateway's
997        // two send fees, so using it keeps the result payable even if the
998        // eventual invoice is routed over Lightning instead of settled by a
999        // direct swap.
1000        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    /// Create an incoming contract locked to a public key derived from the
1019    /// recipient's static module public key and fetches the corresponding
1020    /// invoice.
1021    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        // Quoting the claim against this federation's fee consensus is exact, where a
1067        // fixed floor is either too permissive or too strict depending on how the
1068        // federation is configured.
1069        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    // Receive an incoming contract locked to a public key derived from our
1121    // static module public key.
1122    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        // this may only fail if the operation id is already in use, in which case we
1143        // ignore the error such that the method is idempotent
1144        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; // The claim key is not derived from our pk
1176        }
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; // The decryption key is not derived from our pk
1182        }
1183
1184        contract.decrypt_preimage(&agg_decryption_key)?;
1185
1186        Some((claim_keypair, agg_decryption_key))
1187    }
1188
1189    /// Subscribe to all state updates of the receive operation.
1190    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    /// Await the final state of the receive operation.
1236    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    /// Generate an lnurl for the client. You can optionally specify a gateway
1269    /// to use for testing purposes.
1270    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        // Read the stream cursor with a short-lived transaction. It must NOT stay open
1329        // across the long-poll below: RocksDB's optimistic transactions validate a
1330        // commit against bounded memtable history, so a transaction held open
1331        // for minutes fails with `DatabaseError::SnapshotTooOld` once enough
1332        // concurrent writes flush that history — not because anything conflicted,
1333        // but because the backend can no longer tell — and the panicking
1334        // `commit_tx()` then killed this task permanently, silently
1335        // stalling every future receive for the lifetime of the process. Long-lived
1336        // clients (daemons) hit this reproducibly under concurrent lnv2 activity.
1337        // See <https://github.com/fedimint/fedimint/issues/8872>.
1338        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            // The stream carries every incoming contract the federation funded, and
1354            // anyone can address one to a published lnurl key. Check that it is ours
1355            // before quoting - recovering the keys is local arithmetic, the quote
1356            // reads the wallet - and skip the contracts the claim fee would swallow,
1357            // so that unsolicited dust never becomes an operation in the first place.
1358            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        // Advance the cursor in its own short transaction. This is the only writer of
1393        // this key and it runs in a single sequential loop, so there is no concurrent
1394        // writer to guard against; and because this transaction is short-lived — opened
1395        // after the long-poll and committed immediately — it cannot hit the
1396        // `SnapshotTooOld` failure that a transaction held open across the long-poll
1397        // would, so a plain `commit_tx()` is safe. Ordering is unchanged: the cursor
1398        // only moves after the batch above was processed, and a crash in between
1399        // re-fetches the same batch on the next iteration.
1400        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/// The status of the latest send attempt for an invoice, derived from the
1420/// operation log.
1421#[derive(Debug, Clone, Eq, PartialEq)]
1422pub enum InvoiceSendStatus {
1423    /// No payment operation exists for this invoice.
1424    NotAttempted,
1425    /// A payment operation is still being driven to completion.
1426    InFlight(OperationId),
1427    /// The payment succeeded; paying the invoice again would be a double
1428    /// payment.
1429    Succeeded(OperationId),
1430    /// The payment failed and any funds were refunded.
1431    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}