cdk-lnd 0.18.1

//! CDK lightning backend for LND

// Copyright (c) 2023 Steffen (MIT)

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

use std::cmp::max;
use std::path::PathBuf;
use std::pin::Pin;
use std::str::FromStr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

use anyhow::anyhow;
use async_trait::async_trait;
use cdk_common::amount::{Amount, MSAT_IN_SAT};
use cdk_common::bitcoin::hashes::Hash;
use cdk_common::common::FeeReserve;
use cdk_common::database::DynKVStore;
use cdk_common::nuts::{CurrencyUnit, MeltOptions, MeltQuoteState};
use cdk_common::payment::{
    self, CreateIncomingPaymentResponse, Event, IncomingPaymentOptions, MakePaymentResponse,
    MintPayment, OutgoingPaymentOptions, PaymentIdentifier, PaymentQuoteResponse, SettingsResponse,
    WaitPaymentResponse,
};
use cdk_common::util::{hex, unix_time};
use cdk_common::Bolt11Invoice;
use error::Error;
use futures::{Stream, StreamExt};
use lnrpc::fee_limit::Limit;
use lnrpc::payment::PaymentStatus;
use lnrpc::{FeeLimit, Hop, MppRecord};
use tokio_util::sync::CancellationToken;
use tracing::instrument;

mod client;
pub mod error;

mod proto;
pub(crate) use proto::{lnrpc, routerrpc};

use crate::lnrpc::invoice::InvoiceState;

/// LND KV Store constants
const LND_KV_PRIMARY_NAMESPACE: &str = "cdk_lnd_lightning_backend";
const LND_KV_SECONDARY_NAMESPACE: &str = "payment_indices";
const LAST_ADD_INDEX_KV_KEY: &str = "last_add_index";
const LAST_SETTLE_INDEX_KV_KEY: &str = "last_settle_index";

/// Lnd mint backend
#[derive(Clone)]
pub struct Lnd {
    _address: String,
    _cert_file: PathBuf,
    _macaroon_file: PathBuf,
    lnd_client: client::Client,
    fee_reserve: FeeReserve,
    kv_store: DynKVStore,
    wait_invoice_cancel_token: CancellationToken,
    wait_invoice_is_active: Arc<AtomicBool>,
    settings: SettingsResponse,
    unit: CurrencyUnit,
}

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

impl Lnd {
    /// Maximum number of attempts at a partial payment
    pub const MAX_ROUTE_RETRIES: usize = 50;

    /// Create new [`Lnd`]
    pub async fn new(
        address: String,
        cert_file: PathBuf,
        macaroon_file: PathBuf,
        fee_reserve: FeeReserve,
        kv_store: DynKVStore,
    ) -> Result<Self, Error> {
        // Validate address is not empty
        if address.is_empty() {
            return Err(Error::InvalidConfig("LND address cannot be empty".into()));
        }

        // Validate cert_file exists and is not empty
        if !cert_file.exists() || cert_file.metadata().map(|m| m.len() == 0).unwrap_or(true) {
            return Err(Error::InvalidConfig(format!(
                "LND certificate file not found or empty: {cert_file:?}"
            )));
        }

        // Validate macaroon_file exists and is not empty
        if !macaroon_file.exists()
            || macaroon_file
                .metadata()
                .map(|m| m.len() == 0)
                .unwrap_or(true)
        {
            return Err(Error::InvalidConfig(format!(
                "LND macaroon file not found or empty: {macaroon_file:?}"
            )));
        }

        let lnd_client = client::connect(&address, &cert_file, &macaroon_file)
            .await
            .map_err(|err| {
                tracing::error!("Connection error: {}", err.to_string());
                Error::Connection
            })?;

        let unit = CurrencyUnit::Msat;
        Ok(Self {
            _address: address,
            _cert_file: cert_file,
            _macaroon_file: macaroon_file,
            lnd_client,
            fee_reserve,
            kv_store,
            wait_invoice_cancel_token: CancellationToken::new(),
            wait_invoice_is_active: Arc::new(AtomicBool::new(false)),
            settings: SettingsResponse {
                unit: unit.to_string(),
                bolt11: Some(payment::Bolt11Settings {
                    mpp: true,
                    amountless: true,
                    invoice_description: true,
                }),
                bolt12: None,
                onchain: None,
                custom: std::collections::HashMap::new(),
            },
            unit,
        })
    }

    /// Get last add and settle indices from KV store
    #[instrument(skip_all)]
    async fn get_last_indices(&self) -> Result<(Option<u64>, Option<u64>), Error> {
        let add_index = if let Some(stored_index) = self
            .kv_store
            .kv_read(
                LND_KV_PRIMARY_NAMESPACE,
                LND_KV_SECONDARY_NAMESPACE,
                LAST_ADD_INDEX_KV_KEY,
            )
            .await
            .map_err(|e| Error::Database(e.to_string()))?
        {
            if let Ok(index_str) = std::str::from_utf8(stored_index.as_slice()) {
                index_str.parse::<u64>().ok()
            } else {
                None
            }
        } else {
            None
        };

        let settle_index = if let Some(stored_index) = self
            .kv_store
            .kv_read(
                LND_KV_PRIMARY_NAMESPACE,
                LND_KV_SECONDARY_NAMESPACE,
                LAST_SETTLE_INDEX_KV_KEY,
            )
            .await
            .map_err(|e| Error::Database(e.to_string()))?
        {
            if let Ok(index_str) = std::str::from_utf8(stored_index.as_slice()) {
                index_str.parse::<u64>().ok()
            } else {
                None
            }
        } else {
            None
        };

        tracing::debug!(
            "LND: Retrieved last indices from KV store - add_index: {:?}, settle_index: {:?}",
            add_index,
            settle_index
        );
        Ok((add_index, settle_index))
    }
}

fn lnrpc_payment_total_spent(payment: &lnrpc::Payment) -> Result<Amount<CurrencyUnit>, Error> {
    let total_msat = payment
        .value_msat
        .checked_add(payment.fee_msat)
        .ok_or(Error::AmountOverflow)?;
    let total_msat = u64::try_from(total_msat).map_err(|_| Error::AmountOverflow)?;

    Ok(Amount::new(total_msat, CurrencyUnit::Msat))
}

fn msat_total_spent_for_unit(
    total_msat: u64,
    unit: &CurrencyUnit,
) -> Result<Amount<CurrencyUnit>, Error> {
    match unit {
        CurrencyUnit::Msat => Ok(Amount::new(total_msat, CurrencyUnit::Msat)),
        CurrencyUnit::Sat => Ok(Amount::new(
            total_msat.div_ceil(MSAT_IN_SAT),
            CurrencyUnit::Sat,
        )),
        _ => Amount::new(total_msat, CurrencyUnit::Msat)
            .convert_to(unit)
            .map_err(Error::from),
    }
}

/// Build an authoritative terminal-failure response for a payment that was
/// rejected before dispatch.
///
/// The mint treats an `Ok` response with `MeltQuoteState::Failed` as
/// authoritative (it may compensate the melt), unlike an `Err`, whose dispatch
/// phase is unknown and which is therefore kept indeterminate. Pre-dispatch
/// rejections must be returned as this response so the melt can be rolled back
/// instead of parked pending.
///
/// Conversely, errors that straddle the dispatch boundary — a gRPC `Status`
/// error from `send_*` (`Error::LndError`), or a stream that drops after
/// dispatch began (`Error::AmbiguousDispatch`) — must stay `Err` so the melt
/// stays indeterminate. Do not convert those to this response.
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()),
    }
}

/// Preserve an existing payment, or reject an expired invoice before dispatch.
fn bolt11_pre_dispatch_response(
    unit: &CurrencyUnit,
    bolt11: &Bolt11Invoice,
    pay_state: MakePaymentResponse,
) -> Result<Option<MakePaymentResponse>, payment::Error> {
    let payment_lookup_id = PaymentIdentifier::PaymentHash(*bolt11.payment_hash().as_ref());
    Ok(match pay_state.status {
        MeltQuoteState::Paid | MeltQuoteState::Pending => Some(MakePaymentResponse {
            payment_lookup_id,
            total_spent: match (pay_state.total_spent.unit(), unit) {
                (CurrencyUnit::Msat, CurrencyUnit::Sat) => Amount::new(
                    pay_state.total_spent.value().div_ceil(MSAT_IN_SAT),
                    CurrencyUnit::Sat,
                ),
                _ => pay_state.total_spent.convert_to(unit)?,
            },
            ..pay_state
        }),
        MeltQuoteState::Unpaid | MeltQuoteState::Unknown | MeltQuoteState::Failed => {
            // LND rejects expired invoices before recording a payment, so a
            // later lookup cannot resolve that rejection. Return an authoritative
            // failure locally while we know no dispatch has been attempted.
            bolt11
                .is_expired()
                .then(|| outgoing_payment_failure_response(unit, payment_lookup_id))
        }
    })
}

#[async_trait]
impl MintPayment for Lnd {
    type Err = payment::Error;

    #[instrument(skip_all)]
    async fn get_settings(&self) -> Result<SettingsResponse, Self::Err> {
        Ok(self.settings.clone())
    }

    #[instrument(skip_all)]
    fn is_payment_event_stream_active(&self) -> bool {
        self.wait_invoice_is_active.load(Ordering::SeqCst)
    }

    #[instrument(skip_all)]
    fn cancel_payment_event_stream(&self) {
        self.wait_invoice_cancel_token.cancel()
    }

    #[instrument(skip_all)]
    async fn wait_payment_event(
        &self,
    ) -> Result<Pin<Box<dyn Stream<Item = Event> + Send>>, Self::Err> {
        let mut lnd_client = self.lnd_client.clone();

        // Get last indices from KV store
        let (last_add_index, last_settle_index) =
            self.get_last_indices().await.unwrap_or((None, None));

        let stream_req = lnrpc::InvoiceSubscription {
            add_index: last_add_index.unwrap_or(0),
            settle_index: last_settle_index.unwrap_or(0),
        };

        tracing::debug!(
            "LND: Starting invoice subscription with add_index: {}, settle_index: {}",
            stream_req.add_index,
            stream_req.settle_index
        );

        let stream = lnd_client
            .lightning()
            .subscribe_invoices(stream_req)
            .await
            .map_err(|_err| {
                tracing::error!("Could not subscribe to invoice");
                Error::Connection
            })?
            .into_inner();

        let cancel_token = self.wait_invoice_cancel_token.clone();
        let kv_store = self.kv_store.clone();

        let event_stream = futures::stream::unfold(
            (
                stream,
                cancel_token,
                Arc::clone(&self.wait_invoice_is_active),
                kv_store,
                last_add_index.unwrap_or(0),
                last_settle_index.unwrap_or(0),
            ),
            |(
                mut stream,
                cancel_token,
                is_active,
                kv_store,
                mut current_add_index,
                mut current_settle_index,
            )| async move {
                is_active.store(true, Ordering::SeqCst);

                loop {
                    tokio::select! {
                        _ = cancel_token.cancelled() => {
                            // Stream is cancelled
                            is_active.store(false, Ordering::SeqCst);
                            tracing::info!("Waiting for lnd invoice ending");
                            return None;
                        }
                        msg = stream.message() => {
                            match msg {
                                Ok(Some(msg)) => {
                                    // Update indices based on the message
                                    current_add_index = current_add_index.max(msg.add_index);
                                    current_settle_index = current_settle_index.max(msg.settle_index);

                                    // Store the updated indices in KV store regardless of settlement status
                                    let add_index_str = current_add_index.to_string();
                                    let settle_index_str = current_settle_index.to_string();

                                    if let Ok(mut tx) = kv_store.begin_transaction().await {
                                        let mut has_error = false;

                                        if let Err(e) = tx.kv_write(LND_KV_PRIMARY_NAMESPACE, LND_KV_SECONDARY_NAMESPACE, LAST_ADD_INDEX_KV_KEY, add_index_str.as_bytes()).await {
                                            tracing::warn!("LND: Failed to write add_index {} to KV store: {}", current_add_index, e);
                                            has_error = true;
                                        }

                                        if let Err(e) = tx.kv_write(LND_KV_PRIMARY_NAMESPACE, LND_KV_SECONDARY_NAMESPACE, LAST_SETTLE_INDEX_KV_KEY, settle_index_str.as_bytes()).await {
                                            tracing::warn!("LND: Failed to write settle_index {} to KV store: {}", current_settle_index, e);
                                            has_error = true;
                                        }

                                        if !has_error {
                                            if let Err(e) = tx.commit().await {
                                                tracing::warn!("LND: Failed to commit indices to KV store: {}", e);
                                            } else {
                                                tracing::debug!("LND: Stored updated indices - add_index: {}, settle_index: {}", current_add_index, current_settle_index);
                                            }
                                        }
                                    } else {
                                        tracing::warn!("LND: Failed to begin KV transaction for storing indices");
                                    }

                                    // Only emit event for settled invoices
                                    if msg.state() == InvoiceState::Settled {
                                        let hash_slice: Result<[u8;32], _> = msg.r_hash.try_into();

                                        if let Ok(hash_slice) = hash_slice {
                                            let hash = hex::encode(hash_slice);

                                            tracing::info!("LND: Payment for {} with amount {} msat", hash,  msg.amt_paid_msat);

                                            let wait_response = WaitPaymentResponse {
                                                payment_identifier: PaymentIdentifier::PaymentHash(hash_slice),
                                                payment_amount: Amount::new(msg.amt_paid_msat as u64, CurrencyUnit::Msat),
                                                payment_id: hash,
                                            };
                                            let event = Event::PaymentReceived(wait_response);
                                            return Some((event, (stream, cancel_token, is_active, kv_store, current_add_index, current_settle_index)));
                                        } else {
                                            // Invalid hash, skip this message but continue streaming
                                            tracing::error!("LND returned invalid payment hash");
                                            // Continue the loop without yielding
                                            continue;
                                        }
                                    } else {
                                        // Not a settled invoice, continue but don't emit event
                                        tracing::debug!("LND: Received non-settled invoice, continuing to wait for settled invoices");
                                        // Continue the loop without yielding
                                        continue;
                                    }
                                }
                                Ok(None) => {
                                    is_active.store(false, Ordering::SeqCst);
                                    tracing::info!("LND invoice stream ended.");
                                    return None;
                                }
                                Err(err) => {
                                    is_active.store(false, Ordering::SeqCst);
                                    tracing::warn!("Encountered error in LND invoice stream. Stream ending");
                                    tracing::error!("{:?}", err);
                                    return None;
                                }
                            }
                        }
                    }
                }
            },
        );

        Ok(Box::pin(event_stream))
    }

    #[instrument(skip_all)]
    async fn get_payment_quote(
        &self,
        unit: &CurrencyUnit,
        options: OutgoingPaymentOptions,
    ) -> Result<PaymentQuoteResponse, Self::Err> {
        match options {
            OutgoingPaymentOptions::Bolt11(bolt11_options) => {
                let amount_msat = match bolt11_options.melt_options {
                    Some(MeltOptions::Amountless { amountless }) => {
                        let amount_msat = amountless.amount_msat;

                        if let Some(invoice_amount) = bolt11_options.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_options
                        .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 = max(relative_fee_reserve, absolute_fee_reserve);

                Ok(PaymentQuoteResponse {
                    request_lookup_id: Some(PaymentIdentifier::PaymentHash(
                        *bolt11_options.bolt11.payment_hash().as_ref(),
                    )),
                    amount,
                    fee: Amount::new(fee, unit.clone()),
                    state: MeltQuoteState::Unpaid,
                    extra_json: None,
                    estimated_blocks: None,
                    fee_options: None,
                })
            }
            OutgoingPaymentOptions::Bolt12(_) => {
                Err(Self::Err::Anyhow(anyhow!("BOLT12 not supported by LND")))
            }
            OutgoingPaymentOptions::Custom(_) | OutgoingPaymentOptions::Onchain(_) => {
                Err(payment::Error::UnsupportedPaymentOption)
            }
        }
    }

    #[instrument(skip_all)]
    async fn make_payment(
        &self,
        unit: &CurrencyUnit,
        options: OutgoingPaymentOptions,
    ) -> Result<MakePaymentResponse, Self::Err> {
        match options {
            OutgoingPaymentOptions::Bolt11(bolt11_options) => {
                let bolt11 = bolt11_options.bolt11;
                let payment_lookup_id =
                    PaymentIdentifier::PaymentHash(*bolt11.payment_hash().as_ref());

                // A prior lookup is authoritative evidence, not an error:
                // report the already-recorded outcome so the mint reconciles
                // against durable state instead of treating the duplicate melt
                // as an ambiguous dispatch failure.
                let pay_state = self.check_outgoing_payment(&payment_lookup_id).await?;

                if let Some(response) = bolt11_pre_dispatch_response(unit, &bolt11, pay_state)? {
                    return Ok(response);
                }

                // Detect partial payments
                match bolt11_options.melt_options {
                    Some(MeltOptions::Mpp { mpp }) => {
                        let amount_msat: u64 = match bolt11.amount_milli_satoshis() {
                            Some(amount_msat) => amount_msat,
                            None => {
                                // Invoice carries no amount; a local parse
                                // failure before any dispatch.
                                tracing::warn!(
                                    payment_lookup_id = %payment_lookup_id,
                                    "LND MPP payment rejected before dispatch: invoice has no amount",
                                );
                                return Ok(outgoing_payment_failure_response(
                                    unit,
                                    payment_lookup_id,
                                ));
                            }
                        };
                        {
                            let partial_amount_msat = mpp.amount;
                            let invoice = bolt11;
                            let max_fee: Option<Amount<CurrencyUnit>> =
                                bolt11_options.max_fee_amount.clone();

                            // Extract information from invoice
                            let pub_key = invoice.get_payee_pub_key();
                            let payer_addr = invoice.payment_secret().0.to_vec();
                            let payment_hash = invoice.payment_hash();

                            let mut lnd_client = self.lnd_client.clone();

                            for attempt in 0..Self::MAX_ROUTE_RETRIES {
                                // Create a request for the routes
                                let route_req = lnrpc::QueryRoutesRequest {
                                    pub_key: hex::encode(pub_key.serialize()),
                                    amt_msat: u64::from(partial_amount_msat) as i64,
                                    fee_limit: max_fee
                                        .clone()
                                        .map(|f| {
                                            let fee_msat = f.to_msat()?;
                                            let limit = Limit::FixedMsat(fee_msat as i64);
                                            Ok::<_, Error>(FeeLimit { limit: Some(limit) })
                                        })
                                        .transpose()?,
                                    use_mission_control: true,
                                    ..Default::default()
                                };

                                // Query the routes
                                let mut routes_response = lnd_client
                                    .lightning()
                                    .query_routes(route_req)
                                    .await
                                    .inspect_err(|err| {
                                        tracing::warn!(
                                            payment_lookup_id = %payment_lookup_id,
                                            attempt = attempt + 1,
                                            rpc_code = %err.code(),
                                            error = %err.message(),
                                            "LND MPP route query failed",
                                        );
                                    })
                                    .map_err(Error::LndError)?
                                    .into_inner();

                                // Get first route and update its MPP record. An
                                // empty route set means LND found no path; the
                                // payment was never dispatched.
                                let route = match routes_response.routes.first_mut() {
                                    Some(route) => route,
                                    None => {
                                        tracing::warn!(
                                            payment_lookup_id = %payment_lookup_id,
                                            attempt = attempt + 1,
                                            "LND MPP route query returned no routes",
                                        );
                                        return Ok(outgoing_payment_failure_response(
                                            unit,
                                            payment_lookup_id,
                                        ));
                                    }
                                };

                                // attempt it and check the result
                                let last_hop: &mut Hop = match route.hops.last_mut() {
                                    Some(last_hop) => last_hop,
                                    None => {
                                        tracing::warn!(
                                            payment_lookup_id = %payment_lookup_id,
                                            attempt = attempt + 1,
                                            "LND MPP route has no hops",
                                        );
                                        return Ok(outgoing_payment_failure_response(
                                            unit,
                                            payment_lookup_id,
                                        ));
                                    }
                                };
                                let mpp_record = MppRecord {
                                    payment_addr: payer_addr.clone(),
                                    total_amt_msat: amount_msat as i64,
                                };
                                last_hop.mpp_record = Some(mpp_record);

                                let payment_response = lnd_client
                                    .router()
                                    .send_to_route_v2(routerrpc::SendToRouteRequest {
                                        payment_hash: payment_hash.to_byte_array().to_vec(),
                                        route: Some(route.clone()),
                                        ..Default::default()
                                    })
                                    .await
                                    .inspect_err(|err| {
                                        tracing::warn!(
                                            payment_lookup_id = %payment_lookup_id,
                                            attempt = attempt + 1,
                                            rpc_code = %err.code(),
                                            error = %err.message(),
                                            "LND MPP dispatch RPC failed; payment outcome requires verification",
                                        );
                                    })
                                    .map_err(Error::LndError)?
                                    .into_inner();

                                if let Some(failure) = payment_response.failure {
                                    if failure.code == 15 {
                                        tracing::debug!(
                                            payment_lookup_id = %payment_lookup_id,
                                            attempt = attempt + 1,
                                            failure_code = failure.code,
                                            failure_reason = failure.code().as_str_name(),
                                            failure_source_index = failure.failure_source_index,
                                            "LND MPP route failed; querying another route",
                                        );
                                        continue;
                                    }
                                    tracing::warn!(
                                        payment_lookup_id = %payment_lookup_id,
                                        attempt = attempt + 1,
                                        failure_code = failure.code,
                                        failure_reason = failure.code().as_str_name(),
                                        failure_source_index = failure.failure_source_index,
                                        "LND MPP attempt returned a failure",
                                    );
                                }

                                // Get status and maybe the preimage
                                let (status, payment_preimage) = match payment_response.status {
                                    0 => (MeltQuoteState::Pending, None),
                                    1 => (
                                        MeltQuoteState::Paid,
                                        Some(hex::encode(payment_response.preimage)),
                                    ),
                                    2 => (MeltQuoteState::Unpaid, None),
                                    _ => (MeltQuoteState::Unknown, None),
                                };

                                // Get the actual amount paid in msats
                                let total_amt_msat: u64 = payment_response
                                    .route
                                    .map_or(0, |route| route.total_amt_msat as u64);

                                return Ok(MakePaymentResponse {
                                    payment_lookup_id: PaymentIdentifier::PaymentHash(
                                        payment_hash.to_byte_array(),
                                    ),
                                    payment_proof: payment_preimage,
                                    status,
                                    total_spent: msat_total_spent_for_unit(total_amt_msat, unit)?,
                                });
                            }

                            // "We have exhausted all tactical options" -- STEM, Upgrade (2018)
                            // All route attempts returned retryable failures.
                            tracing::warn!(
                                payment_lookup_id = %payment_lookup_id,
                                attempts = Self::MAX_ROUTE_RETRIES,
                                "LND MPP payment exhausted route retries",
                            );
                            Ok(outgoing_payment_failure_response(unit, payment_lookup_id))
                        }
                    }
                    _ => {
                        let mut lnd_client = self.lnd_client.clone();

                        let max_fee: Option<Amount<CurrencyUnit>> = bolt11_options.max_fee_amount;

                        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) {
                                        // Invoice/request amount disagreement is
                                        // a local validation failure, before any
                                        // dispatch to LND.
                                        tracing::warn!(
                                            payment_lookup_id = %payment_lookup_id,
                                            invoice_amount_msat = invoice_amount,
                                            requested_amount_msat = u64::from(amount_msat),
                                            "LND payment rejected before dispatch: invoice and requested amounts differ",
                                        );
                                        return Ok(outgoing_payment_failure_response(
                                            unit,
                                            payment_lookup_id,
                                        ));
                                    }
                                }

                                u64::from(amount_msat)
                            }
                            Some(MeltOptions::Mpp { mpp }) => u64::from(mpp.amount),
                            None => 0,
                        };

                        let fee_limit_msat = match max_fee {
                            Some(fee) => fee.convert_to(&CurrencyUnit::Msat)?.value() as i64,
                            None => 0,
                        };

                        let pay_req = routerrpc::SendPaymentRequest {
                            payment_request: bolt11.to_string(),
                            fee_limit_msat,
                            amt_msat: amount_msat as i64,
                            ..Default::default()
                        };

                        let mut payment_stream = lnd_client
                            .router()
                            .send_payment_v2(pay_req)
                            .await
                            .map_err(|err| {
                                tracing::warn!(
                                    payment_lookup_id = %payment_lookup_id,
                                    rpc_code = %err.code(),
                                    error = %err.message(),
                                    "LND payment dispatch RPC failed; payment outcome requires verification",
                                );
                                // A gRPC error here may arrive after LND accepted
                                // the payment; the dispatch outcome is unknown.
                                Error::AmbiguousDispatch
                            })?
                            .into_inner();

                        while let Some(update) = payment_stream.message().await.map_err(|err| {
                            tracing::warn!(
                                payment_lookup_id = %payment_lookup_id,
                                rpc_code = %err.code(),
                                error = %err.message(),
                                "LND payment stream failed after dispatch; payment may still settle",
                            );
                            // The stream dropped after dispatch began; the payment
                            // may still settle.
                            Error::AmbiguousDispatch
                        })? {
                            let status = update.status();

                            let response_status = match status {
                                PaymentStatus::InFlight | PaymentStatus::Initiated => {
                                    continue;
                                }
                                PaymentStatus::Succeeded => MeltQuoteState::Paid,
                                PaymentStatus::Failed => {
                                    tracing::warn!(
                                        payment_lookup_id = %payment_lookup_id,
                                        failure_code = update.failure_reason,
                                        failure_reason = update.failure_reason().as_str_name(),
                                        "LND outgoing payment failed",
                                    );
                                    MeltQuoteState::Failed
                                }
                                #[allow(deprecated)]
                                PaymentStatus::Unknown => MeltQuoteState::Unknown,
                            };

                            let total_msat = update
                                .value_msat
                                .checked_add(update.fee_msat)
                                .ok_or(Error::AmountOverflow)?;

                            let payment_preimage = if update.payment_preimage.is_empty() {
                                None
                            } else {
                                Some(update.payment_preimage)
                            };

                            let payment_identifier =
                                PaymentIdentifier::PaymentHash(*bolt11.payment_hash().as_ref());

                            return Ok(MakePaymentResponse {
                                payment_lookup_id: payment_identifier,
                                payment_proof: payment_preimage,
                                status: response_status,
                                total_spent: msat_total_spent_for_unit(total_msat as u64, unit)?,
                            });
                        }

                        tracing::warn!(
                            payment_lookup_id = %payment_lookup_id,
                            "LND payment stream ended without a terminal result; payment outcome remains unknown",
                        );
                        Err(Error::UnknownPaymentStatus.into())
                    }
                }
            }
            OutgoingPaymentOptions::Bolt12(_) => {
                Err(Self::Err::Anyhow(anyhow!("BOLT12 not supported by LND")))
            }
            OutgoingPaymentOptions::Custom(_) | OutgoingPaymentOptions::Onchain(_) => {
                Err(payment::Error::UnsupportedPaymentOption)
            }
        }
    }

    #[instrument(skip(self, options))]
    async fn create_incoming_payment_request(
        &self,
        options: IncomingPaymentOptions,
    ) -> Result<CreateIncomingPaymentResponse, Self::Err> {
        match options {
            IncomingPaymentOptions::Bolt11(bolt11_options) => {
                let description = bolt11_options.description.unwrap_or_default();
                let amount = bolt11_options.amount;
                let unix_expiry = bolt11_options.unix_expiry;

                let amount_msat: Amount = amount.convert_to(&CurrencyUnit::Msat)?.into();

                let invoice_request = lnrpc::Invoice {
                    value_msat: u64::from(amount_msat) as i64,
                    memo: description,
                    expiry: unix_expiry
                        .map(|t| {
                            t.checked_sub(unix_time())
                                .ok_or(payment::Error::InvalidExpiry)
                        })
                        .transpose()?
                        .unwrap_or_default() as i64,
                    ..Default::default()
                };

                let mut lnd_client = self.lnd_client.clone();

                let invoice = lnd_client
                    .lightning()
                    .add_invoice(tonic::Request::new(invoice_request))
                    .await
                    .map_err(|e| payment::Error::Anyhow(anyhow!(e)))?
                    .into_inner();

                let bolt11 = Bolt11Invoice::from_str(&invoice.payment_request)?;

                let payment_identifier =
                    PaymentIdentifier::PaymentHash(*bolt11.payment_hash().as_ref());

                let expiry = bolt11.expires_at().map(|t| t.as_secs());

                Ok(CreateIncomingPaymentResponse {
                    request_lookup_id: payment_identifier,
                    request: bolt11.to_string(),
                    expiry,
                    extra_json: None,
                })
            }
            IncomingPaymentOptions::Bolt12(_) => {
                Err(Self::Err::Anyhow(anyhow!("BOLT12 not supported by LND")))
            }
            IncomingPaymentOptions::Custom(_) | IncomingPaymentOptions::Onchain(_) => {
                Err(payment::Error::UnsupportedPaymentOption)
            }
        }
    }

    #[instrument(skip(self))]
    async fn check_incoming_payment_status(
        &self,
        payment_identifier: &PaymentIdentifier,
    ) -> Result<Vec<WaitPaymentResponse>, Self::Err> {
        let mut lnd_client = self.lnd_client.clone();

        let invoice_request = lnrpc::PaymentHash {
            r_hash: hex::decode(payment_identifier.to_string())?,
            ..Default::default()
        };

        let invoice = lnd_client
            .lightning()
            .lookup_invoice(tonic::Request::new(invoice_request))
            .await
            .map_err(|e| payment::Error::Anyhow(anyhow!(e)))?
            .into_inner();

        if invoice.state() == InvoiceState::Settled {
            Ok(vec![WaitPaymentResponse {
                payment_identifier: payment_identifier.clone(),
                payment_amount: Amount::new(invoice.amt_paid_msat as u64, CurrencyUnit::Msat),
                payment_id: hex::encode(invoice.r_hash),
            }])
        } else {
            Ok(vec![])
        }
    }

    #[instrument(skip(self))]
    async fn check_outgoing_payment(
        &self,
        payment_identifier: &PaymentIdentifier,
    ) -> Result<MakePaymentResponse, Self::Err> {
        let mut lnd_client = self.lnd_client.clone();

        let payment_hash = &payment_identifier.to_string();

        let track_request = routerrpc::TrackPaymentRequest {
            payment_hash: hex::decode(payment_hash).map_err(|_| Error::InvalidHash)?,
            no_inflight_updates: true,
        };

        let payment_response = lnd_client.router().track_payment_v2(track_request).await;

        let mut payment_stream = match payment_response {
            Ok(stream) => stream.into_inner(),
            Err(err) => {
                let err_code = err.code();
                if err_code == tonic::Code::NotFound {
                    tracing::debug!(
                        payment_lookup_id = %payment_identifier,
                        "LND does not know this outgoing payment; reporting Unknown because absence is not authoritative proof of permanent failure",
                    );
                    return Ok(MakePaymentResponse {
                        payment_lookup_id: payment_identifier.clone(),
                        payment_proof: None,
                        status: MeltQuoteState::Unknown,
                        total_spent: Amount::new(0, self.unit.clone()),
                    });
                } else {
                    tracing::warn!(
                        payment_lookup_id = %payment_identifier,
                        rpc_code = %err_code,
                        error = %err.message(),
                        "LND outgoing payment status RPC failed; payment outcome remains unknown",
                    );
                    return Err(payment::Error::UnknownPaymentState);
                }
            }
        };

        while let Some(update_result) = payment_stream.next().await {
            match update_result {
                Ok(update) => {
                    let status = update.status();

                    let response = match status {
                        #[allow(deprecated)]
                        PaymentStatus::Unknown => MakePaymentResponse {
                            payment_lookup_id: payment_identifier.clone(),
                            payment_proof: Some(update.payment_preimage),
                            status: MeltQuoteState::Unknown,
                            total_spent: Amount::new(0, self.unit.clone()),
                        },
                        PaymentStatus::InFlight | PaymentStatus::Initiated => {
                            // Continue waiting for the next update
                            continue;
                        }
                        PaymentStatus::Succeeded => {
                            let total_spent = lnrpc_payment_total_spent(&update)?;

                            MakePaymentResponse {
                                payment_lookup_id: payment_identifier.clone(),
                                payment_proof: Some(update.payment_preimage),
                                status: MeltQuoteState::Paid,
                                total_spent,
                            }
                        }
                        PaymentStatus::Failed => {
                            // Status checks also run before dispatch and may
                            // repeatedly observe the same recorded failure.
                            tracing::debug!(
                                payment_lookup_id = %payment_identifier,
                                failure_code = update.failure_reason,
                                failure_reason = update.failure_reason().as_str_name(),
                                "LND outgoing payment status is failed",
                            );
                            MakePaymentResponse {
                                payment_lookup_id: payment_identifier.clone(),
                                payment_proof: Some(update.payment_preimage),
                                status: MeltQuoteState::Failed,
                                total_spent: Amount::new(0, self.unit.clone()),
                            }
                        }
                    };

                    return Ok(response);
                }
                Err(err) => {
                    // Handle the case where the update itself is an error (e.g., stream failure)
                    tracing::warn!(
                        payment_lookup_id = %payment_identifier,
                        rpc_code = %err.code(),
                        error = %err.message(),
                        "LND outgoing payment status stream failed; payment outcome remains unknown",
                    );
                    return Err(Error::UnknownPaymentStatus.into());
                }
            }
        }

        // If the stream is exhausted without a final status
        tracing::warn!(
            payment_lookup_id = %payment_identifier,
            "LND outgoing payment status stream ended without a terminal result; payment outcome remains unknown",
        );
        Err(Error::UnknownPaymentStatus.into())
    }
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use cdk_common::bitcoin::hashes::sha256;
    use cdk_common::bitcoin::secp256k1::{Secp256k1, SecretKey};
    use cdk_common::lightning_invoice::{Currency, InvoiceBuilder, PaymentSecret};

    use super::*;

    fn invoice_with_timestamp(timestamp: Duration) -> Bolt11Invoice {
        let key = SecretKey::from_slice(&[1; 32]).unwrap();
        InvoiceBuilder::new(Currency::Regtest)
            .description("expiry test".to_owned())
            .payment_hash(sha256::Hash::from_byte_array([42; 32]))
            .payment_secret(PaymentSecret([43; 32]))
            .duration_since_epoch(timestamp)
            .expiry_time(Duration::from_secs(3600))
            .min_final_cltv_expiry_delta(144)
            .build_signed(|hash| Secp256k1::new().sign_ecdsa_recoverable(hash, &key))
            .unwrap()
    }

    #[test]
    fn expired_invoice_without_active_payment_fails_before_dispatch() {
        let invoice = invoice_with_timestamp(Duration::from_secs(1));
        let payment_lookup_id = PaymentIdentifier::PaymentHash(*invoice.payment_hash().as_ref());

        for status in [
            MeltQuoteState::Unknown,
            MeltQuoteState::Unpaid,
            MeltQuoteState::Failed,
        ] {
            let pay_state = MakePaymentResponse {
                status,
                ..outgoing_payment_failure_response(&CurrencyUnit::Msat, payment_lookup_id.clone())
            };
            let response = bolt11_pre_dispatch_response(&CurrencyUnit::Sat, &invoice, pay_state)
                .unwrap()
                .unwrap();

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

    #[test]
    fn existing_payment_response_uses_requested_unit() {
        for timestamp in [Duration::from_secs(1), Duration::from_secs(unix_time())] {
            let invoice = invoice_with_timestamp(timestamp);
            let payment_lookup_id =
                PaymentIdentifier::PaymentHash(*invoice.payment_hash().as_ref());

            for status in [MeltQuoteState::Paid, MeltQuoteState::Pending] {
                for (total_msat, total_sat) in [(0, 0), (1_234, 2), (2_000, 2)] {
                    for (unit, expected) in [
                        (CurrencyUnit::Sat, total_sat),
                        (CurrencyUnit::Msat, total_msat),
                    ] {
                        let pay_state = MakePaymentResponse {
                            payment_lookup_id: payment_lookup_id.clone(),
                            payment_proof: Some("existing preimage".to_owned()),
                            status,
                            total_spent: Amount::new(total_msat, CurrencyUnit::Msat),
                        };
                        let response = bolt11_pre_dispatch_response(&unit, &invoice, pay_state)
                            .unwrap()
                            .unwrap();

                        assert_eq!(response.status, status);
                        assert_eq!(response.payment_lookup_id, payment_lookup_id);
                        assert_eq!(response.total_spent, Amount::new(expected, unit));
                        assert_eq!(response.payment_proof.as_deref(), Some("existing preimage"));
                    }
                }
            }
        }
    }

    #[test]
    fn unexpired_invoice_without_active_payment_can_dispatch() {
        let invoice = invoice_with_timestamp(Duration::from_secs(unix_time()));
        let payment_lookup_id = PaymentIdentifier::PaymentHash(*invoice.payment_hash().as_ref());

        for status in [
            MeltQuoteState::Unknown,
            MeltQuoteState::Unpaid,
            MeltQuoteState::Failed,
        ] {
            let pay_state = MakePaymentResponse {
                status,
                ..outgoing_payment_failure_response(&CurrencyUnit::Msat, payment_lookup_id.clone())
            };
            assert!(
                bolt11_pre_dispatch_response(&CurrencyUnit::Sat, &invoice, pay_state)
                    .unwrap()
                    .is_none()
            );
        }
    }

    #[test]
    fn lnrpc_payment_total_spent_uses_msat_fields() {
        let payment = lnrpc::Payment {
            value_msat: 1500,
            fee_msat: 500,
            value_sat: 1,
            fee_sat: 0,
            ..Default::default()
        };

        let total_spent = lnrpc_payment_total_spent(&payment)
            .expect("sub-sat payment total should be calculated");

        assert_eq!(
            total_spent
                .convert_to(&CurrencyUnit::Msat)
                .expect("msat amount should convert to msat")
                .value(),
            2000
        );
    }

    #[test]
    fn lnrpc_payment_total_spent_rejects_overflow() {
        let payment = lnrpc::Payment {
            value_msat: i64::MAX,
            fee_msat: 1,
            ..Default::default()
        };

        let err = lnrpc_payment_total_spent(&payment)
            .expect_err("overflowing payment total should be rejected");

        assert!(matches!(err, Error::AmountOverflow));
    }

    #[test]
    fn msat_total_spent_for_unit_rounds_up_sats() {
        let total_spent = msat_total_spent_for_unit(1501, &CurrencyUnit::Sat)
            .expect("msat total should convert to sat");

        assert_eq!(total_spent, Amount::new(2, CurrencyUnit::Sat));
    }

    #[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::Sat, 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::Sat));
        assert!(response.payment_proof.is_none());
    }

    /// The dispatch-boundary variants must remain distinct from the
    /// pre-dispatch `PaymentFailed`, so only the former stay `Err` (ambiguous)
    /// and the latter can be converted to an authoritative `Failed` response.
    /// This guards against a future change re-collapsing the two.
    #[test]
    fn dispatch_boundary_errors_are_distinct_from_pre_dispatch_failure() {
        // `AmbiguousDispatch` is returned by send_* / stream failures (may have
        // been accepted by LND) and must never be treated as a terminal
        // pre-dispatch failure. It is a separate variant from `PaymentFailed`.
        assert_ne!(
            Error::AmbiguousDispatch.to_string(),
            Error::PaymentFailed.to_string()
        );
        assert_ne!(
            Error::UnknownPaymentStatus.to_string(),
            Error::PaymentFailed.to_string()
        );
    }
}