cdk_ldk_node/

lib.rs

//! CDK lightning backend for ldk-node

#![doc = include_str!("../README.md")]

use std::net::SocketAddr;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use bip39::Mnemonic;
use cdk_common::common::FeeReserve;
use cdk_common::payment::{self, *};
use cdk_common::util::{hex, unix_time};
use cdk_common::{Amount, CurrencyUnit, MeltOptions, MeltQuoteState};
use futures::{Stream, StreamExt};
use ldk_node::bitcoin::hashes::Hash;
use ldk_node::bitcoin::Network;
use ldk_node::lightning::ln::channelmanager::PaymentId;
use ldk_node::lightning::ln::msgs::SocketAddress;
use ldk_node::lightning::routing::router::RouteParametersConfig;
use ldk_node::lightning_invoice::{Bolt11InvoiceDescription, Description};
use ldk_node::lightning_types::payment::PaymentHash;
use ldk_node::logger::{LogLevel, LogWriter};
use ldk_node::payment::{PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus};
use ldk_node::{Builder, Event, Node};
use tokio_stream::wrappers::BroadcastStream;
use tokio_util::sync::CancellationToken;
use tracing::instrument;

use crate::error::Error;
use crate::log::StdoutLogWriter;

mod error;
mod log;
mod web;

/// Maximum time a synchronous payment request waits for an LDK terminal event
const PAYMENT_WAIT_TIMEOUT: Duration = Duration::from_secs(10);
/// Capacity for terminal outgoing payment notifications
const PAYMENT_EVENT_CHANNEL_CAPACITY: usize = 64;

/// CDK Lightning backend using LDK Node
///
/// Provides Lightning Network functionality for CDK with support for Cashu operations.
/// Handles payment creation, processing, and event management using the Lightning Development Kit.
#[derive(Clone)]
pub struct CdkLdkNode {
    inner: Arc<Node>,
    fee_reserve: FeeReserve,
    wait_invoice_cancel_token: CancellationToken,
    wait_invoice_is_active: Arc<AtomicBool>,
    sender: tokio::sync::broadcast::Sender<WaitPaymentResponse>,
    receiver: Arc<tokio::sync::broadcast::Receiver<WaitPaymentResponse>>,
    outgoing_payment_sender: tokio::sync::broadcast::Sender<PaymentId>,
    events_cancel_token: CancellationToken,
    web_addr: Option<SocketAddr>,
}

impl std::fmt::Debug for CdkLdkNode {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CdkLdkNode")
            .field("fee_reserve", &self.fee_reserve)
            .field("web_addr", &self.web_addr)
            .finish_non_exhaustive()
    }
}

/// Configuration for connecting to Bitcoin RPC
///
/// Contains the necessary connection parameters for Bitcoin Core RPC interface.
#[derive(Debug, Clone)]
pub struct BitcoinRpcConfig {
    /// Bitcoin RPC server hostname or IP address
    pub host: String,
    /// Bitcoin RPC server port number
    pub port: u16,
    /// Username for Bitcoin RPC authentication
    pub user: String,
    /// Password for Bitcoin RPC authentication
    pub password: String,
}

/// Source of blockchain data for the Lightning node
///
/// Specifies how the node should connect to the Bitcoin network to retrieve
/// blockchain information and broadcast transactions.
#[derive(Debug, Clone)]
pub enum ChainSource {
    /// Use an Esplora server for blockchain data
    ///
    /// Contains the URL of the Esplora server endpoint
    Esplora(String),
    /// Use Bitcoin Core RPC for blockchain data
    ///
    /// Contains the configuration for connecting to Bitcoin Core
    BitcoinRpc(BitcoinRpcConfig),
}

/// Source of Lightning network gossip data
///
/// Specifies how the node should learn about the Lightning Network topology
/// and routing information.
#[derive(Debug, Clone)]
pub enum GossipSource {
    /// Learn gossip through peer-to-peer connections
    ///
    /// The node will connect to other Lightning nodes and exchange gossip data directly
    P2P,
    /// Use Rapid Gossip Sync for efficient gossip updates
    ///
    /// Contains the URL of the RGS server for compressed gossip data
    RapidGossipSync(String),
}
/// A builder for an [`CdkLdkNode`] instance.
#[derive(Debug)]
pub struct CdkLdkNodeBuilder {
    network: Network,
    chain_source: ChainSource,
    gossip_source: GossipSource,
    log_dir_path: Option<String>,
    storage_dir_path: String,
    fee_reserve: FeeReserve,
    listening_addresses: Vec<SocketAddress>,
    seed: Option<Mnemonic>,
    announcement_addresses: Option<Vec<SocketAddress>>,
}

impl CdkLdkNodeBuilder {
    /// Creates a new builder instance.
    pub fn new(
        network: Network,
        chain_source: ChainSource,
        gossip_source: GossipSource,
        storage_dir_path: String,
        fee_reserve: FeeReserve,
        listening_addresses: Vec<SocketAddress>,
    ) -> Self {
        Self {
            network,
            chain_source,
            gossip_source,
            storage_dir_path,
            fee_reserve,
            listening_addresses,
            seed: None,
            announcement_addresses: None,
            log_dir_path: None,
        }
    }

    /// Configures the [`CdkLdkNode`] to use the Mnemonic for entropy source configuration
    pub fn with_seed(mut self, seed: Mnemonic) -> Self {
        self.seed = Some(seed);
        self
    }
    /// Configures the [`CdkLdkNode`] to use announce this address to the lightning network
    pub fn with_announcement_address(mut self, announcement_addresses: Vec<SocketAddress>) -> Self {
        self.announcement_addresses = Some(announcement_addresses);
        self
    }
    /// Configures the [`CdkLdkNode`] to use announce this address to the lightning network
    pub fn with_log_dir_path(mut self, log_dir_path: String) -> Self {
        self.log_dir_path = Some(log_dir_path);
        self
    }

    /// Builds the [`CdkLdkNode`] instance
    ///
    /// # Errors
    /// Returns an error if the LDK node builder fails to create the node
    pub fn build(self) -> Result<CdkLdkNode, Error> {
        let mut ldk = Builder::new();
        ldk.set_network(self.network);
        tracing::info!("Storage dir of node is {}", self.storage_dir_path);
        ldk.set_storage_dir_path(self.storage_dir_path);

        match self.chain_source {
            ChainSource::Esplora(esplora_url) => {
                ldk.set_chain_source_esplora(esplora_url, None);
            }
            ChainSource::BitcoinRpc(BitcoinRpcConfig {
                host,
                port,
                user,
                password,
            }) => {
                ldk.set_chain_source_bitcoind_rpc(host, port, user, password);
            }
        }

        match self.gossip_source {
            GossipSource::P2P => {
                ldk.set_gossip_source_p2p();
            }
            GossipSource::RapidGossipSync(rgs_url) => {
                ldk.set_gossip_source_rgs(rgs_url);
            }
        }

        ldk.set_listening_addresses(self.listening_addresses)?;
        if self.log_dir_path.is_some() {
            ldk.set_filesystem_logger(self.log_dir_path, Some(LogLevel::Info));
        } else {
            ldk.set_custom_logger(Arc::new(StdoutLogWriter));
        }

        ldk.set_node_alias("cdk-ldk-node".to_string())?;
        // set the seed as bip39 entropy mnemonic
        if let Some(seed) = self.seed {
            ldk.set_entropy_bip39_mnemonic(seed, None);
        }
        // set the announcement addresses
        if let Some(announcement_addresses) = self.announcement_addresses {
            ldk.set_announcement_addresses(announcement_addresses)?;
        }

        let node = ldk.build()?;

        tracing::info!("Creating tokio channel for payment notifications");
        let (sender, receiver) = tokio::sync::broadcast::channel(8);
        let (outgoing_payment_sender, _) =
            tokio::sync::broadcast::channel(PAYMENT_EVENT_CHANNEL_CAPACITY);

        let id = node.node_id();

        let adr = node.announcement_addresses();

        tracing::info!(
            "Created node {} with address {:?} on network {}",
            id,
            adr,
            self.network
        );

        Ok(CdkLdkNode {
            inner: node.into(),
            fee_reserve: self.fee_reserve,
            wait_invoice_cancel_token: CancellationToken::new(),
            wait_invoice_is_active: Arc::new(AtomicBool::new(false)),
            sender,
            receiver: Arc::new(receiver),
            outgoing_payment_sender,
            events_cancel_token: CancellationToken::new(),
            web_addr: None,
        })
    }
}

impl CdkLdkNode {
    /// Set the web server address for the LDK node management interface
    ///
    /// # Arguments
    /// * `addr` - Socket address for the web server. If None, no web server will be started.
    pub fn set_web_addr(&mut self, addr: Option<SocketAddr>) {
        self.web_addr = addr;
    }

    /// Get a default web server address using an unused port
    ///
    /// Returns a SocketAddr with localhost and port 0, which will cause
    /// the system to automatically assign an available port
    pub fn default_web_addr() -> SocketAddr {
        SocketAddr::from(([127, 0, 0, 1], 8091))
    }

    fn make_payment_response_from_details(
        unit: &CurrencyUnit,
        payment_lookup_id: PaymentIdentifier,
        payment_details: &PaymentDetails,
    ) -> Result<MakePaymentResponse, payment::Error> {
        let status = match payment_details.status {
            PaymentStatus::Pending => MeltQuoteState::Pending,
            PaymentStatus::Succeeded => MeltQuoteState::Paid,
            PaymentStatus::Failed => MeltQuoteState::Failed,
        };

        let payment_proof = match &payment_details.kind {
            PaymentKind::Bolt11 { preimage, .. } => preimage.map(|p| p.to_string()),
            PaymentKind::Bolt12Offer { preimage, .. } => preimage.map(|p| p.to_string()),
            _ => return Err(Error::UnexpectedPaymentKind.into()),
        };

        let total_spent = if status == MeltQuoteState::Paid {
            let total_spent = payment_details
                .amount_msat
                .ok_or(Error::CouldNotGetAmountSpent)?
                + payment_details.fee_paid_msat.unwrap_or_default();
            Amount::new(total_spent, CurrencyUnit::Msat).convert_to(unit)?
        } else {
            Amount::new(0, unit.clone())
        };

        Ok(MakePaymentResponse {
            payment_lookup_id,
            payment_proof,
            status,
            total_spent,
        })
    }

    fn select_bolt11_payment_details(
        payment_details: impl IntoIterator<Item = PaymentDetails>,
    ) -> Option<PaymentDetails> {
        payment_details.into_iter().min_by_key(|details| {
            let status_order = match details.status {
                PaymentStatus::Succeeded => 0_u8,
                PaymentStatus::Pending => 1,
                PaymentStatus::Failed => 2,
            };

            (
                status_order,
                std::cmp::Reverse(details.latest_update_timestamp),
            )
        })
    }

    async fn wait_for_terminal_payment_event(
        receiver: &mut tokio::sync::broadcast::Receiver<PaymentId>,
        payment_id: PaymentId,
    ) -> Result<(), tokio::sync::broadcast::error::RecvError> {
        loop {
            match receiver.recv().await {
                Ok(completed_payment_id) if completed_payment_id == payment_id => return Ok(()),
                Ok(_) => continue,
                Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
                    tracing::warn!(
                        payment_id = %payment_id,
                        skipped,
                        "Terminal payment event receiver lagged; continuing to wait"
                    );
                }
                Err(err) => return Err(err),
            }
        }
    }

    async fn wait_for_payment_terminal_status(
        &self,
        payment_id: PaymentId,
        mut receiver: tokio::sync::broadcast::Receiver<PaymentId>,
    ) -> Result<PaymentDetails, payment::Error> {
        let payment_details = self
            .inner
            .payment(&payment_id)
            .ok_or(Error::PaymentNotFound)?;

        if payment_details.status != PaymentStatus::Pending {
            return Ok(payment_details);
        }

        match tokio::time::timeout(
            PAYMENT_WAIT_TIMEOUT,
            Self::wait_for_terminal_payment_event(&mut receiver, payment_id),
        )
        .await
        {
            Ok(Ok(())) => {}
            Ok(Err(err)) => {
                tracing::warn!(
                    payment_id = %payment_id,
                    "Could not wait for terminal LDK payment event: {err}"
                );
            }
            Err(_) => {
                tracing::warn!(
                    payment_id = %payment_id,
                    "Payment did not reach a terminal state within {} seconds",
                    PAYMENT_WAIT_TIMEOUT.as_secs()
                );
            }
        }

        let payment_details = self
            .inner
            .payment(&payment_id)
            .ok_or(Error::PaymentNotFound)?;

        if payment_details.status == PaymentStatus::Pending {
            tracing::debug!(
                payment_id = %payment_id,
                "Payment remains pending after waiting for a terminal event"
            );
        }

        Ok(payment_details)
    }

    /// Start the CDK LDK Node
    ///
    /// Starts the underlying LDK node and begins event processing.
    /// Sets up event handlers to listen for Lightning events like payment received.
    ///
    /// # Returns
    /// Returns `Ok(())` on successful start, error otherwise
    ///
    /// # Errors
    /// Returns an error if the LDK node fails to start or event handling setup fails
    pub fn start_ldk_node(&self) -> Result<(), Error> {
        tracing::info!("Starting cdk-ldk node");
        self.inner.start()?;
        let node_config = self.inner.config();

        tracing::info!("Starting node with network {}", node_config.network);

        tracing::info!("Node status: {:?}", self.inner.status());

        self.handle_events()?;

        Ok(())
    }

    /// Start the web server for the LDK node management interface
    ///
    /// Starts a web server that provides a user interface for managing the LDK node.
    /// The web interface allows users to view balances, manage channels, create invoices,
    /// and send payments.
    ///
    /// # Arguments
    /// * `web_addr` - The socket address to bind the web server to
    ///
    /// # Returns
    /// Returns `Ok(())` on successful start, error otherwise
    ///
    /// # Errors
    /// Returns an error if the web server fails to start
    pub fn start_web_server(&self, web_addr: SocketAddr) -> Result<(), Error> {
        let web_server = crate::web::WebServer::new(Arc::new(self.clone()));

        tokio::spawn(async move {
            if let Err(e) = web_server.serve(web_addr).await {
                tracing::error!("Web server error: {}", e);
            }
        });

        Ok(())
    }

    /// Stop the CDK LDK Node
    ///
    /// Gracefully stops the node by cancelling all active tasks and event handlers.
    /// This includes:
    /// - Cancelling the event handler task
    /// - Cancelling any active wait_invoice streams
    /// - Stopping the underlying LDK node
    ///
    /// # Returns
    /// Returns `Ok(())` on successful shutdown, error otherwise
    ///
    /// # Errors
    /// Returns an error if the underlying LDK node fails to stop
    pub fn stop_ldk_node(&self) -> Result<(), Error> {
        tracing::info!("Stopping CdkLdkNode");
        // Cancel all tokio tasks
        tracing::info!("Cancelling event handler");
        self.events_cancel_token.cancel();

        // Cancel any payment event streams
        if self.is_payment_event_stream_active() {
            tracing::info!("Cancelling payment event stream");
            self.wait_invoice_cancel_token.cancel();
        }

        // Stop the LDK node
        tracing::info!("Stopping LDK node");
        self.inner.stop()?;
        tracing::info!("CdkLdkNode stopped successfully");
        Ok(())
    }

    /// Handle payment received event
    async fn handle_payment_received(
        node: &Arc<Node>,
        sender: &tokio::sync::broadcast::Sender<WaitPaymentResponse>,
        payment_id: Option<PaymentId>,
        payment_hash: PaymentHash,
        amount_msat: u64,
    ) {
        tracing::info!(
            "Received payment for hash={} of amount={} msat",
            payment_hash,
            amount_msat
        );

        let payment_id = match payment_id {
            Some(id) => id,
            None => {
                tracing::warn!("Received payment without payment_id");
                return;
            }
        };

        let payment_id_hex = hex::encode(payment_id.0);

        if amount_msat == 0 {
            tracing::warn!("Payment of no amount");
            return;
        }

        tracing::info!(
            "Processing payment notification: id={}, amount={} msats",
            payment_id_hex,
            amount_msat
        );

        let payment_details = match node.payment(&payment_id) {
            Some(details) => details,
            None => {
                tracing::error!("Could not find payment details for id={}", payment_id_hex);
                return;
            }
        };

        let (payment_identifier, payment_id) = match payment_details.kind {
            PaymentKind::Bolt11 { hash, .. } => {
                (PaymentIdentifier::PaymentHash(hash.0), hash.to_string())
            }
            PaymentKind::Bolt12Offer { hash, offer_id, .. } => match hash {
                Some(h) => (
                    PaymentIdentifier::OfferId(offer_id.to_string()),
                    h.to_string(),
                ),
                None => {
                    tracing::error!("Bolt12 payment missing hash");
                    return;
                }
            },
            k => {
                tracing::warn!("Received payment of kind {:?} which is not supported", k);
                return;
            }
        };

        let wait_payment_response = WaitPaymentResponse {
            payment_identifier,
            payment_amount: Amount::new(amount_msat, CurrencyUnit::Msat),
            payment_id,
        };

        match sender.send(wait_payment_response) {
            Ok(_) => tracing::info!("Successfully sent payment notification to stream"),
            Err(err) => tracing::error!(
                "Could not send payment received notification on channel: {}",
                err
            ),
        }
    }

    /// Set up event handling for the node
    pub fn handle_events(&self) -> Result<(), Error> {
        let node = self.inner.clone();
        let sender = self.sender.clone();
        let outgoing_payment_sender = self.outgoing_payment_sender.clone();
        let cancel_token = self.events_cancel_token.clone();

        tracing::info!("Starting event handler task");

        tokio::spawn(async move {
            tracing::info!("Event handler loop started");
            loop {
                tokio::select! {
                    _ = cancel_token.cancelled() => {
                        tracing::info!("Event handler cancelled");
                        break;
                    }
                    event = node.next_event_async() => {
                        match event {
                            Event::PaymentReceived {
                                payment_id,
                                payment_hash,
                                amount_msat,
                                custom_records: _
                            } => {
                                Self::handle_payment_received(
                                    &node,
                                    &sender,
                                    payment_id,
                                    payment_hash,
                                    amount_msat
                                ).await;
                            }
                            Event::PaymentSuccessful {
                                payment_id,
                                payment_hash,
                                payment_preimage: _,
                                fee_paid_msat: _,
                            } => {
                                tracing::info!(
                                    payment_id = ?payment_id,
                                    payment_hash = %payment_hash,
                                    "LDK node payment succeeded"
                                );
                                if let Some(payment_id) = payment_id {
                                    let _ = outgoing_payment_sender.send(payment_id);
                                }
                            }
                            Event::PaymentFailed {
                                payment_id,
                                payment_hash,
                                reason,
                            } => {
                                tracing::error!(
                                    payment_id = ?payment_id,
                                    payment_hash = ?payment_hash,
                                    reason = ?reason,
                                    "LDK node payment failed"
                                );
                                if let Some(payment_id) = payment_id {
                                    let _ = outgoing_payment_sender.send(payment_id);
                                }
                            }
                            event => {
                                tracing::debug!("Received other ldk node event: {:?}", event);
                            }
                        }

                        if let Err(err) = node.event_handled() {
                            tracing::error!("Error handling node event: {}", err);
                        } else {
                            tracing::debug!("Successfully handled node event");
                        }
                    }
                }
            }
            tracing::info!("Event handler loop terminated");
        });

        tracing::info!("Event handler task spawned");
        Ok(())
    }

    /// Get Node used
    pub fn node(&self) -> Arc<Node> {
        Arc::clone(&self.inner)
    }
}

/// Mint payment trait
#[async_trait]
impl MintPayment for CdkLdkNode {
    type Err = payment::Error;

    /// Start the payment processor
    /// Starts the LDK node and begins event processing
    async fn start(&self) -> Result<(), Self::Err> {
        self.start_ldk_node().map_err(|e| {
            tracing::error!("Failed to start CdkLdkNode: {}", e);
            e
        })?;

        tracing::info!("CdkLdkNode payment processor started successfully");

        // Start web server if configured
        if let Some(web_addr) = self.web_addr {
            tracing::info!("Starting LDK Node web interface on {}", web_addr);
            self.start_web_server(web_addr).map_err(|e| {
                tracing::error!("Failed to start web server: {}", e);
                e
            })?;
        } else {
            tracing::info!("No web server address configured, skipping web interface");
        }

        Ok(())
    }

    /// Stop the payment processor
    /// Gracefully stops the LDK node and cancels all background tasks
    async fn stop(&self) -> Result<(), Self::Err> {
        self.stop_ldk_node().map_err(|e| {
            tracing::error!("Failed to stop CdkLdkNode: {}", e);
            e.into()
        })
    }

    /// Base Settings
    async fn get_settings(&self) -> Result<SettingsResponse, Self::Err> {
        let settings = SettingsResponse {
            unit: CurrencyUnit::Msat.to_string(),
            bolt11: Some(payment::Bolt11Settings {
                mpp: false,
                amountless: true,
                invoice_description: true,
            }),
            bolt12: Some(payment::Bolt12Settings { amountless: true }),
            onchain: None,
            custom: std::collections::HashMap::new(),
        };
        Ok(settings)
    }

    /// Create a new invoice
    #[instrument(skip(self))]
    async fn create_incoming_payment_request(
        &self,
        options: IncomingPaymentOptions,
    ) -> Result<CreateIncomingPaymentResponse, Self::Err> {
        match options {
            IncomingPaymentOptions::Bolt11(bolt11_options) => {
                let amount_msat: Amount = bolt11_options
                    .amount
                    .convert_to(&CurrencyUnit::Msat)?
                    .into();
                let description = bolt11_options.description.unwrap_or_default();
                let time = match bolt11_options.unix_expiry {
                    Some(t) => t
                        .checked_sub(unix_time())
                        .ok_or(payment::Error::InvalidExpiry)?,
                    None => 36000,
                };

                let description = Bolt11InvoiceDescription::Direct(
                    Description::new(description).map_err(|_| Error::InvalidDescription)?,
                );

                let payment = self
                    .inner
                    .bolt11_payment()
                    .receive(amount_msat.into(), &description, time as u32)
                    .map_err(Error::LdkNode)?;

                let payment_hash = payment.payment_hash().to_string();
                let payment_identifier = PaymentIdentifier::PaymentHash(
                    hex::decode(&payment_hash)?
                        .try_into()
                        .map_err(|_| Error::InvalidPaymentHashLength)?,
                );

                Ok(CreateIncomingPaymentResponse {
                    request_lookup_id: payment_identifier,
                    request: payment.to_string(),
                    expiry: Some(unix_time() + time),
                    extra_json: None,
                })
            }
            IncomingPaymentOptions::Bolt12(bolt12_options) => {
                let Bolt12IncomingPaymentOptions {
                    description,
                    amount,
                    unix_expiry,
                } = *bolt12_options;

                let time = unix_expiry
                    .map(|t| {
                        t.checked_sub(unix_time())
                            .ok_or(payment::Error::InvalidExpiry)
                            .map(|t| t as u32)
                    })
                    .transpose()?;

                let offer = match amount {
                    Some(amount) => {
                        let amount_msat: Amount = amount.convert_to(&CurrencyUnit::Msat)?.into();

                        self.inner
                            .bolt12_payment()
                            .receive(
                                amount_msat.into(),
                                &description.unwrap_or("".to_string()),
                                time,
                                None,
                            )
                            .map_err(Error::LdkNode)?
                    }
                    None => self
                        .inner
                        .bolt12_payment()
                        .receive_variable_amount(&description.unwrap_or("".to_string()), time)
                        .map_err(Error::LdkNode)?,
                };
                let payment_identifier = PaymentIdentifier::OfferId(offer.id().to_string());

                Ok(CreateIncomingPaymentResponse {
                    request_lookup_id: payment_identifier,
                    request: offer.to_string(),
                    expiry: unix_expiry,
                    extra_json: None,
                })
            }
            IncomingPaymentOptions::Custom(_) | IncomingPaymentOptions::Onchain(_) => {
                Err(cdk_common::payment::Error::UnsupportedPaymentOption)
            }
        }
    }

    /// Get payment quote
    /// Used to get fee and amount required for a payment request
    #[instrument(skip_all)]
    async fn get_payment_quote(
        &self,
        unit: &CurrencyUnit,
        options: OutgoingPaymentOptions,
    ) -> Result<PaymentQuoteResponse, Self::Err> {
        match options {
            cdk_common::payment::OutgoingPaymentOptions::Custom(_) => {
                Err(cdk_common::payment::Error::UnsupportedPaymentOption)
            }
            OutgoingPaymentOptions::Bolt11(bolt11_options) => {
                let bolt11 = bolt11_options.bolt11;

                let amount_msat = match bolt11_options.melt_options {
                    Some(MeltOptions::Amountless { amountless }) => {
                        let amount_msat = amountless.amount_msat;

                        if let Some(invoice_amount) = bolt11.amount_milli_satoshis() {
                            if invoice_amount != u64::from(amount_msat) {
                                return Err(payment::Error::AmountMismatch);
                            }
                        }

                        amount_msat
                    }
                    Some(MeltOptions::Mpp { mpp }) => mpp.amount,
                    None => bolt11
                        .amount_milli_satoshis()
                        .ok_or(Error::UnknownInvoiceAmount)?
                        .into(),
                };

                let amount =
                    Amount::new(amount_msat.into(), CurrencyUnit::Msat).convert_to(unit)?;

                let relative_fee_reserve =
                    (self.fee_reserve.percent_fee_reserve * amount.value() as f32) as u64;

                let absolute_fee_reserve: u64 = self.fee_reserve.min_fee_reserve.into();

                let fee = match relative_fee_reserve > absolute_fee_reserve {
                    true => relative_fee_reserve,
                    false => absolute_fee_reserve,
                };

                let payment_hash = bolt11.payment_hash().to_string();
                let payment_hash_bytes = hex::decode(&payment_hash)?
                    .try_into()
                    .map_err(|_| Error::InvalidPaymentHashLength)?;

                Ok(PaymentQuoteResponse {
                    request_lookup_id: Some(PaymentIdentifier::PaymentHash(payment_hash_bytes)),
                    amount,
                    fee: Amount::new(fee, unit.clone()),
                    state: MeltQuoteState::Unpaid,
                    extra_json: None,
                    estimated_blocks: None,
                    fee_options: None,
                })
            }
            OutgoingPaymentOptions::Bolt12(bolt12_options) => {
                let offer = bolt12_options.offer;

                let amount_msat = match bolt12_options.melt_options {
                    Some(melt_options) => melt_options.amount_msat(),
                    None => {
                        let amount = offer.amount().ok_or(payment::Error::AmountMismatch)?;

                        match amount {
                            ldk_node::lightning::offers::offer::Amount::Bitcoin {
                                amount_msats,
                            } => amount_msats.into(),
                            _ => return Err(payment::Error::AmountMismatch),
                        }
                    }
                };
                let amount =
                    Amount::new(amount_msat.into(), CurrencyUnit::Msat).convert_to(unit)?;

                let relative_fee_reserve =
                    (self.fee_reserve.percent_fee_reserve * amount.value() as f32) as u64;

                let absolute_fee_reserve: u64 = self.fee_reserve.min_fee_reserve.into();

                let fee = match relative_fee_reserve > absolute_fee_reserve {
                    true => relative_fee_reserve,
                    false => absolute_fee_reserve,
                };

                Ok(PaymentQuoteResponse {
                    request_lookup_id: None,
                    amount,
                    fee: Amount::new(fee, unit.clone()),
                    state: MeltQuoteState::Unpaid,
                    extra_json: None,
                    estimated_blocks: None,
                    fee_options: None,
                })
            }
            OutgoingPaymentOptions::Onchain(_) => {
                Err(cdk_common::payment::Error::UnsupportedPaymentOption)
            }
        }
    }

    /// Pay request
    #[instrument(skip(self, options))]
    async fn make_payment(
        &self,
        unit: &CurrencyUnit,
        options: OutgoingPaymentOptions,
    ) -> Result<MakePaymentResponse, Self::Err> {
        match options {
            cdk_common::payment::OutgoingPaymentOptions::Custom(_) => {
                Err(cdk_common::payment::Error::UnsupportedPaymentOption)
            }
            OutgoingPaymentOptions::Bolt11(bolt11_options) => {
                let bolt11 = bolt11_options.bolt11;

                let send_params = match bolt11_options
                    .max_fee_amount
                    .map(|f| {
                        f.convert_to(&CurrencyUnit::Msat)
                            .map(|amount_msat| RouteParametersConfig {
                                max_total_routing_fee_msat: Some(amount_msat.value()),
                                ..Default::default()
                            })
                    })
                    .transpose()
                {
                    Ok(params) => params,
                    Err(err) => {
                        tracing::error!("Failed to convert fee amount: {}", err);
                        return Err(payment::Error::Custom(format!("Invalid fee amount: {err}")));
                    }
                };

                // Subscribe before dispatch so an immediately completed
                // payment cannot race ahead of the waiter.
                let payment_event_receiver = self.outgoing_payment_sender.subscribe();

                let payment_id = match bolt11_options.melt_options {
                    Some(MeltOptions::Amountless { amountless }) => {
                        if let Some(invoice_amount) = bolt11.amount_milli_satoshis() {
                            if invoice_amount != u64::from(amountless.amount_msat) {
                                return Err(payment::Error::AmountMismatch);
                            }
                        }

                        self.inner
                            .bolt11_payment()
                            .send_using_amount(&bolt11, amountless.amount_msat.into(), send_params)
                            .map_err(|err| {
                                tracing::error!("Could not send send amountless bolt11: {}", err);
                                Error::CouldNotSendBolt11WithoutAmount
                            })?
                    }
                    None => self
                        .inner
                        .bolt11_payment()
                        .send(&bolt11, send_params)
                        .map_err(|err| {
                            tracing::error!("Could not send bolt11 {}", err);
                            Error::CouldNotSendBolt11
                        })?,
                    _ => return Err(payment::Error::UnsupportedPaymentOption),
                };

                let payment_details = self
                    .wait_for_payment_terminal_status(payment_id, payment_event_receiver)
                    .await?;

                if payment_details.status == PaymentStatus::Failed {
                    tracing::error!(payment_id = %payment_id, "Bolt11 payment failed");
                }

                Self::make_payment_response_from_details(
                    unit,
                    PaymentIdentifier::PaymentHash(bolt11.payment_hash().to_byte_array()),
                    &payment_details,
                )
            }
            OutgoingPaymentOptions::Bolt12(bolt12_options) => {
                let offer = bolt12_options.offer;

                let send_params = match bolt12_options
                    .max_fee_amount
                    .map(|f| {
                        f.convert_to(&CurrencyUnit::Msat)
                            .map(|amount_msat| RouteParametersConfig {
                                max_total_routing_fee_msat: Some(amount_msat.value()),
                                ..Default::default()
                            })
                    })
                    .transpose()
                {
                    Ok(params) => params,
                    Err(err) => {
                        tracing::error!("Failed to convert fee amount: {}", err);
                        return Err(payment::Error::Custom(format!("Invalid fee amount: {err}")));
                    }
                };

                // BOLT12 payment ids are assigned by `send`, so subscribe
                // first and filter the queued terminal events once it returns.
                let payment_event_receiver = self.outgoing_payment_sender.subscribe();

                let payment_id = match bolt12_options.melt_options {
                    Some(MeltOptions::Amountless { amountless }) => self
                        .inner
                        .bolt12_payment()
                        .send_using_amount(
                            &offer,
                            amountless.amount_msat.into(),
                            None,
                            None,
                            send_params,
                        )
                        .map_err(Error::LdkNode)?,
                    None => self
                        .inner
                        .bolt12_payment()
                        .send(&offer, None, None, send_params)
                        .map_err(Error::LdkNode)?,
                    _ => return Err(payment::Error::UnsupportedPaymentOption),
                };

                let payment_details = self
                    .wait_for_payment_terminal_status(payment_id, payment_event_receiver)
                    .await?;

                if payment_details.status == PaymentStatus::Failed {
                    tracing::error!(
                        payment_id = %payment_id,
                        amount_msat = ?payment_details.amount_msat,
                        fee_paid_msat = ?payment_details.fee_paid_msat,
                        payment_kind = ?payment_details.kind,
                        "Bolt12 payment failed"
                    );
                }

                Self::make_payment_response_from_details(
                    unit,
                    PaymentIdentifier::PaymentId(payment_id.0),
                    &payment_details,
                )
            }
            OutgoingPaymentOptions::Onchain(_) => {
                Err(cdk_common::payment::Error::UnsupportedPaymentOption)
            }
        }
    }

    /// Listen for invoices to be paid to the mint
    /// Returns a stream of request_lookup_id once invoices are paid
    #[instrument(skip(self))]
    async fn wait_payment_event(
        &self,
    ) -> Result<Pin<Box<dyn Stream<Item = cdk_common::payment::Event> + Send>>, Self::Err> {
        tracing::info!("Starting stream for invoices - wait_any_incoming_payment called");

        // Set active flag to indicate stream is active
        self.wait_invoice_is_active.store(true, Ordering::SeqCst);
        tracing::debug!("wait_invoice_is_active set to true");

        let receiver = self.receiver.clone();

        tracing::info!("Receiver obtained successfully, creating response stream");

        // Transform the String stream into a WaitPaymentResponse stream
        let response_stream = BroadcastStream::new(receiver.resubscribe());

        // Map the stream to handle BroadcastStreamRecvError and wrap in Event
        let response_stream = response_stream.filter_map(|result| async move {
            match result {
                Ok(payment) => Some(cdk_common::payment::Event::PaymentReceived(payment)),
                Err(err) => {
                    tracing::warn!("Error in broadcast stream: {}", err);
                    None
                }
            }
        });

        // Create a combined stream that also handles cancellation
        let cancel_token = self.wait_invoice_cancel_token.clone();
        let is_active = self.wait_invoice_is_active.clone();

        let stream = Box::pin(response_stream);

        // Set up a task to clean up when the stream is dropped
        tokio::spawn(async move {
            cancel_token.cancelled().await;
            tracing::info!("wait_invoice stream cancelled");
            is_active.store(false, Ordering::SeqCst);
        });

        tracing::info!("wait_any_incoming_payment returning stream");
        Ok(stream)
    }

    /// Is payment event stream active
    fn is_payment_event_stream_active(&self) -> bool {
        self.wait_invoice_is_active.load(Ordering::SeqCst)
    }

    /// Cancel payment event stream
    fn cancel_payment_event_stream(&self) {
        self.wait_invoice_cancel_token.cancel()
    }

    /// Check the status of an incoming payment
    async fn check_incoming_payment_status(
        &self,
        payment_identifier: &PaymentIdentifier,
    ) -> Result<Vec<WaitPaymentResponse>, Self::Err> {
        // Bolt12 offers are identified by offer id and can be paid more than
        // once, so collect every settled inbound payment for the offer.
        if let PaymentIdentifier::OfferId(offer_id) = payment_identifier {
            let payments = self.inner.list_payments_with_filter(|p| {
                p.direction == PaymentDirection::Inbound
                    && p.status == PaymentStatus::Succeeded
                    && matches!(
                        &p.kind,
                        PaymentKind::Bolt12Offer { offer_id: oid, .. } if oid.to_string() == *offer_id
                    )
            });

            return Ok(payments
                .into_iter()
                .filter_map(|p| {
                    let payment_id = match &p.kind {
                        PaymentKind::Bolt12Offer {
                            hash: Some(hash), ..
                        } => hash.to_string(),
                        _ => {
                            tracing::warn!("Bolt12 payment for offer {} missing hash", offer_id);
                            return None;
                        }
                    };

                    Some(WaitPaymentResponse {
                        payment_identifier: payment_identifier.clone(),
                        payment_amount: Amount::new(p.amount_msat?, CurrencyUnit::Msat),
                        payment_id,
                    })
                })
                .collect());
        }

        let payment_id_str = match payment_identifier {
            PaymentIdentifier::PaymentHash(hash) => hex::encode(hash),
            PaymentIdentifier::CustomId(id) => id.clone(),
            _ => return Err(Error::UnsupportedPaymentIdentifierType.into()),
        };

        let payment_id = PaymentId(
            hex::decode(&payment_id_str)?
                .try_into()
                .map_err(|_| Error::InvalidPaymentIdLength)?,
        );

        let payment_details = self
            .inner
            .payment(&payment_id)
            .ok_or(Error::PaymentNotFound)?;

        if payment_details.direction == PaymentDirection::Outbound {
            return Err(Error::InvalidPaymentDirection.into());
        }

        let amount = if payment_details.status == PaymentStatus::Succeeded {
            payment_details
                .amount_msat
                .ok_or(Error::CouldNotGetPaymentAmount)?
        } else {
            return Ok(vec![]);
        };

        let response = WaitPaymentResponse {
            payment_identifier: payment_identifier.clone(),
            payment_amount: Amount::new(amount, CurrencyUnit::Msat),
            payment_id: payment_id_str,
        };

        Ok(vec![response])
    }

    /// Check the status of an outgoing payment
    async fn check_outgoing_payment(
        &self,
        request_lookup_id: &PaymentIdentifier,
    ) -> Result<MakePaymentResponse, Self::Err> {
        let payment_details = match request_lookup_id {
            PaymentIdentifier::PaymentHash(id_hash) => {
                Self::select_bolt11_payment_details(self.inner.list_payments_with_filter(|p| {
                    p.direction == PaymentDirection::Outbound
                        && matches!(&p.kind, PaymentKind::Bolt11 { hash, .. } if &hash.0 == id_hash)
                }))
            }
            PaymentIdentifier::PaymentId(id) => self.inner.payment(&PaymentId(*id)),
            _ => {
                return Ok(MakePaymentResponse {
                    payment_lookup_id: request_lookup_id.clone(),
                    payment_proof: None,
                    status: MeltQuoteState::Unknown,
                    total_spent: Amount::new(0, CurrencyUnit::Msat),
                });
            }
        }
        .ok_or(Error::PaymentNotFound)?;

        if payment_details.direction != PaymentDirection::Outbound {
            return Err(Error::InvalidPaymentDirection.into());
        }

        Self::make_payment_response_from_details(
            &CurrencyUnit::Msat,
            request_lookup_id.clone(),
            &payment_details,
        )
    }
}

impl Drop for CdkLdkNode {
    fn drop(&mut self) {
        tracing::info!("Drop called on CdkLdkNode");
        self.wait_invoice_cancel_token.cancel();
        tracing::debug!("Cancelled wait_invoice token in drop");
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn test_payment_details(status: PaymentStatus, amount_msat: Option<u64>) -> PaymentDetails {
        PaymentDetails {
            id: PaymentId([2; 32]),
            kind: PaymentKind::Bolt11 {
                hash: PaymentHash([1; 32]),
                preimage: None,
                secret: None,
            },
            amount_msat,
            fee_paid_msat: None,
            direction: PaymentDirection::Outbound,
            status,
            latest_update_timestamp: 0,
        }
    }

    fn test_payment_details_with_id(
        id: [u8; 32],
        status: PaymentStatus,
        latest_update_timestamp: u64,
    ) -> PaymentDetails {
        PaymentDetails {
            id: PaymentId(id),
            latest_update_timestamp,
            ..test_payment_details(status, None)
        }
    }

    #[test]
    fn failed_payment_response_does_not_require_amount() {
        let details = test_payment_details(PaymentStatus::Failed, None);

        let response = CdkLdkNode::make_payment_response_from_details(
            &CurrencyUnit::Msat,
            PaymentIdentifier::PaymentId([2; 32]),
            &details,
        )
        .expect("failed payment details should map without amount");

        assert_eq!(response.status, MeltQuoteState::Failed);
        assert_eq!(response.total_spent, Amount::new(0, CurrencyUnit::Msat));
    }

    #[test]
    fn pending_payment_response_does_not_require_amount() {
        let details = test_payment_details(PaymentStatus::Pending, None);

        let response = CdkLdkNode::make_payment_response_from_details(
            &CurrencyUnit::Msat,
            PaymentIdentifier::PaymentId([2; 32]),
            &details,
        )
        .expect("pending payment details should map without amount");

        assert_eq!(response.status, MeltQuoteState::Pending);
        assert_eq!(response.total_spent, Amount::new(0, CurrencyUnit::Msat));
    }

    #[test]
    fn paid_payment_response_requires_amount() {
        let details = test_payment_details(PaymentStatus::Succeeded, None);

        let err = CdkLdkNode::make_payment_response_from_details(
            &CurrencyUnit::Msat,
            PaymentIdentifier::PaymentId([2; 32]),
            &details,
        )
        .expect_err("paid payment details without amount should fail");

        assert!(matches!(err, payment::Error::Lightning(_)));
    }

    #[test]
    fn bolt11_payment_selection_prefers_pending_over_failed() {
        let failed = test_payment_details_with_id([1; 32], PaymentStatus::Failed, 2);
        let pending = test_payment_details_with_id([2; 32], PaymentStatus::Pending, 1);

        let selected = CdkLdkNode::select_bolt11_payment_details([failed, pending])
            .expect("payment details should be selected");

        assert_eq!(selected.id, PaymentId([2; 32]));
        assert_eq!(selected.status, PaymentStatus::Pending);
    }

    #[test]
    fn bolt11_payment_selection_prefers_succeeded_over_pending() {
        let pending = test_payment_details_with_id([1; 32], PaymentStatus::Pending, 2);
        let succeeded = PaymentDetails {
            amount_msat: Some(1000),
            ..test_payment_details_with_id([2; 32], PaymentStatus::Succeeded, 1)
        };

        let selected = CdkLdkNode::select_bolt11_payment_details([pending, succeeded])
            .expect("payment details should be selected");

        assert_eq!(selected.id, PaymentId([2; 32]));
        assert_eq!(selected.status, PaymentStatus::Succeeded);
    }

    #[test]
    fn bolt11_payment_selection_uses_latest_failed_when_all_failed() {
        let older_failed = test_payment_details_with_id([1; 32], PaymentStatus::Failed, 1);
        let newer_failed = test_payment_details_with_id([2; 32], PaymentStatus::Failed, 2);

        let selected = CdkLdkNode::select_bolt11_payment_details([older_failed, newer_failed])
            .expect("payment details should be selected");

        assert_eq!(selected.id, PaymentId([2; 32]));
        assert_eq!(selected.status, PaymentStatus::Failed);
    }

    #[tokio::test]
    async fn terminal_payment_event_wait_ignores_other_payments() {
        let (sender, mut receiver) = tokio::sync::broadcast::channel(4);
        let payment_id = PaymentId([2; 32]);

        // Queue both events before entering the wait to exercise the race where
        // LDK completes immediately after dispatch returns.
        sender
            .send(PaymentId([1; 32]))
            .expect("receiver should be subscribed");
        sender
            .send(payment_id)
            .expect("receiver should be subscribed");

        CdkLdkNode::wait_for_terminal_payment_event(&mut receiver, payment_id)
            .await
            .expect("matching terminal event should wake the waiter");
    }

    #[tokio::test]
    async fn terminal_payment_event_wait_recovers_from_lagged_channel() {
        let (sender, mut receiver) = tokio::sync::broadcast::channel(2);
        let payment_id = PaymentId([3; 32]);

        sender
            .send(PaymentId([1; 32]))
            .expect("receiver should be subscribed");
        sender
            .send(PaymentId([2; 32]))
            .expect("receiver should be subscribed");
        sender
            .send(PaymentId([4; 32]))
            .expect("receiver should be subscribed");
        sender
            .send(payment_id)
            .expect("receiver should be subscribed");

        CdkLdkNode::wait_for_terminal_payment_event(&mut receiver, payment_id)
            .await
            .expect("receiver lag should not prevent a matching event from waking the waiter");
    }

    #[tokio::test]
    async fn terminal_payment_event_wait_reports_closed_channel() {
        let (sender, mut receiver) = tokio::sync::broadcast::channel(1);
        drop(sender);

        let err = CdkLdkNode::wait_for_terminal_payment_event(&mut receiver, PaymentId([2; 32]))
            .await
            .expect_err("a closed event channel should stop the wait");

        assert!(matches!(
            err,
            tokio::sync::broadcast::error::RecvError::Closed
        ));
    }
}