Skip to main content

cdk_ldk_node/
lib.rs

1//! CDK lightning backend for ldk-node
2
3#![doc = include_str!("../README.md")]
4
5use std::fmt;
6use std::net::SocketAddr;
7use std::pin::Pin;
8use std::sync::atomic::{AtomicBool, Ordering};
9use std::sync::Arc;
10use std::time::Duration;
11
12use async_trait::async_trait;
13use bip39::Mnemonic;
14use cdk_common::amount::MSAT_IN_SAT;
15use cdk_common::common::FeeReserve;
16use cdk_common::database::DynKVStore;
17use cdk_common::payment::{self, *};
18use cdk_common::redact::url_for_logs;
19use cdk_common::util::{hex, unix_time};
20use cdk_common::{Amount, CurrencyUnit, MeltOptions, MeltQuoteState, QuoteId};
21use futures::{Stream, StreamExt};
22use ldk_node::bitcoin::hashes::Hash;
23use ldk_node::bitcoin::Network;
24use ldk_node::lightning::ln::channelmanager::PaymentId;
25use ldk_node::lightning::ln::msgs::SocketAddress;
26use ldk_node::lightning::routing::router::RouteParametersConfig;
27use ldk_node::lightning_invoice::{Bolt11InvoiceDescription, Description};
28use ldk_node::lightning_types::payment::PaymentHash;
29use ldk_node::logger::{LogLevel, LogWriter};
30use ldk_node::payment::{PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus};
31use ldk_node::{Builder, Event, Node};
32use tokio_stream::wrappers::BroadcastStream;
33use tokio_util::sync::CancellationToken;
34use tracing::instrument;
35
36use crate::error::Error;
37use crate::log::StdoutLogWriter;
38
39mod error;
40mod log;
41mod web;
42
43/// Primary KV namespace for the ldk-node backend's durable bookkeeping
44const LDK_KV_PRIMARY_NAMESPACE: &str = "cdk_ldk_node_lightning_backend";
45/// Secondary KV namespace holding the bolt12 melt quote id -> payment id
46/// mapping used to resolve `PaymentIdentifier::QuoteId` lookups
47const LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE: &str = "bolt12_outgoing_payments";
48/// Maximum time a synchronous payment request waits for an LDK terminal event
49const PAYMENT_WAIT_TIMEOUT: Duration = Duration::from_secs(10);
50/// Capacity for terminal outgoing payment notifications
51const PAYMENT_EVENT_CHANNEL_CAPACITY: usize = 64;
52const LDK_KV_BOLT12_CLEANUP_MARKER: &[u8] = b"cleanup-in-progress";
53
54/// Result of looking up the payment id recorded for a bolt12 melt quote
55#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56enum Bolt12QuotePaymentIdLookup {
57    /// A payment id was recorded: the payment was dispatched and is tracked
58    Found(PaymentId),
59    /// The dispatch sentinel is present: `send` was started but no payment id
60    /// was recorded (crash during dispatch, or dispatch errored before the
61    /// sentinel could be cleaned up). The payment state is indeterminate.
62    Dispatching,
63    /// No record exists: the payment was never dispatched
64    Missing,
65    /// A record exists but cannot be parsed
66    Malformed,
67}
68
69#[derive(Debug, Clone, Copy, PartialEq, Eq)]
70enum Bolt12QuotePaymentIdResolution {
71    PaymentId(PaymentId),
72    Status(MeltQuoteState),
73}
74
75impl Bolt12QuotePaymentIdLookup {
76    fn resolve(self) -> Bolt12QuotePaymentIdResolution {
77        match self {
78            Self::Found(payment_id) => Bolt12QuotePaymentIdResolution::PaymentId(payment_id),
79            // Dispatch was attempted but no payment id was recorded. Pending
80            // prevents the live melt saga from compensating an indeterminate
81            // payment after a dispatch-ambiguous error.
82            Self::Dispatching => Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Pending),
83            // Without the pre-dispatch sentinel, the payment was never sent.
84            Self::Missing => Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Unpaid),
85            // Corrupt bookkeeping cannot establish any payment state.
86            Self::Malformed => Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Unknown),
87        }
88    }
89}
90
91/// Whether an LDK BOLT12 send error can occur after dispatch was accepted.
92///
93/// In `ldk-node` 0.7, [`ldk_node::NodeError::PersistenceFailed`] can be returned
94/// while persisting the payment record after `ChannelManager::pay_for_offer`
95/// accepted the payment. All other errors returned by the BOLT12 send methods
96/// occur before dispatch or after `pay_for_offer` rejected the attempt.
97fn bolt12_send_error_has_ambiguous_dispatch(err: &ldk_node::NodeError) -> bool {
98    matches!(err, ldk_node::NodeError::PersistenceFailed)
99}
100
101/// Whether a BOLT11 send error authoritatively proves that this invocation
102/// cannot settle.
103///
104/// `PersistenceFailed` is ambiguous because `ldk-node` may return it after the
105/// channel manager accepted the payment but before the pending payment record
106/// was persisted. `DuplicatePayment` is also non-terminal for this invocation:
107/// the existing payment may already be pending or succeeded. The remaining
108/// errors listed here are returned only when dispatch was rejected.
109fn bolt11_send_error_is_explicit_terminal_failure(err: &ldk_node::NodeError) -> bool {
110    matches!(
111        err,
112        ldk_node::NodeError::NotRunning
113            | ldk_node::NodeError::InvalidAmount
114            | ldk_node::NodeError::InvalidInvoice
115            | ldk_node::NodeError::PaymentSendingFailed
116    )
117}
118
119fn outgoing_payment_failure_response(
120    unit: &CurrencyUnit,
121    payment_lookup_id: PaymentIdentifier,
122) -> MakePaymentResponse {
123    MakePaymentResponse {
124        payment_lookup_id,
125        payment_proof: None,
126        status: MeltQuoteState::Failed,
127        total_spent: Amount::new(0, unit.clone()),
128    }
129}
130
131/// CDK Lightning backend using LDK Node
132///
133/// Provides Lightning Network functionality for CDK with support for Cashu operations.
134/// Handles payment creation, processing, and event management using the Lightning Development Kit.
135#[derive(Clone)]
136pub struct CdkLdkNode {
137    inner: Arc<Node>,
138    fee_reserve: FeeReserve,
139    kv_store: DynKVStore,
140    wait_invoice_cancel_token: CancellationToken,
141    wait_invoice_is_active: Arc<AtomicBool>,
142    sender: tokio::sync::broadcast::Sender<WaitPaymentResponse>,
143    receiver: Arc<tokio::sync::broadcast::Receiver<WaitPaymentResponse>>,
144    outgoing_payment_sender: tokio::sync::broadcast::Sender<PaymentId>,
145    events_cancel_token: CancellationToken,
146    web_addr: Option<SocketAddr>,
147}
148
149impl fmt::Debug for CdkLdkNode {
150    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
151        f.debug_struct("CdkLdkNode")
152            .field("fee_reserve", &self.fee_reserve)
153            .field("web_addr", &self.web_addr)
154            .finish_non_exhaustive()
155    }
156}
157
158/// Configuration for connecting to Bitcoin RPC
159///
160/// Contains the necessary connection parameters for Bitcoin Core RPC interface.
161#[derive(Clone)]
162pub struct BitcoinRpcConfig {
163    /// Bitcoin RPC server hostname or IP address
164    pub host: String,
165    /// Bitcoin RPC server port number
166    pub port: u16,
167    /// Username for Bitcoin RPC authentication
168    pub user: String,
169    /// Password for Bitcoin RPC authentication
170    pub password: String,
171}
172
173impl fmt::Debug for BitcoinRpcConfig {
174    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
175        f.debug_struct("BitcoinRpcConfig")
176            .field("host", &self.host)
177            .field("port", &self.port)
178            .field("user", &self.user)
179            .field("password", &"[REDACTED]")
180            .finish()
181    }
182}
183
184/// Source of blockchain data for the Lightning node
185///
186/// Specifies how the node should connect to the Bitcoin network to retrieve
187/// blockchain information and broadcast transactions.
188#[derive(Clone)]
189pub enum ChainSource {
190    /// Use an Esplora server for blockchain data
191    ///
192    /// Contains the URL of the Esplora server endpoint
193    Esplora(String),
194    /// Use an Electrum server for blockchain data
195    ///
196    /// Contains the URL of the Electrum server endpoint
197    Electrum(String),
198    /// Use Bitcoin Core RPC for blockchain data
199    ///
200    /// Contains the configuration for connecting to Bitcoin Core
201    BitcoinRpc(BitcoinRpcConfig),
202}
203
204impl fmt::Debug for ChainSource {
205    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
206        match self {
207            Self::Esplora(url) => f.debug_tuple("Esplora").field(&url_for_logs(url)).finish(),
208            Self::Electrum(url) => f.debug_tuple("Electrum").field(&url_for_logs(url)).finish(),
209            Self::BitcoinRpc(config) => f.debug_tuple("BitcoinRpc").field(config).finish(),
210        }
211    }
212}
213
214/// Source of Lightning network gossip data
215///
216/// Specifies how the node should learn about the Lightning Network topology
217/// and routing information.
218#[derive(Clone)]
219pub enum GossipSource {
220    /// Learn gossip through peer-to-peer connections
221    ///
222    /// The node will connect to other Lightning nodes and exchange gossip data directly
223    P2P,
224    /// Use Rapid Gossip Sync for efficient gossip updates
225    ///
226    /// Contains the URL of the RGS server for compressed gossip data
227    RapidGossipSync(String),
228}
229
230impl fmt::Debug for GossipSource {
231    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
232        match self {
233            Self::P2P => f.write_str("P2P"),
234            Self::RapidGossipSync(url) => f
235                .debug_tuple("RapidGossipSync")
236                .field(&url_for_logs(url))
237                .finish(),
238        }
239    }
240}
241/// A builder for an [`CdkLdkNode`] instance.
242pub struct CdkLdkNodeBuilder {
243    network: Network,
244    chain_source: ChainSource,
245    gossip_source: GossipSource,
246    log_dir_path: Option<String>,
247    storage_dir_path: String,
248    fee_reserve: FeeReserve,
249    kv_store: DynKVStore,
250    listening_addresses: Vec<SocketAddress>,
251    seed: Option<Mnemonic>,
252    announcement_addresses: Option<Vec<SocketAddress>>,
253}
254
255impl std::fmt::Debug for CdkLdkNodeBuilder {
256    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
257        f.debug_struct("CdkLdkNodeBuilder")
258            .field("network", &self.network)
259            .field("chain_source", &self.chain_source)
260            .field("gossip_source", &self.gossip_source)
261            .field("log_dir_path", &self.log_dir_path)
262            .field("storage_dir_path", &self.storage_dir_path)
263            .field("fee_reserve", &self.fee_reserve)
264            .field("listening_addresses", &self.listening_addresses)
265            .field("announcement_addresses", &self.announcement_addresses)
266            .finish_non_exhaustive()
267    }
268}
269
270impl CdkLdkNodeBuilder {
271    /// Creates a new builder instance.
272    pub fn new(
273        network: Network,
274        chain_source: ChainSource,
275        gossip_source: GossipSource,
276        storage_dir_path: String,
277        fee_reserve: FeeReserve,
278        listening_addresses: Vec<SocketAddress>,
279        kv_store: DynKVStore,
280    ) -> Self {
281        Self {
282            network,
283            chain_source,
284            gossip_source,
285            storage_dir_path,
286            fee_reserve,
287            kv_store,
288            listening_addresses,
289            seed: None,
290            announcement_addresses: None,
291            log_dir_path: None,
292        }
293    }
294
295    /// Configures the [`CdkLdkNode`] to use the Mnemonic for entropy source configuration
296    pub fn with_seed(mut self, seed: Mnemonic) -> Self {
297        self.seed = Some(seed);
298        self
299    }
300    /// Configures the [`CdkLdkNode`] to use announce this address to the lightning network
301    pub fn with_announcement_address(mut self, announcement_addresses: Vec<SocketAddress>) -> Self {
302        self.announcement_addresses = Some(announcement_addresses);
303        self
304    }
305    /// Configures the [`CdkLdkNode`] to use announce this address to the lightning network
306    pub fn with_log_dir_path(mut self, log_dir_path: String) -> Self {
307        self.log_dir_path = Some(log_dir_path);
308        self
309    }
310
311    /// Builds the [`CdkLdkNode`] instance
312    ///
313    /// # Errors
314    /// Returns an error if the LDK node builder fails to create the node
315    pub fn build(self) -> Result<CdkLdkNode, Error> {
316        let mut ldk = Builder::new();
317        ldk.set_network(self.network);
318        tracing::info!("Storage dir of node is {}", self.storage_dir_path);
319        ldk.set_storage_dir_path(self.storage_dir_path);
320
321        match self.chain_source {
322            ChainSource::Esplora(esplora_url) => {
323                ldk.set_chain_source_esplora(esplora_url, None);
324            }
325            ChainSource::Electrum(electrum_url) => {
326                ldk.set_chain_source_electrum(electrum_url, None);
327            }
328            ChainSource::BitcoinRpc(BitcoinRpcConfig {
329                host,
330                port,
331                user,
332                password,
333            }) => {
334                ldk.set_chain_source_bitcoind_rpc(host, port, user, password);
335            }
336        }
337
338        match self.gossip_source {
339            GossipSource::P2P => {
340                ldk.set_gossip_source_p2p();
341            }
342            GossipSource::RapidGossipSync(rgs_url) => {
343                ldk.set_gossip_source_rgs(rgs_url);
344            }
345        }
346
347        ldk.set_listening_addresses(self.listening_addresses)?;
348        if self.log_dir_path.is_some() {
349            ldk.set_filesystem_logger(self.log_dir_path, Some(LogLevel::Info));
350        } else {
351            ldk.set_custom_logger(Arc::new(StdoutLogWriter));
352        }
353
354        ldk.set_node_alias("cdk-ldk-node".to_string())?;
355        // set the seed as bip39 entropy mnemonic
356        if let Some(seed) = self.seed {
357            ldk.set_entropy_bip39_mnemonic(seed, None);
358        }
359        // set the announcement addresses
360        if let Some(announcement_addresses) = self.announcement_addresses {
361            ldk.set_announcement_addresses(announcement_addresses)?;
362        }
363
364        let node = ldk.build()?;
365
366        tracing::info!("Creating tokio channel for payment notifications");
367        let (sender, receiver) = tokio::sync::broadcast::channel(8);
368        let (outgoing_payment_sender, _) =
369            tokio::sync::broadcast::channel(PAYMENT_EVENT_CHANNEL_CAPACITY);
370
371        let id = node.node_id();
372
373        let adr = node.announcement_addresses();
374
375        tracing::info!(
376            "Created node {} with address {:?} on network {}",
377            id,
378            adr,
379            self.network
380        );
381
382        Ok(CdkLdkNode {
383            inner: node.into(),
384            fee_reserve: self.fee_reserve,
385            kv_store: self.kv_store,
386            wait_invoice_cancel_token: CancellationToken::new(),
387            wait_invoice_is_active: Arc::new(AtomicBool::new(false)),
388            sender,
389            receiver: Arc::new(receiver),
390            outgoing_payment_sender,
391            events_cancel_token: CancellationToken::new(),
392            web_addr: None,
393        })
394    }
395}
396
397impl CdkLdkNode {
398    /// Set the web server address for the LDK node management interface
399    ///
400    /// # Arguments
401    /// * `addr` - Socket address for the web server. If None, no web server will be started.
402    pub fn set_web_addr(&mut self, addr: Option<SocketAddr>) {
403        self.web_addr = addr;
404    }
405
406    /// Get a default web server address using an unused port
407    ///
408    /// Returns a SocketAddr with localhost and port 0, which will cause
409    /// the system to automatically assign an available port
410    pub fn default_web_addr() -> SocketAddr {
411        SocketAddr::from(([127, 0, 0, 1], 8091))
412    }
413
414    /// Best-effort release of an exact sentinel or payment-id binding that is
415    /// known not to represent an in-flight or successful payment.
416    async fn cleanup_bolt12_dispatch_binding(
417        &self,
418        quote_id: &QuoteId,
419        payment_id: Option<&PaymentId>,
420    ) {
421        match delete_bolt12_quote_payment_id_if_equals(&self.kv_store, quote_id, payment_id).await {
422            Ok(true) => {}
423            Ok(false) => {
424                tracing::debug!(
425                    quote_id = %quote_id,
426                    "BOLT12 dispatch binding changed before cleanup"
427                );
428            }
429            Err(err) => {
430                tracing::warn!(
431                    quote_id = %quote_id,
432                    "Could not release BOLT12 dispatch binding: {err}"
433                );
434            }
435        }
436    }
437
438    fn make_payment_response_from_details(
439        unit: &CurrencyUnit,
440        payment_lookup_id: PaymentIdentifier,
441        payment_details: &PaymentDetails,
442    ) -> Result<MakePaymentResponse, payment::Error> {
443        let status = match payment_details.status {
444            PaymentStatus::Pending => MeltQuoteState::Pending,
445            PaymentStatus::Succeeded => MeltQuoteState::Paid,
446            PaymentStatus::Failed => MeltQuoteState::Failed,
447        };
448
449        let payment_proof = match &payment_details.kind {
450            PaymentKind::Bolt11 { preimage, .. } => preimage.map(|p| p.to_string()),
451            PaymentKind::Bolt12Offer { preimage, .. } => preimage.map(|p| p.to_string()),
452            _ => return Err(Error::UnexpectedPaymentKind.into()),
453        };
454
455        let total_spent = if status == MeltQuoteState::Paid {
456            let total_spent = payment_details
457                .amount_msat
458                .ok_or(Error::CouldNotGetAmountSpent)?
459                + payment_details.fee_paid_msat.unwrap_or_default();
460            // Round the principal and routing fees together, only once.
461            match unit {
462                CurrencyUnit::Sat => {
463                    Amount::new(total_spent.div_ceil(MSAT_IN_SAT), CurrencyUnit::Sat)
464                }
465                _ => Amount::new(total_spent, CurrencyUnit::Msat).convert_to(unit)?,
466            }
467        } else {
468            Amount::new(0, unit.clone())
469        };
470
471        Ok(MakePaymentResponse {
472            payment_lookup_id,
473            payment_proof,
474            status,
475            total_spent,
476        })
477    }
478
479    fn select_bolt11_payment_details(
480        payment_details: impl IntoIterator<Item = PaymentDetails>,
481    ) -> Option<PaymentDetails> {
482        payment_details.into_iter().min_by_key(|details| {
483            let status_order = match details.status {
484                PaymentStatus::Succeeded => 0_u8,
485                PaymentStatus::Pending => 1,
486                PaymentStatus::Failed => 2,
487            };
488
489            (
490                status_order,
491                std::cmp::Reverse(details.latest_update_timestamp),
492            )
493        })
494    }
495
496    async fn wait_for_terminal_payment_event(
497        receiver: &mut tokio::sync::broadcast::Receiver<PaymentId>,
498        payment_id: PaymentId,
499    ) -> Result<(), tokio::sync::broadcast::error::RecvError> {
500        loop {
501            match receiver.recv().await {
502                Ok(completed_payment_id) if completed_payment_id == payment_id => return Ok(()),
503                Ok(_) => continue,
504                Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
505                    tracing::warn!(
506                        payment_id = %payment_id,
507                        skipped,
508                        "Terminal payment event receiver lagged; continuing to wait"
509                    );
510                }
511                Err(err) => return Err(err),
512            }
513        }
514    }
515
516    async fn wait_for_payment_terminal_status(
517        &self,
518        payment_id: PaymentId,
519        mut receiver: tokio::sync::broadcast::Receiver<PaymentId>,
520    ) -> Result<PaymentDetails, payment::Error> {
521        let payment_details = self
522            .inner
523            .payment(&payment_id)
524            .ok_or(Error::PaymentNotFound)?;
525
526        if payment_details.status != PaymentStatus::Pending {
527            return Ok(payment_details);
528        }
529
530        match tokio::time::timeout(
531            PAYMENT_WAIT_TIMEOUT,
532            Self::wait_for_terminal_payment_event(&mut receiver, payment_id),
533        )
534        .await
535        {
536            Ok(Ok(())) => {}
537            Ok(Err(err)) => {
538                tracing::warn!(
539                    payment_id = %payment_id,
540                    "Could not wait for terminal LDK payment event: {err}"
541                );
542            }
543            Err(_) => {
544                tracing::warn!(
545                    payment_id = %payment_id,
546                    "Payment did not reach a terminal state within {} seconds",
547                    PAYMENT_WAIT_TIMEOUT.as_secs()
548                );
549            }
550        }
551
552        let payment_details = self
553            .inner
554            .payment(&payment_id)
555            .ok_or(Error::PaymentNotFound)?;
556
557        if payment_details.status == PaymentStatus::Pending {
558            tracing::debug!(
559                payment_id = %payment_id,
560                "Payment remains pending after waiting for a terminal event"
561            );
562        }
563
564        Ok(payment_details)
565    }
566
567    /// Start the CDK LDK Node
568    ///
569    /// Starts the underlying LDK node and begins event processing.
570    /// Sets up event handlers to listen for Lightning events like payment received.
571    ///
572    /// # Returns
573    /// Returns `Ok(())` on successful start, error otherwise
574    ///
575    /// # Errors
576    /// Returns an error if the LDK node fails to start or event handling setup fails
577    pub fn start_ldk_node(&self) -> Result<(), Error> {
578        tracing::info!("Starting cdk-ldk node");
579        self.inner.start()?;
580        let node_config = self.inner.config();
581
582        tracing::info!("Starting node with network {}", node_config.network);
583
584        tracing::info!("Node status: {:?}", self.inner.status());
585
586        self.handle_events()?;
587
588        Ok(())
589    }
590
591    /// Start the web server for the LDK node management interface
592    ///
593    /// Starts a web server that provides a user interface for managing the LDK node.
594    /// The web interface allows users to view balances, manage channels, create invoices,
595    /// and send payments.
596    ///
597    /// # Arguments
598    /// * `web_addr` - The socket address to bind the web server to
599    ///
600    /// # Returns
601    /// Returns `Ok(())` on successful start, error otherwise
602    ///
603    /// # Errors
604    /// Returns an error if the web server fails to start
605    pub fn start_web_server(&self, web_addr: SocketAddr) -> Result<(), Error> {
606        let web_server = crate::web::WebServer::new(Arc::new(self.clone()));
607
608        tokio::spawn(async move {
609            if let Err(e) = web_server.serve(web_addr).await {
610                tracing::error!("Web server error: {}", e);
611            }
612        });
613
614        Ok(())
615    }
616
617    /// Stop the CDK LDK Node
618    ///
619    /// Gracefully stops the node by cancelling all active tasks and event handlers.
620    /// This includes:
621    /// - Cancelling the event handler task
622    /// - Cancelling any active wait_invoice streams
623    /// - Stopping the underlying LDK node
624    ///
625    /// # Returns
626    /// Returns `Ok(())` on successful shutdown, error otherwise
627    ///
628    /// # Errors
629    /// Returns an error if the underlying LDK node fails to stop
630    pub fn stop_ldk_node(&self) -> Result<(), Error> {
631        tracing::info!("Stopping CdkLdkNode");
632        // Cancel all tokio tasks
633        tracing::info!("Cancelling event handler");
634        self.events_cancel_token.cancel();
635
636        // Cancel any payment event streams
637        if self.is_payment_event_stream_active() {
638            tracing::info!("Cancelling payment event stream");
639            self.wait_invoice_cancel_token.cancel();
640        }
641
642        // Stop the LDK node
643        tracing::info!("Stopping LDK node");
644        self.inner.stop()?;
645        tracing::info!("CdkLdkNode stopped successfully");
646        Ok(())
647    }
648
649    /// Handle payment received event
650    async fn handle_payment_received(
651        node: &Arc<Node>,
652        sender: &tokio::sync::broadcast::Sender<WaitPaymentResponse>,
653        payment_id: Option<PaymentId>,
654        payment_hash: PaymentHash,
655        amount_msat: u64,
656    ) {
657        tracing::info!(
658            "Received payment for hash={} of amount={} msat",
659            payment_hash,
660            amount_msat
661        );
662
663        let payment_id = match payment_id {
664            Some(id) => id,
665            None => {
666                tracing::warn!("Received payment without payment_id");
667                return;
668            }
669        };
670
671        let payment_id_hex = hex::encode(payment_id.0);
672
673        if amount_msat == 0 {
674            tracing::warn!("Payment of no amount");
675            return;
676        }
677
678        tracing::info!(
679            "Processing payment notification: id={}, amount={} msats",
680            payment_id_hex,
681            amount_msat
682        );
683
684        let payment_details = match node.payment(&payment_id) {
685            Some(details) => details,
686            None => {
687                tracing::error!("Could not find payment details for id={}", payment_id_hex);
688                return;
689            }
690        };
691
692        let (payment_identifier, payment_id) = match payment_details.kind {
693            PaymentKind::Bolt11 { hash, .. } => {
694                (PaymentIdentifier::PaymentHash(hash.0), hash.to_string())
695            }
696            PaymentKind::Bolt12Offer { hash, offer_id, .. } => match hash {
697                Some(h) => (
698                    PaymentIdentifier::OfferId(offer_id.to_string()),
699                    h.to_string(),
700                ),
701                None => {
702                    tracing::error!("Bolt12 payment missing hash");
703                    return;
704                }
705            },
706            k => {
707                tracing::warn!("Received payment of kind {:?} which is not supported", k);
708                return;
709            }
710        };
711
712        let wait_payment_response = WaitPaymentResponse {
713            payment_identifier,
714            payment_amount: Amount::new(amount_msat, CurrencyUnit::Msat),
715            payment_id,
716        };
717
718        match sender.send(wait_payment_response) {
719            Ok(_) => tracing::info!("Successfully sent payment notification to stream"),
720            Err(err) => tracing::error!(
721                "Could not send payment received notification on channel: {}",
722                err
723            ),
724        }
725    }
726
727    /// Set up event handling for the node
728    pub fn handle_events(&self) -> Result<(), Error> {
729        let node = self.inner.clone();
730        let sender = self.sender.clone();
731        let outgoing_payment_sender = self.outgoing_payment_sender.clone();
732        let cancel_token = self.events_cancel_token.clone();
733
734        tracing::info!("Starting event handler task");
735
736        tokio::spawn(async move {
737            tracing::info!("Event handler loop started");
738            loop {
739                tokio::select! {
740                    _ = cancel_token.cancelled() => {
741                        tracing::info!("Event handler cancelled");
742                        break;
743                    }
744                    event = node.next_event_async() => {
745                        match event {
746                            Event::PaymentReceived {
747                                payment_id,
748                                payment_hash,
749                                amount_msat,
750                                custom_records: _
751                            } => {
752                                Self::handle_payment_received(
753                                    &node,
754                                    &sender,
755                                    payment_id,
756                                    payment_hash,
757                                    amount_msat
758                                ).await;
759                            }
760                            Event::PaymentSuccessful {
761                                payment_id,
762                                payment_hash,
763                                payment_preimage: _,
764                                fee_paid_msat: _,
765                            } => {
766                                tracing::info!(
767                                    payment_id = ?payment_id,
768                                    payment_hash = %payment_hash,
769                                    "LDK node payment succeeded"
770                                );
771                                if let Some(payment_id) = payment_id {
772                                    let _ = outgoing_payment_sender.send(payment_id);
773                                }
774                            }
775                            Event::PaymentFailed {
776                                payment_id,
777                                payment_hash,
778                                reason,
779                            } => {
780                                tracing::error!(
781                                    payment_id = ?payment_id,
782                                    payment_hash = ?payment_hash,
783                                    reason = ?reason,
784                                    "LDK node payment failed"
785                                );
786                                if let Some(payment_id) = payment_id {
787                                    let _ = outgoing_payment_sender.send(payment_id);
788                                }
789                            }
790                            event => {
791                                tracing::debug!("Received other ldk node event: {:?}", event);
792                            }
793                        }
794
795                        if let Err(err) = node.event_handled() {
796                            tracing::error!("Error handling node event: {}", err);
797                        } else {
798                            tracing::debug!("Successfully handled node event");
799                        }
800                    }
801                }
802            }
803            tracing::info!("Event handler loop terminated");
804        });
805
806        tracing::info!("Event handler task spawned");
807        Ok(())
808    }
809
810    /// Get Node used
811    pub fn node(&self) -> Arc<Node> {
812        Arc::clone(&self.inner)
813    }
814}
815
816/// Mint payment trait
817#[async_trait]
818impl MintPayment for CdkLdkNode {
819    type Err = payment::Error;
820
821    /// Start the payment processor
822    /// Starts the LDK node and begins event processing
823    async fn start(&self) -> Result<(), Self::Err> {
824        self.start_ldk_node().map_err(|e| {
825            tracing::error!("Failed to start CdkLdkNode: {}", e);
826            e
827        })?;
828
829        tracing::info!("CdkLdkNode payment processor started successfully");
830
831        // Start web server if configured
832        if let Some(web_addr) = self.web_addr {
833            tracing::info!("Starting LDK Node web interface on {}", web_addr);
834            self.start_web_server(web_addr).map_err(|e| {
835                tracing::error!("Failed to start web server: {}", e);
836                e
837            })?;
838        } else {
839            tracing::info!("No web server address configured, skipping web interface");
840        }
841
842        Ok(())
843    }
844
845    /// Stop the payment processor
846    /// Gracefully stops the LDK node and cancels all background tasks
847    async fn stop(&self) -> Result<(), Self::Err> {
848        self.stop_ldk_node().map_err(|e| {
849            tracing::error!("Failed to stop CdkLdkNode: {}", e);
850            e.into()
851        })
852    }
853
854    /// Base Settings
855    async fn get_settings(&self) -> Result<SettingsResponse, Self::Err> {
856        let settings = SettingsResponse {
857            unit: CurrencyUnit::Msat.to_string(),
858            bolt11: Some(payment::Bolt11Settings {
859                mpp: false,
860                amountless: true,
861                invoice_description: true,
862            }),
863            bolt12: Some(payment::Bolt12Settings {
864                amountless: true,
865                invoice_description: true,
866            }),
867            onchain: None,
868            custom: std::collections::HashMap::new(),
869        };
870        Ok(settings)
871    }
872
873    /// Create a new invoice
874    #[instrument(skip(self))]
875    async fn create_incoming_payment_request(
876        &self,
877        options: IncomingPaymentOptions,
878    ) -> Result<CreateIncomingPaymentResponse, Self::Err> {
879        match options {
880            IncomingPaymentOptions::Bolt11(bolt11_options) => {
881                let amount_msat: Amount = bolt11_options
882                    .amount
883                    .convert_to(&CurrencyUnit::Msat)?
884                    .into();
885                let description = bolt11_options.description.unwrap_or_default();
886                let time = match bolt11_options.unix_expiry {
887                    Some(t) => t
888                        .checked_sub(unix_time())
889                        .ok_or(payment::Error::InvalidExpiry)?,
890                    None => 36000,
891                };
892
893                let description = Bolt11InvoiceDescription::Direct(
894                    Description::new(description).map_err(|_| Error::InvalidDescription)?,
895                );
896
897                let payment = self
898                    .inner
899                    .bolt11_payment()
900                    .receive(amount_msat.into(), &description, time as u32)
901                    .map_err(Error::LdkNode)?;
902
903                let payment_hash = payment.payment_hash().to_string();
904                let payment_identifier = PaymentIdentifier::PaymentHash(
905                    hex::decode(&payment_hash)?
906                        .try_into()
907                        .map_err(|_| Error::InvalidPaymentHashLength)?,
908                );
909
910                Ok(CreateIncomingPaymentResponse {
911                    request_lookup_id: payment_identifier,
912                    request: payment.to_string(),
913                    expiry: Some(unix_time() + time),
914                    extra_json: None,
915                })
916            }
917            IncomingPaymentOptions::Bolt12(bolt12_options) => {
918                let Bolt12IncomingPaymentOptions {
919                    description,
920                    amount,
921                    unix_expiry,
922                } = *bolt12_options;
923
924                let time = unix_expiry
925                    .map(|t| {
926                        t.checked_sub(unix_time())
927                            .ok_or(payment::Error::InvalidExpiry)
928                            .map(|t| t as u32)
929                    })
930                    .transpose()?;
931
932                let offer = match amount {
933                    Some(amount) => {
934                        let amount_msat: Amount = amount.convert_to(&CurrencyUnit::Msat)?.into();
935
936                        self.inner
937                            .bolt12_payment()
938                            .receive(
939                                amount_msat.into(),
940                                &description.unwrap_or("".to_string()),
941                                time,
942                                None,
943                            )
944                            .map_err(Error::LdkNode)?
945                    }
946                    None => self
947                        .inner
948                        .bolt12_payment()
949                        .receive_variable_amount(&description.unwrap_or("".to_string()), time)
950                        .map_err(Error::LdkNode)?,
951                };
952                let payment_identifier = PaymentIdentifier::OfferId(offer.id().to_string());
953
954                Ok(CreateIncomingPaymentResponse {
955                    request_lookup_id: payment_identifier,
956                    request: offer.to_string(),
957                    expiry: unix_expiry,
958                    extra_json: None,
959                })
960            }
961            IncomingPaymentOptions::Custom(_) | IncomingPaymentOptions::Onchain(_) => {
962                Err(cdk_common::payment::Error::UnsupportedPaymentOption)
963            }
964        }
965    }
966
967    /// Get payment quote
968    /// Used to get fee and amount required for a payment request
969    #[instrument(skip_all)]
970    async fn get_payment_quote(
971        &self,
972        unit: &CurrencyUnit,
973        options: OutgoingPaymentOptions,
974    ) -> Result<PaymentQuoteResponse, Self::Err> {
975        match options {
976            cdk_common::payment::OutgoingPaymentOptions::Custom(_) => {
977                Err(cdk_common::payment::Error::UnsupportedPaymentOption)
978            }
979            OutgoingPaymentOptions::Bolt11(bolt11_options) => {
980                let bolt11 = bolt11_options.bolt11;
981
982                let amount_msat = match bolt11_options.melt_options {
983                    Some(MeltOptions::Amountless { amountless }) => {
984                        let amount_msat = amountless.amount_msat;
985
986                        if let Some(invoice_amount) = bolt11.amount_milli_satoshis() {
987                            if invoice_amount != u64::from(amount_msat) {
988                                return Err(payment::Error::AmountMismatch);
989                            }
990                        }
991
992                        amount_msat
993                    }
994                    Some(MeltOptions::Mpp { mpp }) => mpp.amount,
995                    None => bolt11
996                        .amount_milli_satoshis()
997                        .ok_or(Error::UnknownInvoiceAmount)?
998                        .into(),
999                };
1000
1001                let amount = match unit {
1002                    // The quote must cover the entire millisatoshi principal.
1003                    CurrencyUnit::Sat => Amount::new(
1004                        u64::from(amount_msat).div_ceil(MSAT_IN_SAT),
1005                        CurrencyUnit::Sat,
1006                    ),
1007                    _ => Amount::new(amount_msat.into(), CurrencyUnit::Msat).convert_to(unit)?,
1008                };
1009
1010                let relative_fee_reserve =
1011                    (self.fee_reserve.percent_fee_reserve * amount.value() as f32) as u64;
1012
1013                let absolute_fee_reserve: u64 = self.fee_reserve.min_fee_reserve.into();
1014
1015                let fee = match relative_fee_reserve > absolute_fee_reserve {
1016                    true => relative_fee_reserve,
1017                    false => absolute_fee_reserve,
1018                };
1019
1020                let payment_hash = bolt11.payment_hash().to_string();
1021                let payment_hash_bytes = hex::decode(&payment_hash)?
1022                    .try_into()
1023                    .map_err(|_| Error::InvalidPaymentHashLength)?;
1024
1025                Ok(PaymentQuoteResponse {
1026                    request_lookup_id: Some(PaymentIdentifier::PaymentHash(payment_hash_bytes)),
1027                    amount,
1028                    fee: Amount::new(fee, unit.clone()),
1029                    state: MeltQuoteState::Unpaid,
1030                    extra_json: None,
1031                    estimated_blocks: None,
1032                    fee_options: None,
1033                })
1034            }
1035            OutgoingPaymentOptions::Bolt12(bolt12_options) => {
1036                let offer = bolt12_options.offer;
1037
1038                let amount_msat = match bolt12_options.melt_options {
1039                    Some(melt_options) => melt_options.amount_msat(),
1040                    None => {
1041                        let amount = offer.amount().ok_or(payment::Error::AmountMismatch)?;
1042
1043                        match amount {
1044                            ldk_node::lightning::offers::offer::Amount::Bitcoin {
1045                                amount_msats,
1046                            } => amount_msats.into(),
1047                            _ => return Err(payment::Error::AmountMismatch),
1048                        }
1049                    }
1050                };
1051                let amount = match unit {
1052                    // The quote must cover the entire millisatoshi principal.
1053                    CurrencyUnit::Sat => Amount::new(
1054                        u64::from(amount_msat).div_ceil(MSAT_IN_SAT),
1055                        CurrencyUnit::Sat,
1056                    ),
1057                    _ => Amount::new(amount_msat.into(), CurrencyUnit::Msat).convert_to(unit)?,
1058                };
1059
1060                let relative_fee_reserve =
1061                    (self.fee_reserve.percent_fee_reserve * amount.value() as f32) as u64;
1062
1063                let absolute_fee_reserve: u64 = self.fee_reserve.min_fee_reserve.into();
1064
1065                let fee = match relative_fee_reserve > absolute_fee_reserve {
1066                    true => relative_fee_reserve,
1067                    false => absolute_fee_reserve,
1068                };
1069
1070                Ok(PaymentQuoteResponse {
1071                    request_lookup_id: Some(PaymentIdentifier::QuoteId(
1072                        bolt12_options.quote_id.clone(),
1073                    )),
1074                    amount,
1075                    fee: Amount::new(fee, unit.clone()),
1076                    state: MeltQuoteState::Unpaid,
1077                    extra_json: None,
1078                    estimated_blocks: None,
1079                    fee_options: None,
1080                })
1081            }
1082            OutgoingPaymentOptions::Onchain(_) => {
1083                Err(cdk_common::payment::Error::UnsupportedPaymentOption)
1084            }
1085        }
1086    }
1087
1088    /// Pay request
1089    #[instrument(skip(self, options))]
1090    async fn make_payment(
1091        &self,
1092        unit: &CurrencyUnit,
1093        options: OutgoingPaymentOptions,
1094    ) -> Result<MakePaymentResponse, Self::Err> {
1095        match options {
1096            cdk_common::payment::OutgoingPaymentOptions::Custom(options) => {
1097                Ok(outgoing_payment_failure_response(
1098                    unit,
1099                    PaymentIdentifier::QuoteId(options.quote_id),
1100                ))
1101            }
1102            OutgoingPaymentOptions::Bolt11(bolt11_options) => {
1103                let bolt11 = bolt11_options.bolt11;
1104                let payment_lookup_id =
1105                    PaymentIdentifier::PaymentHash(bolt11.payment_hash().to_byte_array());
1106
1107                let send_params = match bolt11_options
1108                    .max_fee_amount
1109                    .map(|f| {
1110                        f.convert_to(&CurrencyUnit::Msat)
1111                            .map(|amount_msat| RouteParametersConfig {
1112                                max_total_routing_fee_msat: Some(amount_msat.value()),
1113                                ..Default::default()
1114                            })
1115                    })
1116                    .transpose()
1117                {
1118                    Ok(params) => params,
1119                    Err(err) => {
1120                        tracing::error!("Failed to convert fee amount: {}", err);
1121                        return Ok(outgoing_payment_failure_response(unit, payment_lookup_id));
1122                    }
1123                };
1124
1125                // Subscribe before dispatch so an immediately completed
1126                // payment cannot race ahead of the waiter.
1127                let payment_event_receiver = self.outgoing_payment_sender.subscribe();
1128
1129                let payment_id = match bolt11_options.melt_options {
1130                    Some(MeltOptions::Amountless { amountless }) => {
1131                        if let Some(invoice_amount) = bolt11.amount_milli_satoshis() {
1132                            if invoice_amount != u64::from(amountless.amount_msat) {
1133                                return Ok(outgoing_payment_failure_response(
1134                                    unit,
1135                                    payment_lookup_id,
1136                                ));
1137                            }
1138                        }
1139
1140                        self.inner.bolt11_payment().send_using_amount(
1141                            &bolt11,
1142                            amountless.amount_msat.into(),
1143                            send_params,
1144                        )
1145                    }
1146                    None => self.inner.bolt11_payment().send(&bolt11, send_params),
1147                    _ => {
1148                        return Ok(outgoing_payment_failure_response(unit, payment_lookup_id));
1149                    }
1150                };
1151
1152                let payment_id = match payment_id {
1153                    Ok(payment_id) => payment_id,
1154                    Err(err) if bolt11_send_error_is_explicit_terminal_failure(&err) => {
1155                        tracing::warn!(
1156                            payment_hash = %bolt11.payment_hash(),
1157                            "LDK rejected BOLT11 payment before dispatch: {err}"
1158                        );
1159                        return Ok(outgoing_payment_failure_response(unit, payment_lookup_id));
1160                    }
1161                    Err(err) => {
1162                        tracing::warn!(
1163                            payment_hash = %bolt11.payment_hash(),
1164                            "LDK BOLT11 send outcome is indeterminate: {err}"
1165                        );
1166                        return Err(Error::LdkNode(err).into());
1167                    }
1168                };
1169
1170                let payment_details = self
1171                    .wait_for_payment_terminal_status(payment_id, payment_event_receiver)
1172                    .await?;
1173
1174                if payment_details.status == PaymentStatus::Failed {
1175                    tracing::error!(payment_id = %payment_id, "Bolt11 payment failed");
1176                }
1177
1178                Self::make_payment_response_from_details(unit, payment_lookup_id, &payment_details)
1179            }
1180            OutgoingPaymentOptions::Bolt12(bolt12_options) => {
1181                let offer = bolt12_options.offer;
1182                let quote_id = bolt12_options.quote_id.clone();
1183                let quote_payment_identifier = PaymentIdentifier::QuoteId(quote_id.clone());
1184
1185                let send_params = match bolt12_options
1186                    .max_fee_amount
1187                    .map(|f| {
1188                        f.convert_to(&CurrencyUnit::Msat)
1189                            .map(|amount_msat| RouteParametersConfig {
1190                                max_total_routing_fee_msat: Some(amount_msat.value()),
1191                                ..Default::default()
1192                            })
1193                    })
1194                    .transpose()
1195                {
1196                    Ok(params) => params,
1197                    Err(err) => {
1198                        tracing::error!("Failed to convert fee amount: {}", err);
1199                        return Ok(outgoing_payment_failure_response(
1200                            unit,
1201                            quote_payment_identifier,
1202                        ));
1203                    }
1204                };
1205
1206                // Claim the quote with an absent-only sentinel write before
1207                // attempting the send: a duplicate or concurrent dispatch for
1208                // the same quote is rejected here, before any funds move. The
1209                // sentinel also keeps a crash during dispatch distinguishable
1210                // from "never dispatched" (mirrors cdk-cln's pre-dispatch
1211                // bolt12 quote mapping). The payment must not be attempted
1212                // unless this marker is durable.
1213                if let Err(err) =
1214                    write_bolt12_quote_payment_id(&self.kv_store, &quote_id, None).await
1215                {
1216                    tracing::error!(
1217                        quote_id = %quote_id,
1218                        "Could not persist BOLT12 dispatch claim before sending: {err}"
1219                    );
1220                    if matches!(&err, Error::Bolt12QuoteAlreadyClaimed { .. }) {
1221                        // A previous invocation may still settle. Resolve its
1222                        // durable state instead of reporting a terminal failure.
1223                        let mut response = self
1224                            .check_outgoing_payment(&quote_payment_identifier)
1225                            .await?;
1226                        // Status lookups return the combined principal and fees
1227                        // in msat. Round up once, as on the initial payment path.
1228                        response.total_spent = match (response.total_spent.unit(), unit) {
1229                            (CurrencyUnit::Msat, CurrencyUnit::Sat) => Amount::new(
1230                                response.total_spent.value().div_ceil(MSAT_IN_SAT),
1231                                CurrencyUnit::Sat,
1232                            ),
1233                            _ => response.total_spent.convert_to(unit)?,
1234                        };
1235                        return Ok(response);
1236                    }
1237
1238                    return Err(err.into());
1239                }
1240
1241                // BOLT12 payment ids are assigned by `send`, so subscribe
1242                // first and filter the queued terminal events once it returns.
1243                let payment_event_receiver = self.outgoing_payment_sender.subscribe();
1244
1245                let payment_id = match bolt12_options.melt_options {
1246                    Some(MeltOptions::Amountless { amountless }) => {
1247                        self.inner.bolt12_payment().send_using_amount(
1248                            &offer,
1249                            amountless.amount_msat.into(),
1250                            None,
1251                            None,
1252                            send_params,
1253                        )
1254                    }
1255                    None => self
1256                        .inner
1257                        .bolt12_payment()
1258                        .send(&offer, None, None, send_params),
1259                    _ => {
1260                        self.cleanup_bolt12_dispatch_binding(&quote_id, None).await;
1261                        return Ok(outgoing_payment_failure_response(
1262                            unit,
1263                            quote_payment_identifier,
1264                        ));
1265                    }
1266                };
1267
1268                let payment_id = match payment_id {
1269                    Ok(payment_id) => payment_id,
1270                    Err(err) => {
1271                        match bolt12_send_error_has_ambiguous_dispatch(&err) {
1272                            true => {
1273                                tracing::warn!(
1274                                    quote_id = %quote_id,
1275                                    "LDK payment persistence failed after BOLT12 send; retaining \
1276                                     the dispatch sentinel because the payment may have been dispatched"
1277                                );
1278                            }
1279                            false => {
1280                                self.cleanup_bolt12_dispatch_binding(&quote_id, None).await;
1281                                tracing::warn!(
1282                                    quote_id = %quote_id,
1283                                    "LDK rejected BOLT12 payment before dispatch: {err}"
1284                                );
1285                                return Ok(outgoing_payment_failure_response(
1286                                    unit,
1287                                    quote_payment_identifier,
1288                                ));
1289                            }
1290                        }
1291                        return Err(Error::LdkNode(err).into());
1292                    }
1293                };
1294
1295                // Record the payment id so QuoteId lookups resolve to the
1296                // dispatched payment. The write is conditional on owning the
1297                // dispatch claim. Best-effort: if this write fails the
1298                // sentinel remains and the payment resolves as Pending, keeping
1299                // the melt proofs reserved.
1300                if let Err(err) =
1301                    write_bolt12_quote_payment_id(&self.kv_store, &quote_id, Some(&payment_id))
1302                        .await
1303                {
1304                    tracing::error!(
1305                        "Could not record BOLT12 payment id for quote {quote_id}: {err}. \
1306                         The payment will remain Pending until manual intervention."
1307                    );
1308                }
1309
1310                let payment_details = self
1311                    .wait_for_payment_terminal_status(payment_id, payment_event_receiver)
1312                    .await?;
1313
1314                if payment_details.status == PaymentStatus::Failed {
1315                    tracing::error!(
1316                        payment_id = %payment_id,
1317                        amount_msat = ?payment_details.amount_msat,
1318                        fee_paid_msat = ?payment_details.fee_paid_msat,
1319                        payment_kind = ?payment_details.kind,
1320                        "Bolt12 payment failed"
1321                    );
1322                    self.cleanup_bolt12_dispatch_binding(&quote_id, Some(&payment_id))
1323                        .await;
1324                }
1325
1326                Self::make_payment_response_from_details(
1327                    unit,
1328                    quote_payment_identifier,
1329                    &payment_details,
1330                )
1331            }
1332            OutgoingPaymentOptions::Onchain(options) => Ok(outgoing_payment_failure_response(
1333                unit,
1334                PaymentIdentifier::QuoteId(options.quote_id),
1335            )),
1336        }
1337    }
1338
1339    /// Listen for invoices to be paid to the mint
1340    /// Returns a stream of request_lookup_id once invoices are paid
1341    #[instrument(skip(self))]
1342    async fn wait_payment_event(
1343        &self,
1344    ) -> Result<Pin<Box<dyn Stream<Item = cdk_common::payment::Event> + Send>>, Self::Err> {
1345        tracing::info!("Starting stream for invoices - wait_any_incoming_payment called");
1346
1347        // Set active flag to indicate stream is active
1348        self.wait_invoice_is_active.store(true, Ordering::SeqCst);
1349        tracing::debug!("wait_invoice_is_active set to true");
1350
1351        let receiver = self.receiver.clone();
1352
1353        tracing::info!("Receiver obtained successfully, creating response stream");
1354
1355        // Transform the String stream into a WaitPaymentResponse stream
1356        let response_stream = BroadcastStream::new(receiver.resubscribe());
1357
1358        // Map the stream to handle BroadcastStreamRecvError and wrap in Event
1359        let response_stream = response_stream.filter_map(|result| async move {
1360            match result {
1361                Ok(payment) => Some(cdk_common::payment::Event::PaymentReceived(payment)),
1362                Err(err) => {
1363                    tracing::warn!("Error in broadcast stream: {}", err);
1364                    None
1365                }
1366            }
1367        });
1368
1369        // Create a combined stream that also handles cancellation
1370        let cancel_token = self.wait_invoice_cancel_token.clone();
1371        let is_active = self.wait_invoice_is_active.clone();
1372
1373        let stream = Box::pin(response_stream);
1374
1375        // Set up a task to clean up when the stream is dropped
1376        tokio::spawn(async move {
1377            cancel_token.cancelled().await;
1378            tracing::info!("wait_invoice stream cancelled");
1379            is_active.store(false, Ordering::SeqCst);
1380        });
1381
1382        tracing::info!("wait_any_incoming_payment returning stream");
1383        Ok(stream)
1384    }
1385
1386    /// Is payment event stream active
1387    fn is_payment_event_stream_active(&self) -> bool {
1388        self.wait_invoice_is_active.load(Ordering::SeqCst)
1389    }
1390
1391    /// Cancel payment event stream
1392    fn cancel_payment_event_stream(&self) {
1393        self.wait_invoice_cancel_token.cancel()
1394    }
1395
1396    /// Check the status of an incoming payment
1397    async fn check_incoming_payment_status(
1398        &self,
1399        payment_identifier: &PaymentIdentifier,
1400    ) -> Result<Vec<WaitPaymentResponse>, Self::Err> {
1401        // Bolt12 offers are identified by offer id and can be paid more than
1402        // once, so collect every settled inbound payment for the offer.
1403        if let PaymentIdentifier::OfferId(offer_id) = payment_identifier {
1404            let payments = self.inner.list_payments_with_filter(|p| {
1405                p.direction == PaymentDirection::Inbound
1406                    && p.status == PaymentStatus::Succeeded
1407                    && matches!(
1408                        &p.kind,
1409                        PaymentKind::Bolt12Offer { offer_id: oid, .. } if oid.to_string() == *offer_id
1410                    )
1411            });
1412
1413            return Ok(payments
1414                .into_iter()
1415                .filter_map(|p| {
1416                    let payment_id = match &p.kind {
1417                        PaymentKind::Bolt12Offer {
1418                            hash: Some(hash), ..
1419                        } => hash.to_string(),
1420                        _ => {
1421                            tracing::warn!("Bolt12 payment for offer {} missing hash", offer_id);
1422                            return None;
1423                        }
1424                    };
1425
1426                    Some(WaitPaymentResponse {
1427                        payment_identifier: payment_identifier.clone(),
1428                        payment_amount: Amount::new(p.amount_msat?, CurrencyUnit::Msat),
1429                        payment_id,
1430                    })
1431                })
1432                .collect());
1433        }
1434
1435        let payment_id_str = match payment_identifier {
1436            PaymentIdentifier::PaymentHash(hash) => hex::encode(hash),
1437            PaymentIdentifier::CustomId(id) => id.clone(),
1438            _ => return Err(Error::UnsupportedPaymentIdentifierType.into()),
1439        };
1440
1441        let payment_id = PaymentId(
1442            hex::decode(&payment_id_str)?
1443                .try_into()
1444                .map_err(|_| Error::InvalidPaymentIdLength)?,
1445        );
1446
1447        let payment_details = self
1448            .inner
1449            .payment(&payment_id)
1450            .ok_or(Error::PaymentNotFound)?;
1451
1452        if payment_details.direction == PaymentDirection::Outbound {
1453            return Err(Error::InvalidPaymentDirection.into());
1454        }
1455
1456        let amount = if payment_details.status == PaymentStatus::Succeeded {
1457            payment_details
1458                .amount_msat
1459                .ok_or(Error::CouldNotGetPaymentAmount)?
1460        } else {
1461            return Ok(vec![]);
1462        };
1463
1464        let response = WaitPaymentResponse {
1465            payment_identifier: payment_identifier.clone(),
1466            payment_amount: Amount::new(amount, CurrencyUnit::Msat),
1467            payment_id: payment_id_str,
1468        };
1469
1470        Ok(vec![response])
1471    }
1472
1473    /// Check the status of an outgoing payment
1474    async fn check_outgoing_payment(
1475        &self,
1476        request_lookup_id: &PaymentIdentifier,
1477    ) -> Result<MakePaymentResponse, Self::Err> {
1478        let payment_details = match request_lookup_id {
1479            PaymentIdentifier::PaymentHash(id_hash) => {
1480                Self::select_bolt11_payment_details(self.inner.list_payments_with_filter(|p| {
1481                    p.direction == PaymentDirection::Outbound
1482                        && matches!(&p.kind, PaymentKind::Bolt11 { hash, .. } if &hash.0 == id_hash)
1483                }))
1484            }
1485            PaymentIdentifier::PaymentId(id) => self.inner.payment(&PaymentId(*id)),
1486            PaymentIdentifier::QuoteId(quote_id) => {
1487                match read_bolt12_quote_payment_id(&self.kv_store, quote_id)
1488                    .await?
1489                    .resolve()
1490                {
1491                    Bolt12QuotePaymentIdResolution::PaymentId(payment_id) => {
1492                        self.inner.payment(&payment_id)
1493                    }
1494                    Bolt12QuotePaymentIdResolution::Status(status) => {
1495                        return Ok(MakePaymentResponse {
1496                            payment_lookup_id: request_lookup_id.clone(),
1497                            payment_proof: None,
1498                            status,
1499                            total_spent: Amount::new(0, CurrencyUnit::Msat),
1500                        });
1501                    }
1502                }
1503            }
1504            _ => {
1505                return Ok(MakePaymentResponse {
1506                    payment_lookup_id: request_lookup_id.clone(),
1507                    payment_proof: None,
1508                    status: MeltQuoteState::Unknown,
1509                    total_spent: Amount::new(0, CurrencyUnit::Msat),
1510                });
1511            }
1512        }
1513        .ok_or(Error::PaymentNotFound)?;
1514
1515        if payment_details.direction != PaymentDirection::Outbound {
1516            return Err(Error::InvalidPaymentDirection.into());
1517        }
1518
1519        if payment_details.status == PaymentStatus::Failed {
1520            if let PaymentIdentifier::QuoteId(quote_id) = request_lookup_id {
1521                self.cleanup_bolt12_dispatch_binding(quote_id, Some(&payment_details.id))
1522                    .await;
1523            }
1524        }
1525
1526        Self::make_payment_response_from_details(
1527            &CurrencyUnit::Msat,
1528            request_lookup_id.clone(),
1529            &payment_details,
1530        )
1531    }
1532}
1533
1534impl Drop for CdkLdkNode {
1535    fn drop(&mut self) {
1536        tracing::info!("Drop called on CdkLdkNode");
1537        self.wait_invoice_cancel_token.cancel();
1538        tracing::debug!("Cancelled wait_invoice token in drop");
1539    }
1540}
1541
1542/// KV key for the bolt12 melt quote id -> payment id mapping
1543fn bolt12_quote_payment_id_key(quote_id: &QuoteId) -> Result<String, Error> {
1544    match quote_id {
1545        QuoteId::UUID(uuid) => Ok(uuid.to_string()),
1546        QuoteId::BASE64(_) => Err(Error::InvalidQuoteId),
1547    }
1548}
1549
1550/// Records the bolt12 melt quote id -> payment id mapping.
1551///
1552/// `payment_id` of `None` atomically claims an absent quote with the
1553/// pre-dispatch sentinel. `Some` atomically replaces that sentinel with the
1554/// dispatched payment id. Repeating the same payment id is idempotent; every
1555/// other existing binding is rejected.
1556async fn write_bolt12_quote_payment_id(
1557    kv_store: &DynKVStore,
1558    quote_id: &QuoteId,
1559    payment_id: Option<&PaymentId>,
1560) -> Result<(), Error> {
1561    let key = bolt12_quote_payment_id_key(quote_id)?;
1562    let value = payment_id.map(|id| hex::encode(id.0)).unwrap_or_default();
1563    let mut tx = kv_store
1564        .begin_transaction()
1565        .await
1566        .map_err(|e| Error::Database(e.to_string()))?;
1567
1568    let written = match payment_id {
1569        None => {
1570            tx.kv_write_if_absent(
1571                LDK_KV_PRIMARY_NAMESPACE,
1572                LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
1573                &key,
1574                value.as_bytes(),
1575            )
1576            .await
1577        }
1578        Some(_) => {
1579            tx.kv_write_if_equals(
1580                LDK_KV_PRIMARY_NAMESPACE,
1581                LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
1582                &key,
1583                b"",
1584                value.as_bytes(),
1585            )
1586            .await
1587        }
1588    }
1589    .map_err(|e| Error::Database(e.to_string()))?;
1590
1591    if written {
1592        tx.commit()
1593            .await
1594            .map_err(|e| Error::Database(e.to_string()))?;
1595        return Ok(());
1596    }
1597
1598    let existing = tx
1599        .kv_read(
1600            LDK_KV_PRIMARY_NAMESPACE,
1601            LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
1602            &key,
1603        )
1604        .await
1605        .map_err(|e| Error::Database(e.to_string()))?;
1606    tx.rollback()
1607        .await
1608        .map_err(|e| Error::Database(e.to_string()))?;
1609
1610    match existing {
1611        Some(existing) if payment_id.is_some() && existing.as_slice() == value.as_bytes() => Ok(()),
1612        _ => Err(Error::Bolt12QuoteAlreadyClaimed {
1613            quote_id: quote_id.to_string(),
1614        }),
1615    }
1616}
1617
1618/// Reads the bolt12 melt quote id -> payment id mapping
1619async fn read_bolt12_quote_payment_id(
1620    kv_store: &DynKVStore,
1621    quote_id: &QuoteId,
1622) -> Result<Bolt12QuotePaymentIdLookup, Error> {
1623    let key = bolt12_quote_payment_id_key(quote_id)?;
1624    let Some(stored) = kv_store
1625        .kv_read(
1626            LDK_KV_PRIMARY_NAMESPACE,
1627            LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
1628            &key,
1629        )
1630        .await
1631        .map_err(|e| Error::Database(e.to_string()))?
1632    else {
1633        return Ok(Bolt12QuotePaymentIdLookup::Missing);
1634    };
1635
1636    if stored.is_empty() {
1637        return Ok(Bolt12QuotePaymentIdLookup::Dispatching);
1638    }
1639
1640    let payment_id_hex = match String::from_utf8(stored) {
1641        Ok(payment_id_hex) => payment_id_hex,
1642        Err(err) => {
1643            tracing::warn!(
1644                "LDK: invalid UTF-8 in BOLT12 payment id mapping for quote {quote_id}: {err}"
1645            );
1646            return Ok(Bolt12QuotePaymentIdLookup::Malformed);
1647        }
1648    };
1649
1650    let payment_id_bytes = match hex::decode(&payment_id_hex) {
1651        Ok(bytes) => bytes,
1652        Err(err) => {
1653            tracing::warn!(
1654                "LDK: invalid hex in BOLT12 payment id mapping for quote {quote_id}: {err}"
1655            );
1656            return Ok(Bolt12QuotePaymentIdLookup::Malformed);
1657        }
1658    };
1659
1660    let payment_id: [u8; 32] = match payment_id_bytes.try_into() {
1661        Ok(payment_id) => payment_id,
1662        Err(_) => {
1663            tracing::warn!("LDK: invalid payment id length in BOLT12 mapping for quote {quote_id}");
1664            return Ok(Bolt12QuotePaymentIdLookup::Malformed);
1665        }
1666    };
1667
1668    Ok(Bolt12QuotePaymentIdLookup::Found(PaymentId(payment_id)))
1669}
1670
1671/// Atomically removes the bolt12 quote binding only if it still matches the
1672/// expected sentinel (`None`) or payment id (`Some`).
1673async fn delete_bolt12_quote_payment_id_if_equals(
1674    kv_store: &DynKVStore,
1675    quote_id: &QuoteId,
1676    payment_id: Option<&PaymentId>,
1677) -> Result<bool, Error> {
1678    let key = bolt12_quote_payment_id_key(quote_id)?;
1679    let expected = payment_id.map(|id| hex::encode(id.0)).unwrap_or_default();
1680    let mut tx = kv_store
1681        .begin_transaction()
1682        .await
1683        .map_err(|e| Error::Database(e.to_string()))?;
1684
1685    let claimed = tx
1686        .kv_write_if_equals(
1687            LDK_KV_PRIMARY_NAMESPACE,
1688            LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
1689            &key,
1690            expected.as_bytes(),
1691            LDK_KV_BOLT12_CLEANUP_MARKER,
1692        )
1693        .await
1694        .map_err(|e| Error::Database(e.to_string()))?;
1695
1696    if !claimed {
1697        tx.rollback()
1698            .await
1699            .map_err(|e| Error::Database(e.to_string()))?;
1700        return Ok(false);
1701    }
1702
1703    tx.kv_remove(
1704        LDK_KV_PRIMARY_NAMESPACE,
1705        LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
1706        &key,
1707    )
1708    .await
1709    .map_err(|e| Error::Database(e.to_string()))?;
1710    tx.commit()
1711        .await
1712        .map_err(|e| Error::Database(e.to_string()))?;
1713
1714    Ok(true)
1715}
1716
1717#[cfg(test)]
1718mod tests {
1719    use super::*;
1720
1721    #[test]
1722    fn bitcoin_rpc_debug_redacts_password() {
1723        let source = ChainSource::BitcoinRpc(BitcoinRpcConfig {
1724            host: "127.0.0.1".to_string(),
1725            port: 8332,
1726            user: "rpc-user".to_string(),
1727            password: "rpc-password-secret".to_string(),
1728        });
1729
1730        let debug = format!("{source:?}");
1731
1732        assert!(debug.contains("127.0.0.1"));
1733        assert!(debug.contains("rpc-user"));
1734        assert!(debug.contains("[REDACTED]"));
1735        assert!(!debug.contains("rpc-password-secret"));
1736    }
1737
1738    #[test]
1739    fn chain_source_debug_redacts_url_credentials() {
1740        for source in [
1741            ChainSource::Esplora("https://esplora-user:esplora-secret@example.com/api".to_string()),
1742            ChainSource::Electrum(
1743                "ssl://electrum-user:electrum-secret@example.com:50002".to_string(),
1744            ),
1745        ] {
1746            let debug = format!("{source:?}");
1747
1748            assert!(debug.contains("example.com"));
1749            assert!(!debug.contains("-user"));
1750            assert!(!debug.contains("-secret"));
1751        }
1752    }
1753
1754    #[test]
1755    fn gossip_source_debug_redacts_url_credentials() {
1756        let source = GossipSource::RapidGossipSync(
1757            "https://rgs-user:rgs-secret@example.com/snapshot".to_string(),
1758        );
1759
1760        let debug = format!("{source:?}");
1761
1762        assert!(debug.contains("https://example.com/snapshot"));
1763        assert!(!debug.contains("rgs-user"));
1764        assert!(!debug.contains("rgs-secret"));
1765    }
1766
1767    fn test_payment_details(status: PaymentStatus, amount_msat: Option<u64>) -> PaymentDetails {
1768        PaymentDetails {
1769            id: PaymentId([2; 32]),
1770            kind: PaymentKind::Bolt11 {
1771                hash: PaymentHash([1; 32]),
1772                preimage: None,
1773                secret: None,
1774            },
1775            amount_msat,
1776            fee_paid_msat: None,
1777            direction: PaymentDirection::Outbound,
1778            status,
1779            latest_update_timestamp: 0,
1780        }
1781    }
1782
1783    fn test_payment_details_with_id(
1784        id: [u8; 32],
1785        status: PaymentStatus,
1786        latest_update_timestamp: u64,
1787    ) -> PaymentDetails {
1788        PaymentDetails {
1789            id: PaymentId(id),
1790            latest_update_timestamp,
1791            ..test_payment_details(status, None)
1792        }
1793    }
1794
1795    #[test]
1796    fn failed_payment_response_does_not_require_amount() {
1797        let details = test_payment_details(PaymentStatus::Failed, None);
1798
1799        let response = CdkLdkNode::make_payment_response_from_details(
1800            &CurrencyUnit::Msat,
1801            PaymentIdentifier::PaymentId([2; 32]),
1802            &details,
1803        )
1804        .expect("failed payment details should map without amount");
1805
1806        assert_eq!(response.status, MeltQuoteState::Failed);
1807        assert_eq!(response.total_spent, Amount::new(0, CurrencyUnit::Msat));
1808    }
1809
1810    #[test]
1811    fn pending_payment_response_does_not_require_amount() {
1812        let details = test_payment_details(PaymentStatus::Pending, None);
1813
1814        let response = CdkLdkNode::make_payment_response_from_details(
1815            &CurrencyUnit::Msat,
1816            PaymentIdentifier::PaymentId([2; 32]),
1817            &details,
1818        )
1819        .expect("pending payment details should map without amount");
1820
1821        assert_eq!(response.status, MeltQuoteState::Pending);
1822        assert_eq!(response.total_spent, Amount::new(0, CurrencyUnit::Msat));
1823    }
1824
1825    #[tokio::test]
1826    async fn payment_quotes_round_up_sat_principals() {
1827        let storage = std::env::temp_dir().join(format!("cdk-ldk-rounding-{}", QuoteId::new()));
1828        let node = CdkLdkNodeBuilder::new(
1829            Network::Regtest,
1830            ChainSource::Esplora("http://127.0.0.1:1".to_owned()),
1831            GossipSource::P2P,
1832            storage.to_str().unwrap().to_owned(),
1833            FeeReserve {
1834                min_fee_reserve: Amount::ZERO,
1835                percent_fee_reserve: 0.0,
1836            },
1837            vec!["127.0.0.1:0".parse().unwrap()],
1838            test_kv_store().await,
1839        )
1840        .build()
1841        .unwrap();
1842        for (msat, sat) in [(1, 1), (999, 1), (1_000, 1), (1_999, 2), (2_000, 2)] {
1843            let invoice = node
1844                .inner
1845                .bolt11_payment()
1846                .receive(
1847                    msat,
1848                    &Bolt11InvoiceDescription::Direct(
1849                        Description::new("rounding".to_owned()).unwrap(),
1850                    ),
1851                    3600,
1852                )
1853                .unwrap();
1854            let offer = ldk_node::lightning::offers::offer::OfferBuilder::new(node.inner.node_id())
1855                .amount_msats(msat)
1856                .build()
1857                .unwrap();
1858            for (unit, expected) in [(CurrencyUnit::Sat, sat), (CurrencyUnit::Msat, msat)] {
1859                for options in [
1860                    OutgoingPaymentOptions::Bolt11(Box::new(Bolt11OutgoingPaymentOptions {
1861                        bolt11: invoice.clone(),
1862                        max_fee_amount: None,
1863                        timeout_secs: None,
1864                        melt_options: None,
1865                        quote_id: QuoteId::new(),
1866                    })),
1867                    OutgoingPaymentOptions::Bolt12(Box::new(Bolt12OutgoingPaymentOptions {
1868                        offer: offer.clone(),
1869                        max_fee_amount: None,
1870                        timeout_secs: None,
1871                        melt_options: None,
1872                        quote_id: QuoteId::new(),
1873                    })),
1874                ] {
1875                    let quote = node.get_payment_quote(&unit, options).await.unwrap();
1876                    assert_eq!(quote.amount, Amount::new(expected, unit.clone()));
1877                }
1878            }
1879        }
1880        drop(node);
1881        std::fs::remove_dir_all(storage).unwrap();
1882    }
1883
1884    #[test]
1885    fn paid_payment_response_rounds_total_once_for_sat_quotes() {
1886        for (principal, fee, expected_sat) in
1887            [(1_999, 0, 2), (2_000, 1, 3), (1_999, 1, 2), (2_000, 0, 2)]
1888        {
1889            let mut details = test_payment_details(PaymentStatus::Succeeded, Some(principal));
1890            details.fee_paid_msat = Some(fee);
1891            let payment_id = PaymentIdentifier::PaymentId([2; 32]);
1892            let synchronous = CdkLdkNode::make_payment_response_from_details(
1893                &CurrencyUnit::Sat,
1894                payment_id.clone(),
1895                &details,
1896            )
1897            .unwrap();
1898            let recovered = CdkLdkNode::make_payment_response_from_details(
1899                &CurrencyUnit::Msat,
1900                payment_id,
1901                &details,
1902            )
1903            .unwrap();
1904            assert_eq!(
1905                synchronous.total_spent,
1906                Amount::new(expected_sat, CurrencyUnit::Sat)
1907            );
1908            assert_eq!(
1909                recovered.total_spent,
1910                Amount::new(principal + fee, CurrencyUnit::Msat)
1911            );
1912        }
1913    }
1914
1915    #[test]
1916    fn paid_payment_response_requires_amount() {
1917        let details = test_payment_details(PaymentStatus::Succeeded, None);
1918
1919        let err = CdkLdkNode::make_payment_response_from_details(
1920            &CurrencyUnit::Msat,
1921            PaymentIdentifier::PaymentId([2; 32]),
1922            &details,
1923        )
1924        .expect_err("paid payment details without amount should fail");
1925
1926        assert!(matches!(err, payment::Error::Backend(_)));
1927    }
1928
1929    #[test]
1930    fn bolt11_payment_selection_prefers_pending_over_failed() {
1931        let failed = test_payment_details_with_id([1; 32], PaymentStatus::Failed, 2);
1932        let pending = test_payment_details_with_id([2; 32], PaymentStatus::Pending, 1);
1933
1934        let selected = CdkLdkNode::select_bolt11_payment_details([failed, pending])
1935            .expect("payment details should be selected");
1936
1937        assert_eq!(selected.id, PaymentId([2; 32]));
1938        assert_eq!(selected.status, PaymentStatus::Pending);
1939    }
1940
1941    #[test]
1942    fn bolt11_payment_selection_prefers_succeeded_over_pending() {
1943        let pending = test_payment_details_with_id([1; 32], PaymentStatus::Pending, 2);
1944        let succeeded = PaymentDetails {
1945            amount_msat: Some(1000),
1946            ..test_payment_details_with_id([2; 32], PaymentStatus::Succeeded, 1)
1947        };
1948
1949        let selected = CdkLdkNode::select_bolt11_payment_details([pending, succeeded])
1950            .expect("payment details should be selected");
1951
1952        assert_eq!(selected.id, PaymentId([2; 32]));
1953        assert_eq!(selected.status, PaymentStatus::Succeeded);
1954    }
1955
1956    #[test]
1957    fn bolt11_payment_selection_uses_latest_failed_when_all_failed() {
1958        let older_failed = test_payment_details_with_id([1; 32], PaymentStatus::Failed, 1);
1959        let newer_failed = test_payment_details_with_id([2; 32], PaymentStatus::Failed, 2);
1960
1961        let selected = CdkLdkNode::select_bolt11_payment_details([older_failed, newer_failed])
1962            .expect("payment details should be selected");
1963
1964        assert_eq!(selected.id, PaymentId([2; 32]));
1965        assert_eq!(selected.status, PaymentStatus::Failed);
1966    }
1967
1968    #[tokio::test]
1969    async fn terminal_payment_event_wait_ignores_other_payments() {
1970        let (sender, mut receiver) = tokio::sync::broadcast::channel(4);
1971        let payment_id = PaymentId([2; 32]);
1972
1973        // Queue both events before entering the wait to exercise the race where
1974        // LDK completes immediately after dispatch returns.
1975        sender
1976            .send(PaymentId([1; 32]))
1977            .expect("receiver should be subscribed");
1978        sender
1979            .send(payment_id)
1980            .expect("receiver should be subscribed");
1981
1982        CdkLdkNode::wait_for_terminal_payment_event(&mut receiver, payment_id)
1983            .await
1984            .expect("matching terminal event should wake the waiter");
1985    }
1986
1987    #[tokio::test]
1988    async fn terminal_payment_event_wait_recovers_from_lagged_channel() {
1989        let (sender, mut receiver) = tokio::sync::broadcast::channel(2);
1990        let payment_id = PaymentId([3; 32]);
1991
1992        sender
1993            .send(PaymentId([1; 32]))
1994            .expect("receiver should be subscribed");
1995        sender
1996            .send(PaymentId([2; 32]))
1997            .expect("receiver should be subscribed");
1998        sender
1999            .send(PaymentId([4; 32]))
2000            .expect("receiver should be subscribed");
2001        sender
2002            .send(payment_id)
2003            .expect("receiver should be subscribed");
2004
2005        CdkLdkNode::wait_for_terminal_payment_event(&mut receiver, payment_id)
2006            .await
2007            .expect("receiver lag should not prevent a matching event from waking the waiter");
2008    }
2009
2010    #[tokio::test]
2011    async fn terminal_payment_event_wait_reports_closed_channel() {
2012        let (sender, mut receiver) = tokio::sync::broadcast::channel(1);
2013        drop(sender);
2014
2015        let err = CdkLdkNode::wait_for_terminal_payment_event(&mut receiver, PaymentId([2; 32]))
2016            .await
2017            .expect_err("a closed event channel should stop the wait");
2018
2019        assert!(matches!(
2020            err,
2021            tokio::sync::broadcast::error::RecvError::Closed
2022        ));
2023    }
2024
2025    #[test]
2026    fn bolt12_persistence_failure_has_ambiguous_dispatch() {
2027        assert!(bolt12_send_error_has_ambiguous_dispatch(
2028            &ldk_node::NodeError::PersistenceFailed
2029        ));
2030
2031        for not_dispatched in [
2032            ldk_node::NodeError::NotRunning,
2033            ldk_node::NodeError::UnsupportedCurrency,
2034            ldk_node::NodeError::InvalidOffer,
2035            ldk_node::NodeError::InvalidAmount,
2036            ldk_node::NodeError::DuplicatePayment,
2037            ldk_node::NodeError::InvoiceRequestCreationFailed,
2038            ldk_node::NodeError::PaymentSendingFailed,
2039        ] {
2040            assert!(
2041                !bolt12_send_error_has_ambiguous_dispatch(&not_dispatched),
2042                "{not_dispatched} must be treated as not dispatched"
2043            );
2044        }
2045    }
2046
2047    #[test]
2048    fn bolt11_send_errors_only_classify_explicit_rejections_as_terminal() {
2049        for terminal_error in [
2050            ldk_node::NodeError::NotRunning,
2051            ldk_node::NodeError::InvalidAmount,
2052            ldk_node::NodeError::InvalidInvoice,
2053            ldk_node::NodeError::PaymentSendingFailed,
2054        ] {
2055            assert!(
2056                bolt11_send_error_is_explicit_terminal_failure(&terminal_error),
2057                "{terminal_error} must be treated as a definitive failure"
2058            );
2059        }
2060
2061        for ambiguous_error in [
2062            ldk_node::NodeError::PersistenceFailed,
2063            ldk_node::NodeError::DuplicatePayment,
2064        ] {
2065            assert!(
2066                !bolt11_send_error_is_explicit_terminal_failure(&ambiguous_error),
2067                "{ambiguous_error} must not authorize proof release"
2068            );
2069        }
2070    }
2071
2072    #[test]
2073    fn authoritative_outgoing_failure_response_is_terminal_and_spends_nothing() {
2074        let payment_lookup_id = PaymentIdentifier::PaymentHash([42; 32]);
2075        let response =
2076            outgoing_payment_failure_response(&CurrencyUnit::Msat, payment_lookup_id.clone());
2077
2078        assert_eq!(response.payment_lookup_id, payment_lookup_id);
2079        assert_eq!(response.status, MeltQuoteState::Failed);
2080        assert_eq!(response.total_spent, Amount::new(0, CurrencyUnit::Msat));
2081        assert!(response.payment_proof.is_none());
2082    }
2083
2084    #[test]
2085    fn bolt12_quote_payment_id_lookup_resolution_is_safe() {
2086        assert_eq!(
2087            Bolt12QuotePaymentIdLookup::Dispatching.resolve(),
2088            Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Pending),
2089            "an indeterminate dispatch must keep melt proofs reserved"
2090        );
2091        assert_eq!(
2092            Bolt12QuotePaymentIdLookup::Missing.resolve(),
2093            Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Unpaid),
2094            "a missing sentinel means the payment was never dispatched"
2095        );
2096        assert_eq!(
2097            Bolt12QuotePaymentIdLookup::Malformed.resolve(),
2098            Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Unknown),
2099            "corrupt bookkeeping must remain indeterminate"
2100        );
2101    }
2102
2103    async fn test_kv_store() -> DynKVStore {
2104        std::sync::Arc::new(cdk_sqlite::mint::memory::empty().await.unwrap())
2105    }
2106
2107    #[tokio::test]
2108    async fn duplicate_bolt12_claim_resolves_existing_state() {
2109        use ldk_node::lightning::offers::offer::OfferBuilder;
2110
2111        let storage = tempfile::tempdir().unwrap();
2112        let kv_store = test_kv_store().await;
2113        let node = CdkLdkNodeBuilder::new(
2114            Network::Regtest,
2115            ChainSource::Esplora("http://127.0.0.1:1".to_string()),
2116            GossipSource::P2P,
2117            storage.path().to_str().unwrap().to_string(),
2118            FeeReserve {
2119                min_fee_reserve: 0.into(),
2120                percent_fee_reserve: 0.0,
2121            },
2122            vec!["127.0.0.1:0".parse().unwrap()],
2123            kv_store.clone(),
2124        )
2125        .build()
2126        .unwrap();
2127        let offer = OfferBuilder::new(node.inner.node_id())
2128            .amount_msats(1_000)
2129            .build()
2130            .unwrap();
2131
2132        for (stored, expected) in [
2133            (String::new(), Some(MeltQuoteState::Pending)),
2134            ("corrupt".to_string(), Some(MeltQuoteState::Unknown)),
2135            (hex::encode([7; 32]), None),
2136        ] {
2137            let quote_id = QuoteId::new();
2138            let key = bolt12_quote_payment_id_key(&quote_id).unwrap();
2139            let mut tx = kv_store.begin_transaction().await.unwrap();
2140            tx.kv_write(
2141                LDK_KV_PRIMARY_NAMESPACE,
2142                LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
2143                &key,
2144                stored.as_bytes(),
2145            )
2146            .await
2147            .unwrap();
2148            tx.commit().await.unwrap();
2149
2150            let result = node
2151                .make_payment(
2152                    &CurrencyUnit::Sat,
2153                    OutgoingPaymentOptions::Bolt12(Box::new(Bolt12OutgoingPaymentOptions {
2154                        offer: offer.clone(),
2155                        max_fee_amount: None,
2156                        timeout_secs: None,
2157                        melt_options: None,
2158                        quote_id: quote_id.clone(),
2159                    })),
2160                )
2161                .await;
2162
2163            match expected {
2164                Some(status) => {
2165                    let response = result.unwrap();
2166                    assert_eq!(response.status, status);
2167                    assert_eq!(response.total_spent, Amount::new(0, CurrencyUnit::Sat));
2168                    assert_eq!(
2169                        response.payment_lookup_id,
2170                        PaymentIdentifier::QuoteId(quote_id)
2171                    );
2172                }
2173                None => assert!(result.is_err(), "missing LDK payment must be indeterminate"),
2174            }
2175            assert_eq!(
2176                kv_store
2177                    .kv_read(
2178                        LDK_KV_PRIMARY_NAMESPACE,
2179                        LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
2180                        &key,
2181                    )
2182                    .await
2183                    .unwrap(),
2184                Some(stored.into_bytes()),
2185                "a duplicate must preserve the existing dispatch binding"
2186            );
2187        }
2188    }
2189
2190    #[tokio::test]
2191    async fn duplicate_paid_bolt12_claim_rounds_total_once() {
2192        use ldk_node::bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
2193        use ldk_node::io::sqlite_store::{SqliteStore, KV_TABLE_NAME, SQLITE_DB_FILE_NAME};
2194        use ldk_node::lightning::offers::offer::OfferBuilder;
2195        use ldk_node::lightning::util::persist::KVStoreSync;
2196        use ldk_node::lightning::util::ser::Writeable;
2197        use ldk_node::lightning_types::payment::PaymentPreimage;
2198
2199        let storage = tempfile::tempdir().unwrap();
2200        let kv_store = test_kv_store().await;
2201        let signer = PublicKey::from_secret_key(
2202            &Secp256k1::new(),
2203            &SecretKey::from_slice(&[1; 32]).unwrap(),
2204        );
2205        let offer = OfferBuilder::new(signer)
2206            .amount_msats(1_000)
2207            .build()
2208            .unwrap();
2209        let preimage = PaymentPreimage([42; 32]);
2210        let mut cases = Vec::new();
2211        {
2212            let store = SqliteStore::new(
2213                storage.path().to_path_buf(),
2214                Some(SQLITE_DB_FILE_NAME.to_owned()),
2215                Some(KV_TABLE_NAME.to_owned()),
2216            )
2217            .unwrap();
2218            for (index, (principal, fee, expected_sat)) in
2219                [(10_000, 1, 11), (1_999, 1, 2), (1_999, 0, 2), (2_000, 0, 2)]
2220                    .into_iter()
2221                    .enumerate()
2222            {
2223                let payment_id = PaymentId([index as u8 + 1; 32]);
2224                let details = PaymentDetails {
2225                    id: payment_id,
2226                    kind: PaymentKind::Bolt12Offer {
2227                        hash: Some(PaymentHash(
2228                            ldk_node::bitcoin::hashes::sha256::Hash::hash(&preimage.0)
2229                                .to_byte_array(),
2230                        )),
2231                        preimage: Some(preimage),
2232                        secret: None,
2233                        offer_id: offer.id(),
2234                        payer_note: None,
2235                        quantity: None,
2236                    },
2237                    fee_paid_msat: Some(fee),
2238                    ..test_payment_details(PaymentStatus::Succeeded, Some(principal))
2239                };
2240                // Seed LDK's persisted payment history before the node loads it.
2241                store
2242                    .write("payments", "", &hex::encode(payment_id.0), details.encode())
2243                    .unwrap();
2244                let quote_id = QuoteId::new();
2245                write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
2246                    .await
2247                    .unwrap();
2248                write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&payment_id))
2249                    .await
2250                    .unwrap();
2251                cases.push((quote_id, payment_id, principal + fee, expected_sat));
2252            }
2253        }
2254        let node = CdkLdkNodeBuilder::new(
2255            Network::Regtest,
2256            ChainSource::Esplora("http://127.0.0.1:1".to_owned()),
2257            GossipSource::P2P,
2258            storage.path().to_str().unwrap().to_owned(),
2259            FeeReserve {
2260                min_fee_reserve: 0.into(),
2261                percent_fee_reserve: 0.0,
2262            },
2263            vec!["127.0.0.1:0".parse().unwrap()],
2264            kv_store.clone(),
2265        )
2266        .build()
2267        .unwrap();
2268
2269        for (quote_id, payment_id, total_msat, expected_sat) in cases {
2270            for (unit, expected) in [
2271                (CurrencyUnit::Sat, expected_sat),
2272                (CurrencyUnit::Msat, total_msat),
2273            ] {
2274                let response = node
2275                    .make_payment(
2276                        &unit,
2277                        OutgoingPaymentOptions::Bolt12(Box::new(Bolt12OutgoingPaymentOptions {
2278                            offer: offer.clone(),
2279                            max_fee_amount: None,
2280                            timeout_secs: None,
2281                            melt_options: None,
2282                            quote_id: quote_id.clone(),
2283                        })),
2284                    )
2285                    .await
2286                    .unwrap();
2287                assert_eq!(response.status, MeltQuoteState::Paid);
2288                assert_eq!(response.total_spent, Amount::new(expected, unit));
2289                assert_eq!(response.payment_proof, Some(preimage.to_string()));
2290                assert_eq!(
2291                    response.payment_lookup_id,
2292                    PaymentIdentifier::QuoteId(quote_id.clone())
2293                );
2294                assert_eq!(
2295                    read_bolt12_quote_payment_id(&kv_store, &quote_id)
2296                        .await
2297                        .unwrap(),
2298                    Bolt12QuotePaymentIdLookup::Found(payment_id),
2299                );
2300            }
2301        }
2302        assert_eq!(node.inner.list_payments().len(), 4);
2303    }
2304
2305    /// The mapping must resolve Missing before any dispatch, Dispatching
2306    /// (indeterminate) while only the pre-dispatch sentinel exists, and Found
2307    /// after the payment id is recorded.
2308    #[tokio::test]
2309    async fn bolt12_quote_payment_id_mapping_lifecycle() {
2310        let kv_store = test_kv_store().await;
2311        let quote_id = QuoteId::new();
2312
2313        assert_eq!(
2314            read_bolt12_quote_payment_id(&kv_store, &quote_id)
2315                .await
2316                .unwrap(),
2317            Bolt12QuotePaymentIdLookup::Missing,
2318            "no record must resolve as never dispatched"
2319        );
2320
2321        // Pre-dispatch sentinel
2322        write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
2323            .await
2324            .unwrap();
2325        assert_eq!(
2326            read_bolt12_quote_payment_id(&kv_store, &quote_id)
2327                .await
2328                .unwrap(),
2329            Bolt12QuotePaymentIdLookup::Dispatching,
2330            "sentinel must resolve as indeterminate, never terminal"
2331        );
2332
2333        assert!(
2334            delete_bolt12_quote_payment_id_if_equals(&kv_store, &quote_id, None)
2335                .await
2336                .unwrap(),
2337            "an unambiguous pre-dispatch failure should release its sentinel"
2338        );
2339        write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
2340            .await
2341            .expect("a retry should reclaim the quote after sentinel cleanup");
2342
2343        // Record the payment id
2344        let payment_id = PaymentId([7; 32]);
2345        write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&payment_id))
2346            .await
2347            .unwrap();
2348        assert_eq!(
2349            read_bolt12_quote_payment_id(&kv_store, &quote_id)
2350                .await
2351                .unwrap(),
2352            Bolt12QuotePaymentIdLookup::Found(payment_id)
2353        );
2354
2355        // Removal returns to Missing (failed dispatch cleanup)
2356        assert!(
2357            delete_bolt12_quote_payment_id_if_equals(&kv_store, &quote_id, Some(&payment_id))
2358                .await
2359                .unwrap()
2360        );
2361        assert_eq!(
2362            read_bolt12_quote_payment_id(&kv_store, &quote_id)
2363                .await
2364                .unwrap(),
2365            Bolt12QuotePaymentIdLookup::Missing
2366        );
2367    }
2368
2369    #[tokio::test]
2370    async fn bolt12_quote_payment_id_binding_is_write_once() {
2371        let kv_store = test_kv_store().await;
2372        let quote_id = QuoteId::new();
2373        let payment_id = PaymentId([7; 32]);
2374        let conflicting_payment_id = PaymentId([9; 32]);
2375
2376        write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
2377            .await
2378            .expect("first dispatch should claim the quote");
2379        write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&payment_id))
2380            .await
2381            .expect("the claim owner should record its payment id");
2382        write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&payment_id))
2383            .await
2384            .expect("repeating the same payment id must be idempotent");
2385
2386        let duplicate_dispatch = write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
2387            .await
2388            .expect_err("a dispatched quote must not be claimed again");
2389        assert!(
2390            matches!(duplicate_dispatch, Error::Bolt12QuoteAlreadyClaimed { .. }),
2391            "unexpected error: {duplicate_dispatch}"
2392        );
2393
2394        let conflicting_binding =
2395            write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&conflicting_payment_id))
2396                .await
2397                .expect_err("a conflicting payment id must be rejected");
2398        assert!(
2399            matches!(conflicting_binding, Error::Bolt12QuoteAlreadyClaimed { .. }),
2400            "unexpected error: {conflicting_binding}"
2401        );
2402
2403        assert_eq!(
2404            read_bolt12_quote_payment_id(&kv_store, &quote_id)
2405                .await
2406                .expect("payment id should remain readable"),
2407            Bolt12QuotePaymentIdLookup::Found(payment_id),
2408            "a duplicate dispatch must not redirect recovery"
2409        );
2410    }
2411
2412    #[tokio::test]
2413    async fn failed_bolt12_binding_can_be_released_without_removing_a_retry() {
2414        let kv_store = test_kv_store().await;
2415        let quote_id = QuoteId::new();
2416        let failed_payment_id = PaymentId([7; 32]);
2417        let retry_payment_id = PaymentId([9; 32]);
2418
2419        write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
2420            .await
2421            .expect("failed dispatch should claim the quote");
2422        write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&failed_payment_id))
2423            .await
2424            .expect("failed payment id should be recorded");
2425        assert!(delete_bolt12_quote_payment_id_if_equals(
2426            &kv_store,
2427            &quote_id,
2428            Some(&failed_payment_id),
2429        )
2430        .await
2431        .expect("failed binding should be released"));
2432
2433        write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
2434            .await
2435            .expect("retry should claim the released quote");
2436        write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&retry_payment_id))
2437            .await
2438            .expect("retry payment id should be recorded");
2439        assert!(!delete_bolt12_quote_payment_id_if_equals(
2440            &kv_store,
2441            &quote_id,
2442            Some(&failed_payment_id),
2443        )
2444        .await
2445        .expect("stale cleanup should be checked atomically"));
2446
2447        assert_eq!(
2448            read_bolt12_quote_payment_id(&kv_store, &quote_id)
2449                .await
2450                .expect("retry binding should remain readable"),
2451            Bolt12QuotePaymentIdLookup::Found(retry_payment_id),
2452            "stale failed-payment cleanup must not remove a newer retry"
2453        );
2454    }
2455
2456    #[tokio::test]
2457    async fn bolt12_quote_dispatch_concurrent_claims_have_single_winner() {
2458        let kv_store = test_kv_store().await;
2459        let quote_id = QuoteId::new();
2460
2461        let (first_result, second_result) = tokio::join!(
2462            write_bolt12_quote_payment_id(&kv_store, &quote_id, None),
2463            write_bolt12_quote_payment_id(&kv_store, &quote_id, None),
2464        );
2465
2466        let outcomes = [first_result, second_result];
2467        let winners = outcomes.iter().filter(|result| result.is_ok()).count();
2468        let conflicts = outcomes
2469            .iter()
2470            .filter(|result| matches!(result, Err(Error::Bolt12QuoteAlreadyClaimed { .. })))
2471            .count();
2472
2473        assert_eq!(winners, 1, "exactly one dispatch may claim the quote");
2474        assert_eq!(conflicts, 1, "the losing dispatch must be rejected");
2475    }
2476
2477    #[tokio::test]
2478    async fn bolt12_quote_payment_id_concurrent_resolution_has_single_winner() {
2479        let kv_store = test_kv_store().await;
2480        let quote_id = QuoteId::new();
2481        let first_payment_id = PaymentId([7; 32]);
2482        let second_payment_id = PaymentId([9; 32]);
2483
2484        write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
2485            .await
2486            .expect("dispatch should claim the quote");
2487
2488        let (first_result, second_result) = tokio::join!(
2489            write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&first_payment_id)),
2490            write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&second_payment_id)),
2491        );
2492
2493        let outcomes = [&first_result, &second_result];
2494        let winners = outcomes.iter().filter(|result| result.is_ok()).count();
2495        let conflicts = outcomes
2496            .iter()
2497            .filter(|result| matches!(result, Err(Error::Bolt12QuoteAlreadyClaimed { .. })))
2498            .count();
2499
2500        assert_eq!(winners, 1, "exactly one payment id may resolve the claim");
2501        assert_eq!(conflicts, 1, "the losing resolution must be rejected");
2502
2503        let winner = if first_result.is_ok() {
2504            first_payment_id
2505        } else {
2506            second_payment_id
2507        };
2508        assert_eq!(
2509            read_bolt12_quote_payment_id(&kv_store, &quote_id)
2510                .await
2511                .expect("payment id should remain readable"),
2512            Bolt12QuotePaymentIdLookup::Found(winner)
2513        );
2514    }
2515
2516    /// A corrupted mapping must resolve as indeterminate (`Malformed`), never
2517    /// as a terminal state that could trigger compensation.
2518    #[tokio::test]
2519    async fn bolt12_quote_payment_id_mapping_malformed_is_indeterminate() {
2520        let kv_store = test_kv_store().await;
2521        let quote_id = QuoteId::new();
2522        let key = bolt12_quote_payment_id_key(&quote_id).unwrap();
2523
2524        for corrupt in ["not-hex", "0102", "zz"] {
2525            let mut tx = kv_store.begin_transaction().await.unwrap();
2526            tx.kv_write(
2527                LDK_KV_PRIMARY_NAMESPACE,
2528                LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
2529                &key,
2530                corrupt.as_bytes(),
2531            )
2532            .await
2533            .unwrap();
2534            tx.commit().await.unwrap();
2535
2536            assert_eq!(
2537                read_bolt12_quote_payment_id(&kv_store, &quote_id)
2538                    .await
2539                    .unwrap(),
2540                Bolt12QuotePaymentIdLookup::Malformed,
2541                "corrupt value {corrupt} must be indeterminate"
2542            );
2543        }
2544    }
2545}