cdk-ldk-node 0.18.0

//! CDK lightning backend for ldk-node

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

use std::fmt;
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::database::DynKVStore;
use cdk_common::payment::{self, *};
use cdk_common::redact::url_for_logs;
use cdk_common::util::{hex, unix_time};
use cdk_common::{Amount, CurrencyUnit, MeltOptions, MeltQuoteState, QuoteId};
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;

/// Primary KV namespace for the ldk-node backend's durable bookkeeping
const LDK_KV_PRIMARY_NAMESPACE: &str = "cdk_ldk_node_lightning_backend";
/// Secondary KV namespace holding the bolt12 melt quote id -> payment id
/// mapping used to resolve `PaymentIdentifier::QuoteId` lookups
const LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE: &str = "bolt12_outgoing_payments";
/// 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;
const LDK_KV_BOLT12_CLEANUP_MARKER: &[u8] = b"cleanup-in-progress";

/// Result of looking up the payment id recorded for a bolt12 melt quote
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Bolt12QuotePaymentIdLookup {
    /// A payment id was recorded: the payment was dispatched and is tracked
    Found(PaymentId),
    /// The dispatch sentinel is present: `send` was started but no payment id
    /// was recorded (crash during dispatch, or dispatch errored before the
    /// sentinel could be cleaned up). The payment state is indeterminate.
    Dispatching,
    /// No record exists: the payment was never dispatched
    Missing,
    /// A record exists but cannot be parsed
    Malformed,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Bolt12QuotePaymentIdResolution {
    PaymentId(PaymentId),
    Status(MeltQuoteState),
}

impl Bolt12QuotePaymentIdLookup {
    fn resolve(self) -> Bolt12QuotePaymentIdResolution {
        match self {
            Self::Found(payment_id) => Bolt12QuotePaymentIdResolution::PaymentId(payment_id),
            // Dispatch was attempted but no payment id was recorded. Pending
            // prevents the live melt saga from compensating an indeterminate
            // payment after a dispatch-ambiguous error.
            Self::Dispatching => Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Pending),
            // Without the pre-dispatch sentinel, the payment was never sent.
            Self::Missing => Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Unpaid),
            // Corrupt bookkeeping cannot establish any payment state.
            Self::Malformed => Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Unknown),
        }
    }
}

/// Whether an LDK BOLT12 send error can occur after dispatch was accepted.
///
/// In `ldk-node` 0.7, [`ldk_node::NodeError::PersistenceFailed`] can be returned
/// while persisting the payment record after `ChannelManager::pay_for_offer`
/// accepted the payment. All other errors returned by the BOLT12 send methods
/// occur before dispatch or after `pay_for_offer` rejected the attempt.
fn bolt12_send_error_has_ambiguous_dispatch(err: &ldk_node::NodeError) -> bool {
    matches!(err, ldk_node::NodeError::PersistenceFailed)
}

/// Whether a BOLT11 send error authoritatively proves that this invocation
/// cannot settle.
///
/// `PersistenceFailed` is ambiguous because `ldk-node` may return it after the
/// channel manager accepted the payment but before the pending payment record
/// was persisted. `DuplicatePayment` is also non-terminal for this invocation:
/// the existing payment may already be pending or succeeded. The remaining
/// errors listed here are returned only when dispatch was rejected.
fn bolt11_send_error_is_explicit_terminal_failure(err: &ldk_node::NodeError) -> bool {
    matches!(
        err,
        ldk_node::NodeError::NotRunning
            | ldk_node::NodeError::InvalidAmount
            | ldk_node::NodeError::InvalidInvoice
            | ldk_node::NodeError::PaymentSendingFailed
    )
}

fn outgoing_payment_failure_response(
    unit: &CurrencyUnit,
    payment_lookup_id: PaymentIdentifier,
) -> MakePaymentResponse {
    MakePaymentResponse {
        payment_lookup_id,
        payment_proof: None,
        status: MeltQuoteState::Failed,
        total_spent: Amount::new(0, unit.clone()),
    }
}

/// 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,
    kv_store: DynKVStore,
    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 fmt::Debug for CdkLdkNode {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> 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(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,
}

impl fmt::Debug for BitcoinRpcConfig {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("BitcoinRpcConfig")
            .field("host", &self.host)
            .field("port", &self.port)
            .field("user", &self.user)
            .field("password", &"[REDACTED]")
            .finish()
    }
}

/// 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(Clone)]
pub enum ChainSource {
    /// Use an Esplora server for blockchain data
    ///
    /// Contains the URL of the Esplora server endpoint
    Esplora(String),
    /// Use an Electrum server for blockchain data
    ///
    /// Contains the URL of the Electrum server endpoint
    Electrum(String),
    /// Use Bitcoin Core RPC for blockchain data
    ///
    /// Contains the configuration for connecting to Bitcoin Core
    BitcoinRpc(BitcoinRpcConfig),
}

impl fmt::Debug for ChainSource {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Esplora(url) => f.debug_tuple("Esplora").field(&url_for_logs(url)).finish(),
            Self::Electrum(url) => f.debug_tuple("Electrum").field(&url_for_logs(url)).finish(),
            Self::BitcoinRpc(config) => f.debug_tuple("BitcoinRpc").field(config).finish(),
        }
    }
}

/// Source of Lightning network gossip data
///
/// Specifies how the node should learn about the Lightning Network topology
/// and routing information.
#[derive(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),
}

impl fmt::Debug for GossipSource {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::P2P => f.write_str("P2P"),
            Self::RapidGossipSync(url) => f
                .debug_tuple("RapidGossipSync")
                .field(&url_for_logs(url))
                .finish(),
        }
    }
}
/// A builder for an [`CdkLdkNode`] instance.
pub struct CdkLdkNodeBuilder {
    network: Network,
    chain_source: ChainSource,
    gossip_source: GossipSource,
    log_dir_path: Option<String>,
    storage_dir_path: String,
    fee_reserve: FeeReserve,
    kv_store: DynKVStore,
    listening_addresses: Vec<SocketAddress>,
    seed: Option<Mnemonic>,
    announcement_addresses: Option<Vec<SocketAddress>>,
}

impl std::fmt::Debug for CdkLdkNodeBuilder {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CdkLdkNodeBuilder")
            .field("network", &self.network)
            .field("chain_source", &self.chain_source)
            .field("gossip_source", &self.gossip_source)
            .field("log_dir_path", &self.log_dir_path)
            .field("storage_dir_path", &self.storage_dir_path)
            .field("fee_reserve", &self.fee_reserve)
            .field("listening_addresses", &self.listening_addresses)
            .field("announcement_addresses", &self.announcement_addresses)
            .finish_non_exhaustive()
    }
}

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>,
        kv_store: DynKVStore,
    ) -> Self {
        Self {
            network,
            chain_source,
            gossip_source,
            storage_dir_path,
            fee_reserve,
            kv_store,
            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::Electrum(electrum_url) => {
                ldk.set_chain_source_electrum(electrum_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,
            kv_store: self.kv_store,
            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))
    }

    /// Best-effort release of an exact sentinel or payment-id binding that is
    /// known not to represent an in-flight or successful payment.
    async fn cleanup_bolt12_dispatch_binding(
        &self,
        quote_id: &QuoteId,
        payment_id: Option<&PaymentId>,
    ) {
        match delete_bolt12_quote_payment_id_if_equals(&self.kv_store, quote_id, payment_id).await {
            Ok(true) => {}
            Ok(false) => {
                tracing::debug!(
                    quote_id = %quote_id,
                    "BOLT12 dispatch binding changed before cleanup"
                );
            }
            Err(err) => {
                tracing::warn!(
                    quote_id = %quote_id,
                    "Could not release BOLT12 dispatch binding: {err}"
                );
            }
        }
    }

    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,
                invoice_description: 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: Some(PaymentIdentifier::QuoteId(
                        bolt12_options.quote_id.clone(),
                    )),
                    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(options) => {
                Ok(outgoing_payment_failure_response(
                    unit,
                    PaymentIdentifier::QuoteId(options.quote_id),
                ))
            }
            OutgoingPaymentOptions::Bolt11(bolt11_options) => {
                let bolt11 = bolt11_options.bolt11;
                let payment_lookup_id =
                    PaymentIdentifier::PaymentHash(bolt11.payment_hash().to_byte_array());

                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 Ok(outgoing_payment_failure_response(unit, payment_lookup_id));
                    }
                };

                // 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 Ok(outgoing_payment_failure_response(
                                    unit,
                                    payment_lookup_id,
                                ));
                            }
                        }

                        self.inner.bolt11_payment().send_using_amount(
                            &bolt11,
                            amountless.amount_msat.into(),
                            send_params,
                        )
                    }
                    None => self.inner.bolt11_payment().send(&bolt11, send_params),
                    _ => {
                        return Ok(outgoing_payment_failure_response(unit, payment_lookup_id));
                    }
                };

                let payment_id = match payment_id {
                    Ok(payment_id) => payment_id,
                    Err(err) if bolt11_send_error_is_explicit_terminal_failure(&err) => {
                        tracing::warn!(
                            payment_hash = %bolt11.payment_hash(),
                            "LDK rejected BOLT11 payment before dispatch: {err}"
                        );
                        return Ok(outgoing_payment_failure_response(unit, payment_lookup_id));
                    }
                    Err(err) => {
                        tracing::warn!(
                            payment_hash = %bolt11.payment_hash(),
                            "LDK BOLT11 send outcome is indeterminate: {err}"
                        );
                        return Err(Error::LdkNode(err).into());
                    }
                };

                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, payment_lookup_id, &payment_details)
            }
            OutgoingPaymentOptions::Bolt12(bolt12_options) => {
                let offer = bolt12_options.offer;
                let quote_id = bolt12_options.quote_id.clone();
                let quote_payment_identifier = PaymentIdentifier::QuoteId(quote_id.clone());

                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 Ok(outgoing_payment_failure_response(
                            unit,
                            quote_payment_identifier,
                        ));
                    }
                };

                // Claim the quote with an absent-only sentinel write before
                // attempting the send: a duplicate or concurrent dispatch for
                // the same quote is rejected here, before any funds move. The
                // sentinel also keeps a crash during dispatch distinguishable
                // from "never dispatched" (mirrors cdk-cln's pre-dispatch
                // bolt12 quote mapping). The payment must not be attempted
                // unless this marker is durable.
                if let Err(err) =
                    write_bolt12_quote_payment_id(&self.kv_store, &quote_id, None).await
                {
                    tracing::error!(
                        quote_id = %quote_id,
                        "Could not persist BOLT12 dispatch claim before sending: {err}"
                    );
                    return Ok(outgoing_payment_failure_response(
                        unit,
                        quote_payment_identifier,
                    ));
                }

                // 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,
                        )
                    }
                    None => self
                        .inner
                        .bolt12_payment()
                        .send(&offer, None, None, send_params),
                    _ => {
                        self.cleanup_bolt12_dispatch_binding(&quote_id, None).await;
                        return Ok(outgoing_payment_failure_response(
                            unit,
                            quote_payment_identifier,
                        ));
                    }
                };

                let payment_id = match payment_id {
                    Ok(payment_id) => payment_id,
                    Err(err) => {
                        match bolt12_send_error_has_ambiguous_dispatch(&err) {
                            true => {
                                tracing::warn!(
                                    quote_id = %quote_id,
                                    "LDK payment persistence failed after BOLT12 send; retaining \
                                     the dispatch sentinel because the payment may have been dispatched"
                                );
                            }
                            false => {
                                self.cleanup_bolt12_dispatch_binding(&quote_id, None).await;
                                tracing::warn!(
                                    quote_id = %quote_id,
                                    "LDK rejected BOLT12 payment before dispatch: {err}"
                                );
                                return Ok(outgoing_payment_failure_response(
                                    unit,
                                    quote_payment_identifier,
                                ));
                            }
                        }
                        return Err(Error::LdkNode(err).into());
                    }
                };

                // Record the payment id so QuoteId lookups resolve to the
                // dispatched payment. The write is conditional on owning the
                // dispatch claim. Best-effort: if this write fails the
                // sentinel remains and the payment resolves as Pending, keeping
                // the melt proofs reserved.
                if let Err(err) =
                    write_bolt12_quote_payment_id(&self.kv_store, &quote_id, Some(&payment_id))
                        .await
                {
                    tracing::error!(
                        "Could not record BOLT12 payment id for quote {quote_id}: {err}. \
                         The payment will remain Pending until manual intervention."
                    );
                }

                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.cleanup_bolt12_dispatch_binding(&quote_id, Some(&payment_id))
                        .await;
                }

                Self::make_payment_response_from_details(
                    unit,
                    quote_payment_identifier,
                    &payment_details,
                )
            }
            OutgoingPaymentOptions::Onchain(options) => Ok(outgoing_payment_failure_response(
                unit,
                PaymentIdentifier::QuoteId(options.quote_id),
            )),
        }
    }

    /// 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)),
            PaymentIdentifier::QuoteId(quote_id) => {
                match read_bolt12_quote_payment_id(&self.kv_store, quote_id)
                    .await?
                    .resolve()
                {
                    Bolt12QuotePaymentIdResolution::PaymentId(payment_id) => {
                        self.inner.payment(&payment_id)
                    }
                    Bolt12QuotePaymentIdResolution::Status(status) => {
                        return Ok(MakePaymentResponse {
                            payment_lookup_id: request_lookup_id.clone(),
                            payment_proof: None,
                            status,
                            total_spent: Amount::new(0, CurrencyUnit::Msat),
                        });
                    }
                }
            }
            _ => {
                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());
        }

        if payment_details.status == PaymentStatus::Failed {
            if let PaymentIdentifier::QuoteId(quote_id) = request_lookup_id {
                self.cleanup_bolt12_dispatch_binding(quote_id, Some(&payment_details.id))
                    .await;
            }
        }

        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");
    }
}

/// KV key for the bolt12 melt quote id -> payment id mapping
fn bolt12_quote_payment_id_key(quote_id: &QuoteId) -> Result<String, Error> {
    match quote_id {
        QuoteId::UUID(uuid) => Ok(uuid.to_string()),
        QuoteId::BASE64(_) => Err(Error::InvalidQuoteId),
    }
}

/// Records the bolt12 melt quote id -> payment id mapping.
///
/// `payment_id` of `None` atomically claims an absent quote with the
/// pre-dispatch sentinel. `Some` atomically replaces that sentinel with the
/// dispatched payment id. Repeating the same payment id is idempotent; every
/// other existing binding is rejected.
async fn write_bolt12_quote_payment_id(
    kv_store: &DynKVStore,
    quote_id: &QuoteId,
    payment_id: Option<&PaymentId>,
) -> Result<(), Error> {
    let key = bolt12_quote_payment_id_key(quote_id)?;
    let value = payment_id.map(|id| hex::encode(id.0)).unwrap_or_default();
    let mut tx = kv_store
        .begin_transaction()
        .await
        .map_err(|e| Error::Database(e.to_string()))?;

    let written = match payment_id {
        None => {
            tx.kv_write_if_absent(
                LDK_KV_PRIMARY_NAMESPACE,
                LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
                &key,
                value.as_bytes(),
            )
            .await
        }
        Some(_) => {
            tx.kv_write_if_equals(
                LDK_KV_PRIMARY_NAMESPACE,
                LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
                &key,
                b"",
                value.as_bytes(),
            )
            .await
        }
    }
    .map_err(|e| Error::Database(e.to_string()))?;

    if written {
        tx.commit()
            .await
            .map_err(|e| Error::Database(e.to_string()))?;
        return Ok(());
    }

    let existing = tx
        .kv_read(
            LDK_KV_PRIMARY_NAMESPACE,
            LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
            &key,
        )
        .await
        .map_err(|e| Error::Database(e.to_string()))?;
    tx.rollback()
        .await
        .map_err(|e| Error::Database(e.to_string()))?;

    match existing {
        Some(existing) if payment_id.is_some() && existing.as_slice() == value.as_bytes() => Ok(()),
        _ => Err(Error::Bolt12QuoteAlreadyClaimed {
            quote_id: quote_id.to_string(),
        }),
    }
}

/// Reads the bolt12 melt quote id -> payment id mapping
async fn read_bolt12_quote_payment_id(
    kv_store: &DynKVStore,
    quote_id: &QuoteId,
) -> Result<Bolt12QuotePaymentIdLookup, Error> {
    let key = bolt12_quote_payment_id_key(quote_id)?;
    let Some(stored) = kv_store
        .kv_read(
            LDK_KV_PRIMARY_NAMESPACE,
            LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
            &key,
        )
        .await
        .map_err(|e| Error::Database(e.to_string()))?
    else {
        return Ok(Bolt12QuotePaymentIdLookup::Missing);
    };

    if stored.is_empty() {
        return Ok(Bolt12QuotePaymentIdLookup::Dispatching);
    }

    let payment_id_hex = match String::from_utf8(stored) {
        Ok(payment_id_hex) => payment_id_hex,
        Err(err) => {
            tracing::warn!(
                "LDK: invalid UTF-8 in BOLT12 payment id mapping for quote {quote_id}: {err}"
            );
            return Ok(Bolt12QuotePaymentIdLookup::Malformed);
        }
    };

    let payment_id_bytes = match hex::decode(&payment_id_hex) {
        Ok(bytes) => bytes,
        Err(err) => {
            tracing::warn!(
                "LDK: invalid hex in BOLT12 payment id mapping for quote {quote_id}: {err}"
            );
            return Ok(Bolt12QuotePaymentIdLookup::Malformed);
        }
    };

    let payment_id: [u8; 32] = match payment_id_bytes.try_into() {
        Ok(payment_id) => payment_id,
        Err(_) => {
            tracing::warn!("LDK: invalid payment id length in BOLT12 mapping for quote {quote_id}");
            return Ok(Bolt12QuotePaymentIdLookup::Malformed);
        }
    };

    Ok(Bolt12QuotePaymentIdLookup::Found(PaymentId(payment_id)))
}

/// Atomically removes the bolt12 quote binding only if it still matches the
/// expected sentinel (`None`) or payment id (`Some`).
async fn delete_bolt12_quote_payment_id_if_equals(
    kv_store: &DynKVStore,
    quote_id: &QuoteId,
    payment_id: Option<&PaymentId>,
) -> Result<bool, Error> {
    let key = bolt12_quote_payment_id_key(quote_id)?;
    let expected = payment_id.map(|id| hex::encode(id.0)).unwrap_or_default();
    let mut tx = kv_store
        .begin_transaction()
        .await
        .map_err(|e| Error::Database(e.to_string()))?;

    let claimed = tx
        .kv_write_if_equals(
            LDK_KV_PRIMARY_NAMESPACE,
            LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
            &key,
            expected.as_bytes(),
            LDK_KV_BOLT12_CLEANUP_MARKER,
        )
        .await
        .map_err(|e| Error::Database(e.to_string()))?;

    if !claimed {
        tx.rollback()
            .await
            .map_err(|e| Error::Database(e.to_string()))?;
        return Ok(false);
    }

    tx.kv_remove(
        LDK_KV_PRIMARY_NAMESPACE,
        LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
        &key,
    )
    .await
    .map_err(|e| Error::Database(e.to_string()))?;
    tx.commit()
        .await
        .map_err(|e| Error::Database(e.to_string()))?;

    Ok(true)
}

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

    #[test]
    fn bitcoin_rpc_debug_redacts_password() {
        let source = ChainSource::BitcoinRpc(BitcoinRpcConfig {
            host: "127.0.0.1".to_string(),
            port: 8332,
            user: "rpc-user".to_string(),
            password: "rpc-password-secret".to_string(),
        });

        let debug = format!("{source:?}");

        assert!(debug.contains("127.0.0.1"));
        assert!(debug.contains("rpc-user"));
        assert!(debug.contains("[REDACTED]"));
        assert!(!debug.contains("rpc-password-secret"));
    }

    #[test]
    fn chain_source_debug_redacts_url_credentials() {
        for source in [
            ChainSource::Esplora("https://esplora-user:esplora-secret@example.com/api".to_string()),
            ChainSource::Electrum(
                "ssl://electrum-user:electrum-secret@example.com:50002".to_string(),
            ),
        ] {
            let debug = format!("{source:?}");

            assert!(debug.contains("example.com"));
            assert!(!debug.contains("-user"));
            assert!(!debug.contains("-secret"));
        }
    }

    #[test]
    fn gossip_source_debug_redacts_url_credentials() {
        let source = GossipSource::RapidGossipSync(
            "https://rgs-user:rgs-secret@example.com/snapshot".to_string(),
        );

        let debug = format!("{source:?}");

        assert!(debug.contains("https://example.com/snapshot"));
        assert!(!debug.contains("rgs-user"));
        assert!(!debug.contains("rgs-secret"));
    }

    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::Backend(_)));
    }

    #[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
        ));
    }

    #[test]
    fn bolt12_persistence_failure_has_ambiguous_dispatch() {
        assert!(bolt12_send_error_has_ambiguous_dispatch(
            &ldk_node::NodeError::PersistenceFailed
        ));

        for not_dispatched in [
            ldk_node::NodeError::NotRunning,
            ldk_node::NodeError::UnsupportedCurrency,
            ldk_node::NodeError::InvalidOffer,
            ldk_node::NodeError::InvalidAmount,
            ldk_node::NodeError::DuplicatePayment,
            ldk_node::NodeError::InvoiceRequestCreationFailed,
            ldk_node::NodeError::PaymentSendingFailed,
        ] {
            assert!(
                !bolt12_send_error_has_ambiguous_dispatch(&not_dispatched),
                "{not_dispatched} must be treated as not dispatched"
            );
        }
    }

    #[test]
    fn bolt11_send_errors_only_classify_explicit_rejections_as_terminal() {
        for terminal_error in [
            ldk_node::NodeError::NotRunning,
            ldk_node::NodeError::InvalidAmount,
            ldk_node::NodeError::InvalidInvoice,
            ldk_node::NodeError::PaymentSendingFailed,
        ] {
            assert!(
                bolt11_send_error_is_explicit_terminal_failure(&terminal_error),
                "{terminal_error} must be treated as a definitive failure"
            );
        }

        for ambiguous_error in [
            ldk_node::NodeError::PersistenceFailed,
            ldk_node::NodeError::DuplicatePayment,
        ] {
            assert!(
                !bolt11_send_error_is_explicit_terminal_failure(&ambiguous_error),
                "{ambiguous_error} must not authorize proof release"
            );
        }
    }

    #[test]
    fn authoritative_outgoing_failure_response_is_terminal_and_spends_nothing() {
        let payment_lookup_id = PaymentIdentifier::PaymentHash([42; 32]);
        let response =
            outgoing_payment_failure_response(&CurrencyUnit::Msat, payment_lookup_id.clone());

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

    #[test]
    fn bolt12_quote_payment_id_lookup_resolution_is_safe() {
        assert_eq!(
            Bolt12QuotePaymentIdLookup::Dispatching.resolve(),
            Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Pending),
            "an indeterminate dispatch must keep melt proofs reserved"
        );
        assert_eq!(
            Bolt12QuotePaymentIdLookup::Missing.resolve(),
            Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Unpaid),
            "a missing sentinel means the payment was never dispatched"
        );
        assert_eq!(
            Bolt12QuotePaymentIdLookup::Malformed.resolve(),
            Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Unknown),
            "corrupt bookkeeping must remain indeterminate"
        );
    }

    async fn test_kv_store() -> DynKVStore {
        std::sync::Arc::new(cdk_sqlite::mint::memory::empty().await.unwrap())
    }

    /// The mapping must resolve Missing before any dispatch, Dispatching
    /// (indeterminate) while only the pre-dispatch sentinel exists, and Found
    /// after the payment id is recorded.
    #[tokio::test]
    async fn bolt12_quote_payment_id_mapping_lifecycle() {
        let kv_store = test_kv_store().await;
        let quote_id = QuoteId::new();

        assert_eq!(
            read_bolt12_quote_payment_id(&kv_store, &quote_id)
                .await
                .unwrap(),
            Bolt12QuotePaymentIdLookup::Missing,
            "no record must resolve as never dispatched"
        );

        // Pre-dispatch sentinel
        write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
            .await
            .unwrap();
        assert_eq!(
            read_bolt12_quote_payment_id(&kv_store, &quote_id)
                .await
                .unwrap(),
            Bolt12QuotePaymentIdLookup::Dispatching,
            "sentinel must resolve as indeterminate, never terminal"
        );

        assert!(
            delete_bolt12_quote_payment_id_if_equals(&kv_store, &quote_id, None)
                .await
                .unwrap(),
            "an unambiguous pre-dispatch failure should release its sentinel"
        );
        write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
            .await
            .expect("a retry should reclaim the quote after sentinel cleanup");

        // Record the payment id
        let payment_id = PaymentId([7; 32]);
        write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&payment_id))
            .await
            .unwrap();
        assert_eq!(
            read_bolt12_quote_payment_id(&kv_store, &quote_id)
                .await
                .unwrap(),
            Bolt12QuotePaymentIdLookup::Found(payment_id)
        );

        // Removal returns to Missing (failed dispatch cleanup)
        assert!(
            delete_bolt12_quote_payment_id_if_equals(&kv_store, &quote_id, Some(&payment_id))
                .await
                .unwrap()
        );
        assert_eq!(
            read_bolt12_quote_payment_id(&kv_store, &quote_id)
                .await
                .unwrap(),
            Bolt12QuotePaymentIdLookup::Missing
        );
    }

    #[tokio::test]
    async fn bolt12_quote_payment_id_binding_is_write_once() {
        let kv_store = test_kv_store().await;
        let quote_id = QuoteId::new();
        let payment_id = PaymentId([7; 32]);
        let conflicting_payment_id = PaymentId([9; 32]);

        write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
            .await
            .expect("first dispatch should claim the quote");
        write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&payment_id))
            .await
            .expect("the claim owner should record its payment id");
        write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&payment_id))
            .await
            .expect("repeating the same payment id must be idempotent");

        let duplicate_dispatch = write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
            .await
            .expect_err("a dispatched quote must not be claimed again");
        assert!(
            matches!(duplicate_dispatch, Error::Bolt12QuoteAlreadyClaimed { .. }),
            "unexpected error: {duplicate_dispatch}"
        );

        let conflicting_binding =
            write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&conflicting_payment_id))
                .await
                .expect_err("a conflicting payment id must be rejected");
        assert!(
            matches!(conflicting_binding, Error::Bolt12QuoteAlreadyClaimed { .. }),
            "unexpected error: {conflicting_binding}"
        );

        assert_eq!(
            read_bolt12_quote_payment_id(&kv_store, &quote_id)
                .await
                .expect("payment id should remain readable"),
            Bolt12QuotePaymentIdLookup::Found(payment_id),
            "a duplicate dispatch must not redirect recovery"
        );
    }

    #[tokio::test]
    async fn failed_bolt12_binding_can_be_released_without_removing_a_retry() {
        let kv_store = test_kv_store().await;
        let quote_id = QuoteId::new();
        let failed_payment_id = PaymentId([7; 32]);
        let retry_payment_id = PaymentId([9; 32]);

        write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
            .await
            .expect("failed dispatch should claim the quote");
        write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&failed_payment_id))
            .await
            .expect("failed payment id should be recorded");
        assert!(delete_bolt12_quote_payment_id_if_equals(
            &kv_store,
            &quote_id,
            Some(&failed_payment_id),
        )
        .await
        .expect("failed binding should be released"));

        write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
            .await
            .expect("retry should claim the released quote");
        write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&retry_payment_id))
            .await
            .expect("retry payment id should be recorded");
        assert!(!delete_bolt12_quote_payment_id_if_equals(
            &kv_store,
            &quote_id,
            Some(&failed_payment_id),
        )
        .await
        .expect("stale cleanup should be checked atomically"));

        assert_eq!(
            read_bolt12_quote_payment_id(&kv_store, &quote_id)
                .await
                .expect("retry binding should remain readable"),
            Bolt12QuotePaymentIdLookup::Found(retry_payment_id),
            "stale failed-payment cleanup must not remove a newer retry"
        );
    }

    #[tokio::test]
    async fn bolt12_quote_dispatch_concurrent_claims_have_single_winner() {
        let kv_store = test_kv_store().await;
        let quote_id = QuoteId::new();

        let (first_result, second_result) = tokio::join!(
            write_bolt12_quote_payment_id(&kv_store, &quote_id, None),
            write_bolt12_quote_payment_id(&kv_store, &quote_id, None),
        );

        let outcomes = [first_result, second_result];
        let winners = outcomes.iter().filter(|result| result.is_ok()).count();
        let conflicts = outcomes
            .iter()
            .filter(|result| matches!(result, Err(Error::Bolt12QuoteAlreadyClaimed { .. })))
            .count();

        assert_eq!(winners, 1, "exactly one dispatch may claim the quote");
        assert_eq!(conflicts, 1, "the losing dispatch must be rejected");
    }

    #[tokio::test]
    async fn bolt12_quote_payment_id_concurrent_resolution_has_single_winner() {
        let kv_store = test_kv_store().await;
        let quote_id = QuoteId::new();
        let first_payment_id = PaymentId([7; 32]);
        let second_payment_id = PaymentId([9; 32]);

        write_bolt12_quote_payment_id(&kv_store, &quote_id, None)
            .await
            .expect("dispatch should claim the quote");

        let (first_result, second_result) = tokio::join!(
            write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&first_payment_id)),
            write_bolt12_quote_payment_id(&kv_store, &quote_id, Some(&second_payment_id)),
        );

        let outcomes = [&first_result, &second_result];
        let winners = outcomes.iter().filter(|result| result.is_ok()).count();
        let conflicts = outcomes
            .iter()
            .filter(|result| matches!(result, Err(Error::Bolt12QuoteAlreadyClaimed { .. })))
            .count();

        assert_eq!(winners, 1, "exactly one payment id may resolve the claim");
        assert_eq!(conflicts, 1, "the losing resolution must be rejected");

        let winner = if first_result.is_ok() {
            first_payment_id
        } else {
            second_payment_id
        };
        assert_eq!(
            read_bolt12_quote_payment_id(&kv_store, &quote_id)
                .await
                .expect("payment id should remain readable"),
            Bolt12QuotePaymentIdLookup::Found(winner)
        );
    }

    /// A corrupted mapping must resolve as indeterminate (`Malformed`), never
    /// as a terminal state that could trigger compensation.
    #[tokio::test]
    async fn bolt12_quote_payment_id_mapping_malformed_is_indeterminate() {
        let kv_store = test_kv_store().await;
        let quote_id = QuoteId::new();
        let key = bolt12_quote_payment_id_key(&quote_id).unwrap();

        for corrupt in ["not-hex", "0102", "zz"] {
            let mut tx = kv_store.begin_transaction().await.unwrap();
            tx.kv_write(
                LDK_KV_PRIMARY_NAMESPACE,
                LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
                &key,
                corrupt.as_bytes(),
            )
            .await
            .unwrap();
            tx.commit().await.unwrap();

            assert_eq!(
                read_bolt12_quote_payment_id(&kv_store, &quote_id)
                    .await
                    .unwrap(),
                Bolt12QuotePaymentIdLookup::Malformed,
                "corrupt value {corrupt} must be indeterminate"
            );
        }
    }
}