fedimint_lightning/

lnd.rs

use std::collections::BTreeMap;
use std::fmt::{self, Display};
use std::str::FromStr;
use std::sync::Arc;
use std::time::{Duration, UNIX_EPOCH};

use async_trait::async_trait;
use bitcoin::OutPoint;
use bitcoin::hashes::{Hash, sha256};
use fedimint_core::encoding::Encodable;
use fedimint_core::task::{TaskGroup, sleep};
use fedimint_core::util::FmtCompact;
use fedimint_core::{Amount, BitcoinAmountOrAll, crit, secp256k1};
use fedimint_gateway_common::{
    ConnectPeerRequest, ListTransactionsResponse, PaymentDetails, PaymentDirection, PaymentKind,
};
use fedimint_ln_common::PrunedInvoice;
use fedimint_ln_common::contracts::Preimage;
use fedimint_ln_common::route_hints::{RouteHint, RouteHintHop};
use fedimint_logging::LOG_LIGHTNING;
use hex::ToHex;
use secp256k1::PublicKey;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tonic_lnd::invoicesrpc::lookup_invoice_msg::InvoiceRef;
use tonic_lnd::invoicesrpc::{
    AddHoldInvoiceRequest, CancelInvoiceMsg, LookupInvoiceMsg, SettleInvoiceMsg,
    SubscribeSingleInvoiceRequest,
};
use tonic_lnd::lnrpc::channel_point::FundingTxid;
use tonic_lnd::lnrpc::failure::FailureCode;
use tonic_lnd::lnrpc::invoice::InvoiceState;
use tonic_lnd::lnrpc::payment::PaymentStatus;
use tonic_lnd::lnrpc::policy_update_request::Scope as PolicyUpdateScope;
use tonic_lnd::lnrpc::{
    ChanInfoRequest, ChannelBalanceRequest, ChannelPoint, CloseChannelRequest,
    ConnectPeerRequest as LndConnectPeerRequest, FeeReportRequest, GetInfoRequest, Invoice,
    InvoiceHtlc, InvoiceHtlcState, InvoiceSubscription, LightningAddress, ListChannelsRequest,
    ListInvoiceRequest, ListPaymentsRequest, ListPeersRequest, OpenChannelRequest,
    PolicyUpdateRequest, SendCoinsRequest, UpdateFailure, WalletBalanceRequest,
};
use tonic_lnd::routerrpc::{
    CircuitKey, ForwardHtlcInterceptResponse, ResolveHoldForwardAction, SendPaymentRequest,
    TrackPaymentRequest,
};
use tonic_lnd::tonic::Code;
use tonic_lnd::walletrpc::AddrRequest;
use tonic_lnd::{Client as LndClient, connect};
use tracing::{debug, info, trace, warn};

use super::{
    ChannelInfo, ILnRpcClient, LightningRpcError, ListChannelsResponse, Lnv2HoldInvoiceFilter,
    MAX_LIGHTNING_RETRIES, RouteHtlcStream,
};
use crate::{
    CloseChannelsWithPeerRequest, CloseChannelsWithPeerResponse, CreateInvoiceRequest,
    CreateInvoiceResponse, GetBalancesResponse, GetInvoiceRequest, GetInvoiceResponse,
    GetLnOnchainAddressResponse, GetNodeInfoResponse, GetRouteHintsResponse,
    InterceptPaymentRequest, InterceptPaymentResponse, InvoiceDescription, NO_INCOMING_CIRCUIT,
    OpenChannelResponse, PayInvoiceResponse, PaymentAction, SendOnchainRequest,
    SendOnchainResponse, SetChannelFeesRequest,
};

type HtlcSubscriptionSender = mpsc::Sender<InterceptPaymentRequest>;

/// Final CLTV delta of the HOLD invoices created for LNv2 receives. Left unset,
/// LND would use its routing `bitcoin.timelockdelta`, which operators may lower
/// to 24 (18 before v0.21), leaving only a few blocks between accepting a
/// payment and LND cancelling it by itself `invoices.holdexpirydelta` blocks
/// before expiry. LND rejects any HTLC expiring sooner than this many blocks
/// after it arrives, so it bounds every accepted HTLC, not just honest payers.
const LNV2_HOLD_INVOICE_CLTV_EXPIRY: u64 = 144;

/// How many blocks before its earliest HTLC expiry LND is assumed to cancel an
/// accepted HOLD invoice by itself. LND's `invoices.holdexpirydelta` defaults
/// to 18 since v0.21 (16 + 2) and 12 before, has no upper bound, and cannot be
/// read cheaply before v0.21 (`GetDebugInfo` returns the whole log), so this
/// assumes twice the current default. The excess also absorbs blocks found
/// between the deadline check and settlement. It costs nothing for a payment
/// handled promptly, which still has about 108 blocks left under
/// `LNV2_HOLD_INVOICE_CLTV_EXPIRY`.
const LND_ASSUMED_HOLD_EXPIRY_DELTA: u32 = 36;

/// Whether a HOLD invoice accepted an HTLC without an MPP total. For the
/// non-blinded HOLD invoices the gateway creates, standard LND builds require
/// the payment address, which travels in the MPP record, from every HTLC
/// except one carrying a keysend record. Since LND v0.20.3 and v0.21.2 that
/// record must hold the invoice's preimage; before, any value passes unless
/// `accept-keysend` is on.
fn has_accepted_keysend_htlc(htlcs: &[InvoiceHtlc]) -> bool {
    htlcs
        .iter()
        .any(|htlc| htlc.state() == InvoiceHtlcState::Accepted && htlc.mpp_total_amt_msat == 0)
}

/// Block height at which LND is assumed to cancel an accepted HOLD invoice by
/// itself, `LND_ASSUMED_HOLD_EXPIRY_DELTA` blocks before its earliest accepted
/// HTLC expires, after which the gateway can no longer settle it. Without an
/// accepted HTLC there is no deadline to trust, so it is `0`.
///
/// The same holds once a keysend HTLC is accepted (see
/// [`has_accepted_keysend_htlc`]): the invoice then keeps accepting further
/// keysend HTLCs, and until it restarts LND neither reports them nor moves its
/// cancel height for them.
fn hold_invoice_claim_deadline(htlcs: &[InvoiceHtlc]) -> u32 {
    if has_accepted_keysend_htlc(htlcs) {
        return 0;
    }

    htlcs
        .iter()
        .filter(|htlc| htlc.state() == InvoiceHtlcState::Accepted)
        .map(|htlc| (htlc.expiry_height as u32).saturating_sub(LND_ASSUMED_HOLD_EXPIRY_DELTA))
        .min()
        .unwrap_or_default()
}

/// The hex preimage of an LND invoice, if LND knows it. A HOLD invoice has no
/// preimage until it is settled, so `r_preimage` is empty while one is pending
/// or after it was canceled.
fn invoice_preimage_hex(invoice: &Invoice) -> Option<String> {
    <[u8; 32]>::try_from(invoice.r_preimage.as_slice())
        .ok()
        .map(|preimage| preimage.consensus_encode_to_hex())
}

#[derive(Debug, Clone, Copy, Eq, PartialEq)]
enum HoldInvoiceAction {
    Complete,
    AlreadyComplete,
}

#[derive(Debug, Clone, Copy, Eq, PartialEq)]
struct HoldInvoiceStateError {
    failure_reason: &'static str,
    permanent: bool,
}

fn hold_invoice_action(
    requested_action: PaymentActionKind,
    invoice_state: Option<InvoiceState>,
) -> Result<HoldInvoiceAction, HoldInvoiceStateError> {
    match (requested_action, invoice_state) {
        (PaymentActionKind::Settle, Some(InvoiceState::Accepted))
        | (PaymentActionKind::Cancel, Some(InvoiceState::Open | InvoiceState::Accepted)) => {
            Ok(HoldInvoiceAction::Complete)
        }
        (PaymentActionKind::Settle, Some(InvoiceState::Settled))
        | (PaymentActionKind::Cancel, Some(InvoiceState::Canceled)) => {
            Ok(HoldInvoiceAction::AlreadyComplete)
        }
        (PaymentActionKind::Settle, Some(InvoiceState::Canceled)) => Err(HoldInvoiceStateError {
            failure_reason: "HOLD invoice was canceled instead of settled",
            permanent: true,
        }),
        (PaymentActionKind::Cancel, Some(InvoiceState::Settled)) => Err(HoldInvoiceStateError {
            failure_reason: "HOLD invoice was settled instead of canceled",
            permanent: true,
        }),
        (PaymentActionKind::Settle, Some(InvoiceState::Open)) => Err(HoldInvoiceStateError {
            failure_reason: "HOLD invoice is open and has no accepted HTLC to settle",
            permanent: false,
        }),
        (_, None) => Err(HoldInvoiceStateError {
            failure_reason: "HOLD invoice does not exist",
            permanent: true,
        }),
    }
}

#[derive(Debug, Clone, Copy, Eq, PartialEq)]
enum PaymentActionKind {
    Settle,
    Cancel,
}

#[derive(Clone)]
pub struct GatewayLndClient {
    /// LND client
    address: String,
    tls_cert: String,
    macaroon: String,
    time_pref: f64,
    /// How long (in seconds) LND keeps trying to route an outgoing payment
    /// before giving up. Passed as `timeout_seconds` in `SendPaymentRequest`.
    payment_timeout_secs: i32,
    lnd_sender: Option<mpsc::Sender<ForwardHtlcInterceptResponse>>,
    /// Predicate used to distinguish HOLD invoices the gateway created
    /// (federation-bound) from unrelated HOLD invoices on the same LND node.
    /// Without this, every HOLD invoice on a shared LND would be intercepted
    /// as if it were federation-bound, producing invalid LNv1 responses that
    /// crash LND's htlc_interceptor stream.
    lnv2_filter: Lnv2HoldInvoiceFilter,
}

impl GatewayLndClient {
    pub fn new(
        address: String,
        tls_cert: String,
        macaroon: String,
        time_pref: f64,
        payment_timeout_secs: i32,
        lnd_sender: Option<mpsc::Sender<ForwardHtlcInterceptResponse>>,
        lnv2_filter: Lnv2HoldInvoiceFilter,
    ) -> Self {
        info!(
            target: LOG_LIGHTNING,
            address = %address,
            tls_cert_path = %tls_cert,
            macaroon = %macaroon,
            time_pref,
            payment_timeout_secs,
            "Gateway configured to connect to LND LnRpcClient",
        );
        GatewayLndClient {
            address,
            tls_cert,
            macaroon,
            time_pref,
            payment_timeout_secs,
            lnd_sender,
            lnv2_filter,
        }
    }

    async fn connect(&self) -> Result<LndClient, LightningRpcError> {
        let mut retries = 0;
        let client = loop {
            if retries >= MAX_LIGHTNING_RETRIES {
                return Err(LightningRpcError::FailedToConnect);
            }

            retries += 1;

            match connect(
                self.address.clone(),
                self.tls_cert.clone(),
                self.macaroon.clone(),
            )
            .await
            {
                Ok(client) => break client,
                Err(err) => {
                    debug!(target: LOG_LIGHTNING, err = %err.fmt_compact(), "Couldn't connect to LND, retrying in 1 second...");
                    sleep(Duration::from_secs(1)).await;
                }
            }
        };

        Ok(client)
    }

    async fn connect_peer_if_needed(
        &self,
        client: &mut LndClient,
        pubkey: PublicKey,
        host: String,
    ) -> Result<(), LightningRpcError> {
        let peers = client
            .lightning()
            .list_peers(ListPeersRequest { latest_error: true })
            .await
            .map_err(|e| LightningRpcError::FailedToConnectToPeer {
                failure_reason: format!("Could not list peers: {e:?}"),
            })?
            .into_inner();

        if peers.peers.into_iter().any(|peer| {
            PublicKey::from_str(&peer.pub_key).expect("LND returned invalid peer public key")
                == pubkey
        }) {
            return Ok(());
        }

        client
            .lightning()
            .connect_peer(LndConnectPeerRequest {
                addr: Some(LightningAddress {
                    pubkey: pubkey.to_string(),
                    host,
                }),
                perm: false,
                timeout: 10,
            })
            .await
            .map_err(|e| LightningRpcError::FailedToConnectToPeer {
                failure_reason: format!("Failed to connect to peer {e:?}"),
            })?;

        Ok(())
    }

    /// Spawns a new background task that subscribes to updates of a specific
    /// HOLD invoice. When the HOLD invoice is ACCEPTED, we can request the
    /// preimage from the Gateway. A new task is necessary because LND's
    /// global `subscribe_invoices` does not currently emit updates for HOLD invoices: <https://github.com/lightningnetwork/lnd/issues/3120>
    async fn spawn_lnv2_hold_invoice_subscription(
        &self,
        task_group: &TaskGroup,
        payment_stream_group: TaskGroup,
        gateway_sender: HtlcSubscriptionSender,
        payment_hash: Vec<u8>,
    ) -> Result<(), LightningRpcError> {
        let mut client = self.connect().await?;

        let self_copy = self.clone();
        let r_hash = payment_hash.clone();
        task_group.spawn("LND HOLD Invoice Subscription", |handle| async move {
            let future_stream =
                client
                    .invoices()
                    .subscribe_single_invoice(SubscribeSingleInvoiceRequest {
                        r_hash: r_hash.clone(),
                    });

            let mut hold_stream = tokio::select! {
                stream = future_stream => {
                    match stream {
                        Ok(stream) => stream.into_inner(),
                        Err(err) => {
                            crit!(target: LOG_LIGHTNING, err = %err.fmt_compact(), "Failed to subscribe to hold invoice updates, shutting down payment-stream subgroup to trigger gateway reconnect");
                            payment_stream_group.shutdown();
                            return;
                        }
                    }
                },
                () = handle.make_shutdown_rx() => {
                    info!(target: LOG_LIGHTNING, "LND HOLD Invoice Subscription received shutdown signal");
                    return;
                }
            };

            loop {
                let hold = tokio::select! {
                    () = handle.make_shutdown_rx() => {
                        info!(target: LOG_LIGHTNING, "LND HOLD Invoice Subscription received shutdown signal");
                        break;
                    }
                    hold_update = hold_stream.message() => {
                        match hold_update {
                            Ok(Some(hold)) => hold,
                            Ok(None) => {
                                // LND closed the stream because the invoice
                                // reached a terminal state (settled, canceled,
                                // or expired).
                                break;
                            }
                            Err(err) => {
                                crit!(target: LOG_LIGHTNING, err = %err.fmt_compact(), "Error received over hold invoice update stream, shutting down payment-stream subgroup to trigger gateway reconnect");
                                payment_stream_group.shutdown();
                                break;
                            }
                        }
                    }
                };

                debug!(
                    target: LOG_LIGHTNING,
                    payment_hash = %PrettyPaymentHash(&r_hash),
                    state = %hold.state,
                    "LND HOLD Invoice Update",
                );

                if hold.state() == InvoiceState::Accepted {
                    // Only forward HOLD invoices that the gateway created on
                    // behalf of a federation. We check here (rather than at
                    // the new-invoice add-event) because the contract is
                    // saved to gateway_db *after* the HOLD invoice is created
                    // on LND, so the add-event races registration. By the
                    // time `Accepted` fires the HTLC has arrived, which means
                    // the BOLT11 invoice was published and the contract is
                    // committed.
                    let hash = sha256::Hash::from_slice(&hold.r_hash)
                        .expect("LND payment hashes are 32 bytes");
                    if !(self_copy.lnv2_filter)(hash).await {
                        trace!(
                            target: LOG_LIGHTNING,
                            payment_hash = %PrettyPaymentHash(&hold.r_hash),
                            "Ignoring HOLD invoice not created by this gateway",
                        );
                        continue;
                    }

                    if has_accepted_keysend_htlc(&hold.htlcs) {
                        warn!(
                            target: LOG_LIGHTNING,
                            payment_hash = %PrettyPaymentHash(&hold.r_hash),
                            "LNv2 HOLD invoice accepted a keysend HTLC, its claim deadline is not trusted",
                        );
                    }

                    let (incoming_chan_id, htlc_id) = NO_INCOMING_CIRCUIT;
                    let intercept = InterceptPaymentRequest {
                        payment_hash: Hash::from_slice(&hold.r_hash.clone())
                            .expect("Failed to convert to Hash"),
                        // A HOLD invoice reports the real paid amount, so the
                        // two amounts coincide here.
                        amount_msat: hold.amt_paid_msat as u64,
                        incoming_amount_msat: hold.amt_paid_msat as u64,
                        expiry: hold_invoice_claim_deadline(&hold.htlcs),
                        short_channel_id: Some(0),
                        // The payment is held by a HOLD invoice on our own
                        // node rather than by an intercepted forward, which is
                        // how `complete_htlc` knows to resolve it by settling
                        // or canceling that invoice.
                        incoming_chan_id,
                        htlc_id,
                    };

                    match gateway_sender.send(intercept).await {
                        Ok(()) => {}
                        Err(err) => {
                            warn!(
                                target: LOG_LIGHTNING,
                                err = %err.fmt_compact(),
                                "Hold Invoice Subscription failed to send Intercept to gateway"
                            );
                            let _ = self_copy.cancel_hold_invoice(hold.r_hash).await;
                        }
                    }
                }
            }
        });

        Ok(())
    }

    /// Spawns a new background task that subscribes to "add" updates for all
    /// invoices. This is used to detect when a new invoice has been
    /// created. If this invoice is a HOLD invoice, it is potentially destined
    /// for a federation. At this point, we spawn a separate task to monitor the
    /// status of the HOLD invoice.
    async fn spawn_lnv2_invoice_subscription(
        &self,
        task_group: &TaskGroup,
        gateway_sender: HtlcSubscriptionSender,
    ) -> Result<(), LightningRpcError> {
        let mut client = self.connect().await?;

        let list_response = client
            .lightning()
            .list_invoices(ListInvoiceRequest {
                pending_only: true,
                index_offset: 0,
                num_max_invoices: u64::MAX,
                reversed: false,
                ..Default::default()
            })
            .await
            .map_err(|status| {
                warn!(target: LOG_LIGHTNING, status = %status, "Failed to list all invoices");
                LightningRpcError::FailedToRouteHtlcs {
                    failure_reason: "Failed to list all invoices".to_string(),
                }
            })?
            .into_inner();

        let self_copy = self.clone();
        let hold_group = task_group.make_subgroup();
        // See the matching comment in `spawn_lnv1_htlc_interceptor`: if this
        // task exits unexpectedly we shut down the payment-stream subgroup so
        // the gateway transitions to `Disconnected` and reconnects.
        let subgroup = task_group.clone();

        // The `SubscribeInvoices` backlog cannot replay pre-existing pending
        // HOLD invoices reliably: an `add_index` of 0 means "no backlog" to
        // LND, so the oldest pending invoice is skipped whenever it is the
        // first invoice ever created on the node (`add_index == 1`), and
        // invoices already in the `Accepted` state are never delivered to
        // all-invoice subscribers at all. Instead, spawn a monitor task for
        // every pending HOLD invoice (`r_preimage` empty) from the listing
        // directly. Invoices not created by this gateway are filtered out by
        // the monitor task once they reach the `Accepted` state.
        for invoice in &list_response.invoices {
            if invoice.r_preimage.is_empty() {
                info!(
                    target: LOG_LIGHTNING,
                    payment_hash = %PrettyPaymentHash(&invoice.r_hash),
                    "Monitoring pre-existing pending LNv2 invoice",
                );
                self.spawn_lnv2_hold_invoice_subscription(
                    &hold_group,
                    subgroup.clone(),
                    gateway_sender.clone(),
                    invoice.r_hash.clone(),
                )
                .await?;
            }
        }

        // The listing above covers all pending invoices, so the subscription
        // only needs to replay invoices added after the listing was taken.
        // Add indices increase monotonically, so any such invoice has an
        // `add_index` strictly greater than the listing's `last_index_offset`
        // (0 when no invoices were listed) and `SubscribeInvoices` replays
        // exactly those, without duplicating the invoices handled above.
        let add_index = list_response.last_index_offset;
        task_group.spawn("LND Invoice Subscription", move |handle| async move {
            let future_stream = client.lightning().subscribe_invoices(InvoiceSubscription {
                add_index,
                settle_index: u64::MAX, // we do not need settle invoice events
            });
            let mut invoice_stream = tokio::select! {
                stream = future_stream => {
                    match stream {
                        Ok(stream) => stream.into_inner(),
                        Err(err) => {
                            warn!(target: LOG_LIGHTNING, err = %err.fmt_compact(), "Failed to subscribe to all invoice updates");
                            subgroup.shutdown();
                            return;
                        }
                    }
                },
                () = handle.make_shutdown_rx() => {
                    info!(target: LOG_LIGHTNING, "LND Invoice Subscription received shutdown signal");
                    return;
                }
            };

            info!(target: LOG_LIGHTNING, "LND Invoice Subscription: starting to process invoice updates");
            while let Some(invoice) = tokio::select! {
                () = handle.make_shutdown_rx() => {
                    info!(target: LOG_LIGHTNING, "LND Invoice Subscription task received shutdown signal");
                    None
                }
                invoice_update = invoice_stream.message() => {
                    match invoice_update {
                        Ok(invoice) => invoice,
                        Err(err) => {
                            warn!(target: LOG_LIGHTNING, err = %err.fmt_compact(), "Error received over invoice update stream");
                            None
                        }
                    }
                }
            } {
                // If the `r_preimage` is empty and the invoice is OPEN, this means a new HOLD
                // invoice has been created, which is potentially an invoice destined for a
                // federation. We will spawn a new task to monitor the status of
                // the HOLD invoice.
                let payment_hash = invoice.r_hash.clone();

                debug!(
                    target: LOG_LIGHTNING,
                    payment_hash = %PrettyPaymentHash(&payment_hash),
                    state = %invoice.state,
                    "LND HOLD Invoice Update",
                );

                if invoice.r_preimage.is_empty() && invoice.state() == InvoiceState::Open {
                    info!(
                        target: LOG_LIGHTNING,
                        payment_hash = %PrettyPaymentHash(&payment_hash),
                        "Monitoring new LNv2 invoice",
                    );
                    if let Err(err) = self_copy
                        .spawn_lnv2_hold_invoice_subscription(
                            &hold_group,
                            subgroup.clone(),
                            gateway_sender.clone(),
                            payment_hash.clone(),
                        )
                        .await
                    {
                        // Spawning failed because `connect()` exhausted its
                        // retries, a strong signal that LND is unreachable. We
                        // can no longer observe this invoice's `Accepted`
                        // update, so shut down the payment-stream subgroup to
                        // force a gateway reconnect.
                        warn!(
                            target: LOG_LIGHTNING,
                            err = %err.fmt_compact(),
                            payment_hash = %PrettyPaymentHash(&payment_hash),
                            "Failed to spawn HOLD invoice subscription task, shutting down payment-stream subgroup to trigger gateway reconnect",
                        );
                        subgroup.shutdown();
                    }
                }
            }

            if !handle.is_shutting_down() {
                warn!(target: LOG_LIGHTNING, "LND Invoice Subscription exited unexpectedly, shutting down payment-stream subgroup to trigger gateway reconnect");
                subgroup.shutdown();
            }
        });

        Ok(())
    }

    /// Spawns a new background task that intercepts HTLCs from the LND node. In
    /// the LNv1 protocol, this is used as a trigger mechanism for
    /// requesting the Gateway to retrieve the preimage for a payment.
    async fn spawn_lnv1_htlc_interceptor(
        &self,
        task_group: &TaskGroup,
        lnd_sender: mpsc::Sender<ForwardHtlcInterceptResponse>,
        lnd_rx: mpsc::Receiver<ForwardHtlcInterceptResponse>,
        gateway_sender: HtlcSubscriptionSender,
    ) -> Result<(), LightningRpcError> {
        let mut client = self.connect().await?;

        // Verify that LND is reachable via RPC before attempting to spawn a new thread
        // that will intercept HTLCs.
        client
            .lightning()
            .get_info(GetInfoRequest {})
            .await
            .map_err(|status| LightningRpcError::FailedToGetNodeInfo {
                failure_reason: format!("Failed to get node info {status:?}"),
            })?;

        // If the HTLC interceptor exits unexpectedly we shut down the
        // payment-stream subgroup. That cascades to the lnv2 invoice
        // subscription (and its hold-invoice subtasks), which drop their
        // `gateway_sender` clones, closing the gateway's HTLC stream and
        // driving the gateway back to `Disconnected` so it reconnects.
        let subgroup = task_group.clone();
        task_group.spawn("LND HTLC Subscription", |handle| async move {
                let future_stream = client
                    .router()
                    .htlc_interceptor(ReceiverStream::new(lnd_rx));
                let mut htlc_stream = tokio::select! {
                    stream = future_stream => {
                        match stream {
                            Ok(stream) => stream.into_inner(),
                            Err(e) => {
                                crit!(target: LOG_LIGHTNING, err = %e.fmt_compact(), "Failed to establish htlc stream");
                                subgroup.shutdown();
                                return;
                            }
                        }
                    },
                    () = handle.make_shutdown_rx() => {
                        info!(target: LOG_LIGHTNING, "LND HTLC Subscription received shutdown signal while trying to intercept HTLC stream, exiting...");
                        return;
                    }
                };

                debug!(target: LOG_LIGHTNING, "LND HTLC Subscription: starting to process stream");
                // To gracefully handle shutdown signals, we need to be able to receive signals
                // while waiting for the next message from the HTLC stream.
                //
                // If we're in the middle of processing a message from the stream, we need to
                // finish before stopping the spawned task. Checking if the task group is
                // shutting down at the start of each iteration will cause shutdown signals to
                // not process until another message arrives from the HTLC stream, which may
                // take a long time, or never.
                while let Some(htlc) = tokio::select! {
                    () = handle.make_shutdown_rx() => {
                        info!(target: LOG_LIGHTNING, "LND HTLC Subscription task received shutdown signal");
                        None
                    }
                    htlc_message = htlc_stream.message() => {
                        match htlc_message {
                            Ok(htlc) => htlc,
                            Err(err) => {
                                warn!(target: LOG_LIGHTNING, err = %err.fmt_compact(), "Error received over HTLC stream");
                                None
                            }
                    }}
                } {
                    trace!(target: LOG_LIGHTNING, ?htlc, "LND Handling HTLC");

                    let Some(incoming_circuit_key) = htlc.incoming_circuit_key else {
                        // We have no circuit key, so the HTLC cannot be cancelled
                        // either; it will time out at LND. Log enough context to
                        // correlate with the sender's invoice and the target
                        // federation.
                        warn!(
                            target: LOG_LIGHTNING,
                            payment_hash = %PrettyPaymentHash(&htlc.payment_hash),
                            scid = htlc.outgoing_requested_chan_id,
                            amount_msat = htlc.outgoing_amount_msat,
                            "Cannot route HTLC: incoming_circuit_key is None"
                        );
                        continue;
                    };

                    let chan_id = incoming_circuit_key.chan_id;
                    let htlc_id = incoming_circuit_key.htlc_id;

                    // Forward all HTLCs to gatewayd, gatewayd will filter them based on scid
                    let intercept = InterceptPaymentRequest {
                        payment_hash: Hash::from_slice(&htlc.payment_hash).expect("Failed to convert payment Hash"),
                        // `outgoing_amount_msat` is the sender-written onion
                        // `amt_to_forward`; `incoming_amount_msat` is the amount
                        // actually locked in the incoming HTLC. Carry both so
                        // downstream funding is decided against the real value.
                        amount_msat: htlc.outgoing_amount_msat,
                        incoming_amount_msat: htlc.incoming_amount_msat,
                        expiry: htlc.incoming_expiry,
                        short_channel_id: Some(htlc.outgoing_requested_chan_id),
                        incoming_chan_id: chan_id,
                        htlc_id,
                    };

                    match gateway_sender.send(intercept).await {
                        Ok(()) => {}
                        Err(err) => {
                            warn!(
                                target: LOG_LIGHTNING,
                                err = %err.fmt_compact(),
                                payment_hash = %PrettyPaymentHash(&htlc.payment_hash),
                                scid = htlc.outgoing_requested_chan_id,
                                amount_msat = htlc.outgoing_amount_msat,
                                "Failed to send HTLC to gatewayd for processing"
                            );
                            let _ = Self::cancel_htlc(incoming_circuit_key, lnd_sender.clone())
                                .await
                                .map_err(|err| {
                                    warn!(
                                        target: LOG_LIGHTNING,
                                        err = %err.fmt_compact(),
                                        payment_hash = %PrettyPaymentHash(&htlc.payment_hash),
                                        chan_id,
                                        htlc_id,
                                        "Failed to cancel HTLC"
                                    );
                                });
                        }
                    }
                }

                // Loop exited because of an HTLC stream error or end-of-stream
                // (the expected-shutdown case is handled above).
                if !handle.is_shutting_down() {
                    warn!(target: LOG_LIGHTNING, "LND HTLC Subscription exited unexpectedly, shutting down payment-stream subgroup to trigger gateway reconnect");
                    subgroup.shutdown();
                }
            });

        Ok(())
    }

    /// Spawns background tasks for monitoring the status of incoming payments.
    async fn spawn_interceptor(
        &self,
        task_group: &TaskGroup,
        lnd_sender: mpsc::Sender<ForwardHtlcInterceptResponse>,
        lnd_rx: mpsc::Receiver<ForwardHtlcInterceptResponse>,
        gateway_sender: HtlcSubscriptionSender,
    ) -> Result<(), LightningRpcError> {
        self.spawn_lnv1_htlc_interceptor(task_group, lnd_sender, lnd_rx, gateway_sender.clone())
            .await?;

        self.spawn_lnv2_invoice_subscription(task_group, gateway_sender)
            .await?;

        Ok(())
    }

    async fn cancel_htlc(
        key: CircuitKey,
        lnd_sender: mpsc::Sender<ForwardHtlcInterceptResponse>,
    ) -> Result<(), LightningRpcError> {
        // TODO: Specify a failure code and message
        let response = ForwardHtlcInterceptResponse {
            incoming_circuit_key: Some(key),
            action: ResolveHoldForwardAction::Fail.into(),
            preimage: vec![],
            failure_message: vec![],
            failure_code: FailureCode::TemporaryChannelFailure.into(),
            ..Default::default()
        };
        Self::send_lnd_response(lnd_sender, response).await
    }

    async fn send_lnd_response(
        lnd_sender: mpsc::Sender<ForwardHtlcInterceptResponse>,
        response: ForwardHtlcInterceptResponse,
    ) -> Result<(), LightningRpcError> {
        // TODO: Consider retrying this if the send fails
        lnd_sender.send(response).await.map_err(|send_error| {
            LightningRpcError::FailedToCompleteHtlc {
                failure_reason: format!(
                    "Failed to send ForwardHtlcInterceptResponse to LND {send_error:?}"
                ),
            }
        })
    }

    async fn lookup_payment(
        &self,
        payment_hash: Vec<u8>,
        client: &mut LndClient,
    ) -> Result<Option<String>, LightningRpcError> {
        // Loop until we successfully get the status of the payment, or determine that
        // the payment has not been made yet.
        loop {
            let payments = client
                .router()
                .track_payment_v2(TrackPaymentRequest {
                    payment_hash: payment_hash.clone(),
                    no_inflight_updates: true,
                })
                .await;

            match payments {
                Ok(payments) => {
                    let mut updates = payments.into_inner();

                    // Block until LND reports a terminal status. `no_inflight_updates`
                    // asks LND to hold back everything else, but a payment still in
                    // flight must never be read as failed, so only `Failed` counts as
                    // a failure here: `InFlight`, `Initiated` (with
                    // `routerrpc.usestatusinitiated`) and any status this build does
                    // not know, which prost decodes as `Unknown`, keep us waiting.
                    let outcome = loop {
                        match updates.message().await {
                            Ok(Some(payment)) if payment.status() == PaymentStatus::Succeeded => {
                                return Ok(Some(payment.payment_preimage));
                            }
                            Ok(Some(payment)) if payment.status() == PaymentStatus::Failed => {
                                let failure_reason = payment.failure_reason();
                                return Err(LightningRpcError::FailedPayment {
                                    failure_reason: format!("{failure_reason:?}"),
                                });
                            }
                            Ok(Some(payment)) => {
                                debug!(
                                    target: LOG_LIGHTNING,
                                    payment_hash = %PrettyPaymentHash(&payment_hash),
                                    status = ?payment.status(),
                                    "Tracked payment is not terminal yet, waiting",
                                );
                            }
                            outcome => break outcome,
                        }
                    };

                    // A premature end of stream (`Ok(None)`) or a transport fault
                    // (`Err`) is not a payment outcome. Retry rather than reporting a
                    // failure the node never produced.
                    warn!(
                        target: LOG_LIGHTNING,
                        payment_hash = %PrettyPaymentHash(&payment_hash),
                        outcome = ?outcome,
                        "Payment tracking stream ended or faulted. Trying again in 5 seconds"
                    );
                    sleep(Duration::from_secs(5)).await;
                }
                Err(err) => {
                    // Break if we got a response back from the LND node that indicates the payment
                    // hash was not found.
                    if err.code() == Code::NotFound {
                        return Ok(None);
                    }

                    warn!(
                        target: LOG_LIGHTNING,
                        payment_hash = %PrettyPaymentHash(&payment_hash),
                        err = %err.fmt_compact(),
                        "Could not get the status of payment. Trying again in 5 seconds"
                    );
                    sleep(Duration::from_secs(5)).await;
                }
            }
        }
    }

    /// Looks up the invoice carrying `payment_hash`, returning `None` if the
    /// node has no such invoice.
    ///
    /// Any other lookup failure is reported as an error, since callers retry on
    /// error and an unreachable node is a condition a retry can clear.
    async fn lookup_invoice(
        client: &mut LndClient,
        payment_hash: &[u8],
    ) -> Result<Option<Invoice>, LightningRpcError> {
        match client
            .invoices()
            .lookup_invoice_v2(LookupInvoiceMsg {
                invoice_ref: Some(InvoiceRef::PaymentHash(payment_hash.to_vec())),
                lookup_modifier: 0,
            })
            .await
        {
            Ok(invoice) => Ok(Some(invoice.into_inner())),
            Err(err) if err.code() == Code::NotFound => Ok(None),
            Err(err) => Err(LightningRpcError::FailedToCompleteHtlc {
                failure_reason: format!("Failed to look up invoice: {}", err.fmt_compact()),
            }),
        }
    }

    /// Settles the LNv2 HOLD invoice carrying `payment_hash` with `preimage`.
    ///
    /// Only an already-settled invoice is an idempotent success. Missing,
    /// nonterminal, and canceled invoices fail so callers cannot record a
    /// settle outcome that Lightning did not produce.
    async fn settle_hold_invoice(
        &self,
        payment_hash: Vec<u8>,
        preimage: Preimage,
    ) -> Result<(), LightningRpcError> {
        let mut client = self.connect().await?;
        let invoice = Self::lookup_invoice(&mut client, &payment_hash).await?;
        match hold_invoice_action(
            PaymentActionKind::Settle,
            invoice.as_ref().map(Invoice::state),
        ) {
            Ok(HoldInvoiceAction::Complete) => {}
            Ok(HoldInvoiceAction::AlreadyComplete) => {
                info!(
                    target: LOG_LIGHTNING,
                    payment_hash = %PrettyPaymentHash(&payment_hash),
                    "HOLD invoice was already settled",
                );
                return Ok(());
            }
            Err(error) => {
                warn!(
                    target: LOG_LIGHTNING,
                    state = ?invoice.as_ref().map(Invoice::state),
                    payment_hash = %PrettyPaymentHash(&payment_hash),
                    failure_reason = error.failure_reason,
                    "Cannot settle HOLD invoice",
                );
                return Err(if error.permanent {
                    LightningRpcError::HtlcCompletionRejected {
                        failure_reason: error.failure_reason.to_owned(),
                    }
                } else {
                    LightningRpcError::FailedToCompleteHtlc {
                        failure_reason: error.failure_reason.to_owned(),
                    }
                });
            }
        }

        client
            .invoices()
            .settle_invoice(SettleInvoiceMsg {
                preimage: preimage.0.to_vec(),
            })
            .await
            .map_err(|err| {
                warn!(
                    target: LOG_LIGHTNING,
                    err = %err.fmt_compact(),
                    payment_hash = %PrettyPaymentHash(&payment_hash),
                    "Failed to settle HOLD invoice",
                );
                LightningRpcError::FailedToCompleteHtlc {
                    failure_reason: "Failed to settle HOLD invoice".to_string(),
                }
            })?;

        info!(
            target: LOG_LIGHTNING,
            payment_hash = %PrettyPaymentHash(&payment_hash),
            "Successfully settled HOLD invoice",
        );

        Ok(())
    }

    /// Cancels the LNv2 HOLD invoice carrying `payment_hash`, failing back any
    /// HTLC it holds.
    ///
    /// Only an already-canceled invoice is an idempotent success. A settled
    /// invoice fails so callers cannot record a cancel outcome after a racing
    /// settle won.
    async fn cancel_hold_invoice(&self, payment_hash: Vec<u8>) -> Result<(), LightningRpcError> {
        let mut client = self.connect().await?;
        let invoice = Self::lookup_invoice(&mut client, &payment_hash).await?;
        match hold_invoice_action(
            PaymentActionKind::Cancel,
            invoice.as_ref().map(Invoice::state),
        ) {
            Ok(HoldInvoiceAction::Complete) => {}
            Ok(HoldInvoiceAction::AlreadyComplete) => {
                info!(
                    target: LOG_LIGHTNING,
                    payment_hash = %PrettyPaymentHash(&payment_hash),
                    "HOLD invoice was already canceled",
                );
                return Ok(());
            }
            Err(error) => {
                warn!(
                    target: LOG_LIGHTNING,
                    state = ?invoice.as_ref().map(Invoice::state),
                    payment_hash = %PrettyPaymentHash(&payment_hash),
                    failure_reason = error.failure_reason,
                    "Cannot cancel HOLD invoice",
                );
                return Err(LightningRpcError::HtlcCompletionRejected {
                    failure_reason: error.failure_reason.to_owned(),
                });
            }
        }

        client
            .invoices()
            .cancel_invoice(CancelInvoiceMsg {
                payment_hash: payment_hash.clone(),
            })
            .await
            .map_err(|err| {
                warn!(
                    target: LOG_LIGHTNING,
                    err = %err.fmt_compact(),
                    payment_hash = %PrettyPaymentHash(&payment_hash),
                    "Failed to cancel HOLD invoice",
                );
                LightningRpcError::FailedToCompleteHtlc {
                    failure_reason: "Failed to cancel HOLD invoice".to_string(),
                }
            })?;

        info!(
            target: LOG_LIGHTNING,
            payment_hash = %PrettyPaymentHash(&payment_hash),
            "Successfully canceled HOLD invoice",
        );

        Ok(())
    }
}

impl fmt::Debug for GatewayLndClient {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(f, "LndClient")
    }
}

#[async_trait]
impl ILnRpcClient for GatewayLndClient {
    async fn info(&self) -> Result<GetNodeInfoResponse, LightningRpcError> {
        let mut client = self.connect().await?;
        let info = client
            .lightning()
            .get_info(GetInfoRequest {})
            .await
            .map_err(|status| LightningRpcError::FailedToGetNodeInfo {
                failure_reason: format!("Failed to get node info {status:?}"),
            })?
            .into_inner();

        let pub_key: PublicKey =
            info.identity_pubkey
                .parse()
                .map_err(|e| LightningRpcError::FailedToGetNodeInfo {
                    failure_reason: format!("Failed to parse public key {e:?}"),
                })?;

        let network = match info
            .chains
            .first()
            .ok_or_else(|| LightningRpcError::FailedToGetNodeInfo {
                failure_reason: "Failed to parse node network".to_string(),
            })?
            .network
            .as_str()
        {
            // LND uses "mainnet", but rust-bitcoin uses "bitcoin".
            // TODO: create a fedimint `Network` type that understands "mainnet"
            "mainnet" => "bitcoin",
            other => other,
        }
        .to_string();

        return Ok(GetNodeInfoResponse {
            pub_key,
            alias: info.alias,
            network,
            block_height: info.block_height,
            synced_to_chain: info.synced_to_chain,
        });
    }

    async fn routehints(
        &self,
        num_route_hints: usize,
    ) -> Result<GetRouteHintsResponse, LightningRpcError> {
        let mut client = self.connect().await?;
        let mut channels = client
            .lightning()
            .list_channels(ListChannelsRequest {
                active_only: true,
                inactive_only: false,
                public_only: false,
                private_only: false,
                peer: vec![],
                peer_alias_lookup: false,
            })
            .await
            .map_err(|status| LightningRpcError::FailedToGetRouteHints {
                failure_reason: format!("Failed to list channels {status:?}"),
            })?
            .into_inner()
            .channels;

        // Take the channels with the largest incoming capacity
        channels.sort_by_key(|b| std::cmp::Reverse(b.remote_balance));
        channels.truncate(num_route_hints);

        let mut route_hints: Vec<RouteHint> = vec![];
        for chan in &channels {
            let info = client
                .lightning()
                .get_chan_info(ChanInfoRequest {
                    chan_id: chan.chan_id,
                    ..Default::default()
                })
                .await
                .map_err(|status| LightningRpcError::FailedToGetRouteHints {
                    failure_reason: format!("Failed to get channel info {status:?}"),
                })?
                .into_inner();

            // The route hint hop describes the remote peer forwarding a payment
            // *into* the gateway, so it must carry that peer's advertised channel
            // policy. `node1`/`node2` are ordered by pubkey (BOLT 7), not by which
            // side is the gateway, so select the policy belonging to the remote
            // peer rather than always taking node1's.
            let policy = if info.node1_pub == chan.remote_pubkey {
                info.node1_policy
            } else {
                info.node2_policy
            };
            let Some(policy) = policy else {
                continue;
            };
            let src_node_id =
                PublicKey::from_str(&chan.remote_pubkey).expect("Failed to parse pubkey");
            let short_channel_id = chan.chan_id;
            let base_msat = policy.fee_base_msat as u32;
            let proportional_millionths = policy.fee_rate_milli_msat as u32;
            let cltv_expiry_delta = policy.time_lock_delta;
            let htlc_maximum_msat = Some(policy.max_htlc_msat);
            let htlc_minimum_msat = Some(policy.min_htlc as u64);

            let route_hint_hop = RouteHintHop {
                src_node_id,
                short_channel_id,
                base_msat,
                proportional_millionths,
                cltv_expiry_delta: cltv_expiry_delta as u16,
                htlc_minimum_msat,
                htlc_maximum_msat,
            };
            route_hints.push(RouteHint(vec![route_hint_hop]));
        }

        Ok(GetRouteHintsResponse { route_hints })
    }

    async fn pay_private(
        &self,
        invoice: PrunedInvoice,
        max_delay: u64,
        max_fee: Amount,
    ) -> Result<PayInvoiceResponse, LightningRpcError> {
        let payment_hash = invoice.payment_hash.to_byte_array().to_vec();
        info!(
            target: LOG_LIGHTNING,
            payment_hash = %PrettyPaymentHash(&payment_hash),
            "LND Paying invoice",
        );
        let mut client = self.connect().await?;

        debug!(
            target: LOG_LIGHTNING,
            payment_hash = %PrettyPaymentHash(&payment_hash),
            "pay_private checking if payment for invoice exists"
        );

        // If the payment exists, that means we've already tried to pay the invoice
        let preimage: Vec<u8> = match self
            .lookup_payment(invoice.payment_hash.to_byte_array().to_vec(), &mut client)
            .await?
        {
            Some(preimage) => {
                info!(
                    target: LOG_LIGHTNING,
                    payment_hash = %PrettyPaymentHash(&payment_hash),
                    "LND payment already exists for invoice",
                );
                hex::FromHex::from_hex(preimage.as_str()).map_err(|error| {
                    LightningRpcError::FailedPayment {
                        failure_reason: format!("Failed to convert preimage {error:?}"),
                    }
                })?
            }
            _ => 'dispatch: {
                // LND API allows fee limits in the `i64` range, but we use `u64` for
                // max_fee_msat. This means we can only set an enforceable fee limit
                // between 0 and i64::MAX
                let fee_limit_msat: i64 =
                    max_fee
                        .msats
                        .try_into()
                        .map_err(|error| LightningRpcError::FailedPayment {
                            failure_reason: format!(
                                "max_fee_msat exceeds valid LND fee limit ranges {error:?}"
                            ),
                        })?;

                let amt_msat = invoice.amount.msats.try_into().map_err(|error| {
                    LightningRpcError::FailedPayment {
                        failure_reason: format!("amount exceeds valid LND amount ranges {error:?}"),
                    }
                })?;
                let final_cltv_delta =
                    invoice.min_final_cltv_delta.try_into().map_err(|error| {
                        LightningRpcError::FailedPayment {
                            failure_reason: format!(
                                "final cltv delta exceeds valid LND range {error:?}"
                            ),
                        }
                    })?;
                // LND reads a `cltv_limit` of zero as "no limit set" and
                // enforces its `--max-cltv-expiry` default instead, silently
                // lifting the caller's timelock cap.
                if max_delay == 0 {
                    return Err(LightningRpcError::FailedPayment {
                        failure_reason: "a max delay of zero would disable LND's CLTV limit"
                            .to_string(),
                    });
                }
                let cltv_limit =
                    max_delay
                        .try_into()
                        .map_err(|error| LightningRpcError::FailedPayment {
                            failure_reason: format!("max delay exceeds valid LND range {error:?}"),
                        })?;

                let dest_features = wire_features_to_lnd_feature_vec(&invoice.destination_features)
                    .map_err(|e| LightningRpcError::FailedPayment {
                        failure_reason: e.to_string(),
                    })?;

                debug!(
                    target: LOG_LIGHTNING,
                    payment_hash = %PrettyPaymentHash(&payment_hash),
                    "LND payment does not exist, will attempt to pay",
                );
                let payments = match client
                    .router()
                    .send_payment_v2(SendPaymentRequest {
                        amt_msat,
                        dest: invoice.destination.serialize().to_vec(),
                        dest_features,
                        payment_hash: invoice.payment_hash.to_byte_array().to_vec(),
                        payment_addr: invoice.payment_secret.to_vec(),
                        route_hints: route_hints_to_lnd(&invoice.route_hints),
                        final_cltv_delta,
                        cltv_limit,
                        no_inflight_updates: false,
                        timeout_seconds: self.payment_timeout_secs,
                        fee_limit_msat,
                        time_pref: self.time_pref,
                        ..Default::default()
                    })
                    .await
                {
                    Ok(payments) => payments,
                    // An error here is not proof that nothing was paid. LND
                    // registers the payment before its first status update,
                    // so a transport fault on the way back can surface as an
                    // error for a payment that is now in flight. Reporting
                    // failure would forfeit the outgoing contract while the
                    // payment may still settle, so only the node's own record
                    // decides.
                    Err(status) => {
                        warn!(
                            target: LOG_LIGHTNING,
                            status = %status,
                            payment_hash = %PrettyPaymentHash(&payment_hash),
                            "LND payment request failed, checking whether LND registered the payment",
                        );

                        let Some(preimage) = self
                            .lookup_payment(payment_hash.clone(), &mut client)
                            .await?
                        else {
                            return Err(LightningRpcError::FailedPayment {
                                failure_reason: format!(
                                    "Failed to make outgoing payment {status:?}"
                                ),
                            });
                        };

                        break 'dispatch hex::FromHex::from_hex(preimage.as_str()).map_err(
                            |error| LightningRpcError::FailedPayment {
                                failure_reason: format!("Failed to convert preimage {error:?}"),
                            },
                        )?;
                    }
                };

                debug!(
                    target: LOG_LIGHTNING,
                    payment_hash = %PrettyPaymentHash(&payment_hash),
                    "LND payment request sent, waiting for payment status...",
                );
                let mut messages = payments.into_inner();
                loop {
                    match messages.message().await {
                        Ok(Some(payment)) if payment.status() == PaymentStatus::Succeeded => {
                            info!(
                                target: LOG_LIGHTNING,
                                payment_hash = %PrettyPaymentHash(&payment_hash),
                                "LND payment succeeded for invoice",
                            );
                            break hex::FromHex::from_hex(payment.payment_preimage.as_str())
                                .map_err(|error| LightningRpcError::FailedPayment {
                                    failure_reason: format!("Failed to convert preimage {error:?}"),
                                })?;
                        }
                        Ok(Some(payment)) if payment.status() == PaymentStatus::Failed => {
                            // The one terminal status besides `Succeeded`: a
                            // definitive failure that is safe to report.
                            warn!(
                                target: LOG_LIGHTNING,
                                payment_hash = %PrettyPaymentHash(&payment_hash),
                                status = ?payment.status(),
                                "LND payment failed",
                            );
                            let failure_reason = payment.failure_reason();
                            return Err(LightningRpcError::FailedPayment {
                                failure_reason: format!("{failure_reason:?}"),
                            });
                        }
                        // `InFlight`, `Initiated` (delivered before the first HTLC
                        // when `routerrpc.usestatusinitiated` is set) and any status
                        // this build does not know, which prost decodes as `Unknown`,
                        // are not outcomes. Keep waiting; a stream that ends without
                        // a terminal status resumes through `lookup_payment` below.
                        Ok(Some(payment)) => {
                            debug!(
                                target: LOG_LIGHTNING,
                                payment_hash = %PrettyPaymentHash(&payment_hash),
                                status = ?payment.status(),
                                "LND payment is in flight",
                            );
                            continue;
                        }
                        // `Ok(None)` is a premature end of the update stream and `Err`
                        // is a tonic/HTTP2 transport fault. Neither is a payment
                        // outcome: the HTLC may still settle, so reporting failure here
                        // would forfeit the outgoing contract for a payment that is
                        // still in flight, violating the idempotency contract on
                        // `ILnRpcClient::pay`. Resume tracking with `track_payment_v2`
                        // to await the real terminal result instead.
                        stream_end_or_fault => {
                            warn!(
                                target: LOG_LIGHTNING,
                                payment_hash = %PrettyPaymentHash(&payment_hash),
                                outcome = ?stream_end_or_fault,
                                "LND payment status stream ended or faulted before a terminal status; resuming tracking",
                            );
                            match self
                                .lookup_payment(payment_hash.clone(), &mut client)
                                .await?
                            {
                                Some(preimage) => {
                                    break hex::FromHex::from_hex(preimage.as_str()).map_err(
                                        |error| LightningRpcError::FailedPayment {
                                            failure_reason: format!(
                                                "Failed to convert preimage {error:?}"
                                            ),
                                        },
                                    )?;
                                }
                                None => {
                                    return Err(LightningRpcError::FailedPayment {
                                        failure_reason: format!(
                                            "LND has no record of dispatched payment for hash {:?}",
                                            invoice.payment_hash
                                        ),
                                    });
                                }
                            }
                        }
                    }
                }
            }
        };
        Ok(PayInvoiceResponse {
            preimage: Preimage(preimage.try_into().expect("Failed to create preimage")),
        })
    }

    /// Returns true if the lightning backend supports payments without full
    /// invoices
    fn supports_private_payments(&self) -> bool {
        true
    }

    async fn outbound_payment_exists(
        &self,
        payment_hash: sha256::Hash,
    ) -> Result<bool, LightningRpcError> {
        let payment_hash_bytes = payment_hash.to_byte_array().to_vec();
        let mut client = self.connect().await?;

        // Subscribe with in-flight updates enabled so any known payment,
        // pending or terminal, yields an immediate first message instead of
        // blocking until the payment resolves; an unknown hash fails with
        // `NotFound`.
        let stream = match client
            .router()
            .track_payment_v2(TrackPaymentRequest {
                payment_hash: payment_hash_bytes.clone(),
                no_inflight_updates: false,
            })
            .await
        {
            Ok(stream) => stream,
            Err(status) if status.code() == Code::NotFound => return Ok(false),
            Err(status) => {
                return Err(LightningRpcError::FailedPayment {
                    failure_reason: format!(
                        "Failed to look up payment {}: {status:?}",
                        PrettyPaymentHash(&payment_hash_bytes),
                    ),
                });
            }
        };

        match stream.into_inner().message().await {
            Ok(Some(_)) => Ok(true),
            Err(status) if status.code() == Code::NotFound => Ok(false),
            // A premature end of stream or a transport fault is not an
            // answer. Report an error so the caller retries, rather than
            // letting it mistake the payment for never having been
            // dispatched.
            outcome => Err(LightningRpcError::FailedPayment {
                failure_reason: format!(
                    "Payment lookup stream gave no answer for {}: {outcome:?}",
                    PrettyPaymentHash(&payment_hash_bytes),
                ),
            }),
        }
    }

    async fn route_htlcs<'a>(
        self: Box<Self>,
        task_group: &TaskGroup,
    ) -> Result<(RouteHtlcStream<'a>, Arc<dyn ILnRpcClient>), LightningRpcError> {
        const CHANNEL_SIZE: usize = 100;

        // Channel to send intercepted htlc to the gateway for processing
        let (gateway_sender, gateway_receiver) =
            mpsc::channel::<InterceptPaymentRequest>(CHANNEL_SIZE);

        let (lnd_sender, lnd_rx) = mpsc::channel::<ForwardHtlcInterceptResponse>(CHANNEL_SIZE);

        self.spawn_interceptor(
            task_group,
            lnd_sender.clone(),
            lnd_rx,
            gateway_sender.clone(),
        )
        .await?;
        let new_client = Arc::new(Self {
            address: self.address.clone(),
            tls_cert: self.tls_cert.clone(),
            macaroon: self.macaroon.clone(),
            time_pref: self.time_pref,
            payment_timeout_secs: self.payment_timeout_secs,
            lnd_sender: Some(lnd_sender.clone()),
            lnv2_filter: self.lnv2_filter.clone(),
        });
        Ok((Box::pin(ReceiverStream::new(gateway_receiver)), new_client))
    }

    async fn complete_htlc(&self, htlc: InterceptPaymentResponse) -> Result<(), LightningRpcError> {
        let incoming_circuit = htlc.incoming_circuit();
        let InterceptPaymentResponse {
            action,
            payment_hash,
            incoming_chan_id: _,
            htlc_id: _,
        } = htlc;

        let (action, preimage) = match action {
            PaymentAction::Settle(preimage) => (ResolveHoldForwardAction::Settle, preimage),
            PaymentAction::Cancel => (ResolveHoldForwardAction::Fail, Preimage([0; 32])),
            PaymentAction::Forward => (ResolveHoldForwardAction::Resume, Preimage([0; 32])),
        };

        // Resolve the payment the way it arrived. Deciding instead by probing
        // LND for a HOLD invoice carrying the payment hash would conflate the
        // two ways, because the hash is chosen by whoever is being paid: an
        // attacker can register an LNv2 receive and an LNv1 offer for the same
        // hash, and the completion for the intercepted LNv1 HTLC would then
        // settle or cancel the unrelated LNv2 HOLD invoice.
        let Some((chan_id, htlc_id)) = incoming_circuit else {
            // LNv2: the payment is held by a HOLD invoice on our own node, so
            // there is no forward to resolve.
            return match action {
                ResolveHoldForwardAction::Settle => {
                    self.settle_hold_invoice(payment_hash.to_byte_array().to_vec(), preimage)
                        .await
                }
                // Neither `Fail` nor `Resume` has a meaning for a HOLD invoice
                // beyond "the gateway could not claim this payment": there is
                // no next hop to resume towards, so fail it back to the payer.
                _ => {
                    self.cancel_hold_invoice(payment_hash.to_byte_array().to_vec())
                        .await
                }
            };
        };

        // LNv1: hand the interceptor its response for this exact circuit.
        let Some(lnd_sender) = self.lnd_sender.clone() else {
            crit!("Gatewayd has not started to route HTLCs");
            return Err(LightningRpcError::FailedToCompleteHtlc {
                failure_reason: "Gatewayd has not started to route HTLCs".to_string(),
            });
        };

        let response = ForwardHtlcInterceptResponse {
            incoming_circuit_key: Some(CircuitKey { chan_id, htlc_id }),
            action: action.into(),
            preimage: preimage.0.to_vec(),
            failure_message: vec![],
            failure_code: FailureCode::TemporaryChannelFailure.into(),
            ..Default::default()
        };

        Self::send_lnd_response(lnd_sender, response).await
    }

    async fn create_invoice(
        &self,
        create_invoice_request: CreateInvoiceRequest,
    ) -> Result<CreateInvoiceResponse, LightningRpcError> {
        let mut client = self.connect().await?;
        let description = create_invoice_request
            .description
            .unwrap_or(InvoiceDescription::Direct(String::new()));

        if let Some(payment_hash_value) = create_invoice_request.payment_hash {
            let payment_hash = payment_hash_value.to_byte_array().to_vec();
            let hold_invoice_request = match description {
                InvoiceDescription::Direct(description) => AddHoldInvoiceRequest {
                    memo: description,
                    hash: payment_hash.clone(),
                    value_msat: create_invoice_request.amount_msat as i64,
                    expiry: i64::from(create_invoice_request.expiry_secs),
                    cltv_expiry: LNV2_HOLD_INVOICE_CLTV_EXPIRY,
                    ..Default::default()
                },
                InvoiceDescription::Hash(desc_hash) => AddHoldInvoiceRequest {
                    description_hash: desc_hash.to_byte_array().to_vec(),
                    hash: payment_hash.clone(),
                    value_msat: create_invoice_request.amount_msat as i64,
                    expiry: i64::from(create_invoice_request.expiry_secs),
                    cltv_expiry: LNV2_HOLD_INVOICE_CLTV_EXPIRY,
                    ..Default::default()
                },
            };

            let hold_invoice_response = client
                .invoices()
                .add_hold_invoice(hold_invoice_request)
                .await
                .map_err(|e| LightningRpcError::FailedToGetInvoice {
                    failure_reason: e.to_string(),
                })?;

            let invoice = hold_invoice_response.into_inner().payment_request;
            Ok(CreateInvoiceResponse { invoice })
        } else {
            let invoice = match description {
                InvoiceDescription::Direct(description) => Invoice {
                    memo: description,
                    value_msat: create_invoice_request.amount_msat as i64,
                    expiry: i64::from(create_invoice_request.expiry_secs),
                    ..Default::default()
                },
                InvoiceDescription::Hash(desc_hash) => Invoice {
                    description_hash: desc_hash.to_byte_array().to_vec(),
                    value_msat: create_invoice_request.amount_msat as i64,
                    expiry: i64::from(create_invoice_request.expiry_secs),
                    ..Default::default()
                },
            };

            let add_invoice_response =
                client.lightning().add_invoice(invoice).await.map_err(|e| {
                    LightningRpcError::FailedToGetInvoice {
                        failure_reason: e.to_string(),
                    }
                })?;

            let invoice = add_invoice_response.into_inner().payment_request;
            Ok(CreateInvoiceResponse { invoice })
        }
    }

    async fn get_ln_onchain_address(
        &self,
    ) -> Result<GetLnOnchainAddressResponse, LightningRpcError> {
        let mut client = self.connect().await?;

        match client
            .wallet()
            .next_addr(AddrRequest {
                account: String::new(), // Default wallet account.
                r#type: 4,              // Taproot address.
                change: false,
            })
            .await
        {
            Ok(response) => Ok(GetLnOnchainAddressResponse {
                address: response.into_inner().addr,
            }),
            Err(e) => Err(LightningRpcError::FailedToGetLnOnchainAddress {
                failure_reason: format!("Failed to get funding address {e:?}"),
            }),
        }
    }

    async fn send_onchain(
        &self,
        SendOnchainRequest {
            address,
            amount,
            fee_rate_sats_per_vbyte,
        }: SendOnchainRequest,
    ) -> Result<SendOnchainResponse, LightningRpcError> {
        #[allow(deprecated)]
        let request = match amount {
            BitcoinAmountOrAll::All => SendCoinsRequest {
                addr: address.assume_checked().to_string(),
                amount: 0,
                target_conf: 0,
                sat_per_vbyte: fee_rate_sats_per_vbyte,
                sat_per_byte: 0,
                send_all: true,
                label: String::new(),
                min_confs: 0,
                spend_unconfirmed: true,
                ..Default::default()
            },
            BitcoinAmountOrAll::Amount(amount) => SendCoinsRequest {
                addr: address.assume_checked().to_string(),
                amount: amount.to_sat() as i64,
                target_conf: 0,
                sat_per_vbyte: fee_rate_sats_per_vbyte,
                sat_per_byte: 0,
                send_all: false,
                label: String::new(),
                min_confs: 0,
                spend_unconfirmed: true,
                ..Default::default()
            },
        };

        match self.connect().await?.lightning().send_coins(request).await {
            Ok(res) => Ok(SendOnchainResponse {
                txid: res.into_inner().txid,
            }),
            Err(e) => Err(LightningRpcError::FailedToWithdrawOnchain {
                failure_reason: format!("Failed to withdraw funds on-chain {e:?}"),
            }),
        }
    }

    async fn open_channel(
        &self,
        crate::OpenChannelRequest {
            pubkey,
            host,
            channel_size_sats,
            push_amount_sats,
            fee_rate_sats_per_vbyte,
            base_fee_msat,
            parts_per_million,
        }: crate::OpenChannelRequest,
    ) -> Result<OpenChannelResponse, LightningRpcError> {
        let mut client = self.connect().await?;

        self.connect_peer_if_needed(&mut client, pubkey, host)
            .await?;

        // Build the request, leaving unspecified fee fields at their
        // protobuf defaults so LND falls back to its own configuration.
        let mut open_request = OpenChannelRequest {
            node_pubkey: pubkey.serialize().to_vec(),
            local_funding_amount: channel_size_sats.try_into().expect("u64 -> i64"),
            push_sat: push_amount_sats.try_into().expect("u64 -> i64"),
            ..Default::default()
        };
        if let Some(rate) = fee_rate_sats_per_vbyte {
            open_request.sat_per_vbyte = rate;
        }
        if let Some(base_fee) = base_fee_msat {
            open_request.base_fee = base_fee;
            open_request.use_base_fee = true;
        }
        if let Some(ppm) = parts_per_million {
            open_request.fee_rate = ppm;
            open_request.use_fee_rate = true;
        }

        // Open the channel
        match client.lightning().open_channel_sync(open_request).await {
            Ok(res) => Ok(OpenChannelResponse {
                funding_txid: match res.into_inner().funding_txid {
                    Some(txid) => match txid {
                        FundingTxid::FundingTxidBytes(mut bytes) => {
                            bytes.reverse();
                            hex::encode(bytes)
                        }
                        FundingTxid::FundingTxidStr(str) => str,
                    },
                    None => String::new(),
                },
            }),
            Err(e) => Err(LightningRpcError::FailedToOpenChannel {
                failure_reason: format!("Failed to open channel {e:?}"),
            }),
        }
    }

    async fn connect_peer(&self, payload: ConnectPeerRequest) -> Result<(), LightningRpcError> {
        let mut client = self.connect().await?;
        self.connect_peer_if_needed(
            &mut client,
            payload.node_address.pubkey,
            payload.node_address.host_with_port(),
        )
        .await
    }

    async fn close_channels_with_peer(
        &self,
        CloseChannelsWithPeerRequest {
            pubkey,
            force,
            sats_per_vbyte,
        }: CloseChannelsWithPeerRequest,
    ) -> Result<CloseChannelsWithPeerResponse, LightningRpcError> {
        let mut client = self.connect().await?;

        let channels_with_peer = client
            .lightning()
            .list_channels(ListChannelsRequest {
                active_only: false,
                inactive_only: false,
                public_only: false,
                private_only: false,
                peer: pubkey.serialize().to_vec(),
                peer_alias_lookup: false,
            })
            .await
            .map_err(|e| LightningRpcError::FailedToCloseChannelsWithPeer {
                failure_reason: format!("Failed to list channels {e:?}"),
            })?
            .into_inner()
            .channels;

        for channel in &channels_with_peer {
            let channel_point =
                bitcoin::OutPoint::from_str(&channel.channel_point).map_err(|e| {
                    LightningRpcError::FailedToCloseChannelsWithPeer {
                        failure_reason: format!("Failed to parse channel point {e:?}"),
                    }
                })?;

            if force {
                client
                    .lightning()
                    .close_channel(CloseChannelRequest {
                        channel_point: Some(ChannelPoint {
                            funding_txid: Some(
                                tonic_lnd::lnrpc::channel_point::FundingTxid::FundingTxidBytes(
                                    <bitcoin::Txid as AsRef<[u8]>>::as_ref(&channel_point.txid)
                                        .to_vec(),
                                ),
                            ),
                            output_index: channel_point.vout,
                        }),
                        force,
                        ..Default::default()
                    })
                    .await
                    .map_err(|e| LightningRpcError::FailedToCloseChannelsWithPeer {
                        failure_reason: format!("Failed to close channel {e:?}"),
                    })?;
            } else {
                client
                    .lightning()
                    .close_channel(CloseChannelRequest {
                        channel_point: Some(ChannelPoint {
                            funding_txid: Some(
                                tonic_lnd::lnrpc::channel_point::FundingTxid::FundingTxidBytes(
                                    <bitcoin::Txid as AsRef<[u8]>>::as_ref(&channel_point.txid)
                                        .to_vec(),
                                ),
                            ),
                            output_index: channel_point.vout,
                        }),
                        force,
                        sat_per_vbyte: sats_per_vbyte.unwrap_or_default(),
                        ..Default::default()
                    })
                    .await
                    .map_err(|e| LightningRpcError::FailedToCloseChannelsWithPeer {
                        failure_reason: format!("Failed to close channel {e:?}"),
                    })?;
            }
        }

        Ok(CloseChannelsWithPeerResponse {
            num_channels_closed: channels_with_peer.len() as u32,
        })
    }

    async fn list_channels(&self) -> Result<ListChannelsResponse, LightningRpcError> {
        let mut client = self.connect().await?;

        // Fetch peer addresses so we can populate remote_address on each channel
        let peer_addresses: BTreeMap<String, String> = client
            .lightning()
            .list_peers(ListPeersRequest {
                latest_error: false,
            })
            .await
            .map(|resp| {
                resp.into_inner()
                    .peers
                    .into_iter()
                    .filter_map(|peer| {
                        if peer.address.is_empty() {
                            None
                        } else {
                            Some((peer.pub_key, peer.address))
                        }
                    })
                    .collect()
            })
            .unwrap_or_default();

        // Fetch the local fee policy for every channel in one call so we can
        // join it onto each ChannelInfo below.
        let fee_report: BTreeMap<u64, (u64, u64)> = client
            .lightning()
            .fee_report(FeeReportRequest {})
            .await
            .map(|resp| {
                resp.into_inner()
                    .channel_fees
                    .into_iter()
                    .map(|report| {
                        let base_fee_msat = u64::try_from(report.base_fee_msat).unwrap_or_default();
                        let parts_per_million =
                            u64::try_from(report.fee_per_mil).unwrap_or_default();
                        (report.chan_id, (base_fee_msat, parts_per_million))
                    })
                    .collect()
            })
            .unwrap_or_default();

        match client
            .lightning()
            .list_channels(ListChannelsRequest {
                active_only: false,
                inactive_only: false,
                public_only: false,
                private_only: false,
                peer: vec![],
                peer_alias_lookup: true,
            })
            .await
        {
            Ok(response) => Ok(ListChannelsResponse {
                channels: response
                    .into_inner()
                    .channels
                    .into_iter()
                    .map(|channel| {
                        let channel_size_sats = channel.capacity.try_into().expect("i64 -> u64");

                        let local_balance_sats: u64 =
                            channel.local_balance.try_into().expect("i64 -> u64");
                        let local_channel_reserve_sats: u64 = match channel.local_constraints {
                            Some(constraints) => constraints.chan_reserve_sat,
                            None => 0,
                        };

                        let outbound_liquidity_sats =
                            local_balance_sats.saturating_sub(local_channel_reserve_sats);

                        let remote_balance_sats: u64 =
                            channel.remote_balance.try_into().expect("i64 -> u64");
                        let remote_channel_reserve_sats: u64 = match channel.remote_constraints {
                            Some(constraints) => constraints.chan_reserve_sat,
                            None => 0,
                        };

                        let inbound_liquidity_sats =
                            remote_balance_sats.saturating_sub(remote_channel_reserve_sats);

                        let funding_outpoint = OutPoint::from_str(&channel.channel_point).ok();

                        let remote_address = peer_addresses.get(&channel.remote_pubkey).cloned();

                        let (base_fee_msat, parts_per_million) =
                            match fee_report.get(&channel.chan_id) {
                                Some((base, ppm)) => (Some(*base), Some(*ppm)),
                                None => (None, None),
                            };

                        ChannelInfo {
                            remote_pubkey: PublicKey::from_str(&channel.remote_pubkey)
                                .expect("Lightning node returned invalid remote channel pubkey"),
                            channel_size_sats,
                            outbound_liquidity_sats,
                            inbound_liquidity_sats,
                            is_active: channel.active,
                            funding_outpoint,
                            remote_node_alias: if channel.peer_alias.is_empty() {
                                None
                            } else {
                                Some(channel.peer_alias.clone())
                            },
                            remote_address,
                            base_fee_msat,
                            parts_per_million,
                        }
                    })
                    .collect(),
            }),
            Err(e) => Err(LightningRpcError::FailedToListChannels {
                failure_reason: format!("Failed to list active channels {e:?}"),
            }),
        }
    }

    async fn set_channel_fees(
        &self,
        payload: SetChannelFeesRequest,
    ) -> Result<(), LightningRpcError> {
        let mut client = self.connect().await?;

        // LND's `PolicyUpdateRequest` applies every field it receives, so we
        // need the channel's current `time_lock_delta` (and htlc min/max) to
        // avoid clobbering them when only base + ppm are being changed. To
        // look those up we first resolve the funding outpoint to LND's
        // numeric `chan_id`, then call `get_chan_info`.
        let target = format!(
            "{}:{}",
            payload.funding_outpoint.txid, payload.funding_outpoint.vout
        );
        let channel = client
            .lightning()
            .list_channels(ListChannelsRequest::default())
            .await
            .map_err(|e| LightningRpcError::FailedToSetChannelFees {
                failure_reason: format!("Failed to list channels: {e:?}"),
            })?
            .into_inner()
            .channels
            .into_iter()
            .find(|c| c.channel_point == target)
            .ok_or_else(|| LightningRpcError::FailedToSetChannelFees {
                failure_reason: format!("No channel found with funding outpoint {target}"),
            })?;

        let our_pubkey = client
            .lightning()
            .get_info(GetInfoRequest {})
            .await
            .map_err(|e| LightningRpcError::FailedToSetChannelFees {
                failure_reason: format!("Failed to get node info: {e:?}"),
            })?
            .into_inner()
            .identity_pubkey;

        let edge = client
            .lightning()
            .get_chan_info(ChanInfoRequest {
                chan_id: channel.chan_id,
                ..Default::default()
            })
            .await
            .map_err(|e| LightningRpcError::FailedToSetChannelFees {
                failure_reason: format!("Failed to get channel info: {e:?}"),
            })?
            .into_inner();

        // Pick the policy advertised by our node (the local side); fall back
        // to node1_policy when neither pubkey matches, which only happens if
        // the gossip data has not propagated yet.
        let current_policy = if edge.node1_pub == our_pubkey {
            edge.node1_policy
        } else if edge.node2_pub == our_pubkey {
            edge.node2_policy
        } else {
            edge.node1_policy
        };

        let fee_rate_ppm = u32::try_from(payload.parts_per_million).map_err(|_| {
            LightningRpcError::FailedToSetChannelFees {
                failure_reason: format!(
                    "parts_per_million {} does not fit in u32",
                    payload.parts_per_million,
                ),
            }
        })?;

        let base_fee_msat = i64::try_from(payload.base_fee_msat).map_err(|_| {
            LightningRpcError::FailedToSetChannelFees {
                failure_reason: format!(
                    "base_fee_msat {} does not fit in i64",
                    payload.base_fee_msat,
                ),
            }
        })?;

        // Default time_lock_delta of 40 matches LND's CLI default for
        // `lncli updatechanpolicy` when the channel's existing CLTV is not
        // discoverable. max_htlc_msat == 0 means "no max" in LND.
        let time_lock_delta = current_policy
            .as_ref()
            .map(|p| p.time_lock_delta)
            .unwrap_or(40);
        let max_htlc_msat = current_policy
            .as_ref()
            .map(|p| p.max_htlc_msat)
            .unwrap_or(0);
        let min_htlc_msat = current_policy
            .as_ref()
            .map(|p| p.min_htlc as u64)
            .unwrap_or(0);

        let chan_point = ChannelPoint {
            funding_txid: Some(FundingTxid::FundingTxidBytes(
                <bitcoin::Txid as AsRef<[u8]>>::as_ref(&payload.funding_outpoint.txid).to_vec(),
            )),
            output_index: payload.funding_outpoint.vout,
        };

        let request = PolicyUpdateRequest {
            base_fee_msat,
            fee_rate_ppm,
            time_lock_delta,
            max_htlc_msat,
            min_htlc_msat,
            min_htlc_msat_specified: false,
            scope: Some(PolicyUpdateScope::ChanPoint(chan_point)),
            ..Default::default()
        };

        let response = client
            .lightning()
            .update_channel_policy(request)
            .await
            .map_err(|e| LightningRpcError::FailedToSetChannelFees {
                failure_reason: format!("update_channel_policy failed: {e:?}"),
            })?
            .into_inner();

        if !response.failed_updates.is_empty() {
            let details = response
                .failed_updates
                .iter()
                .map(|f| {
                    let outpoint = f
                        .outpoint
                        .as_ref()
                        .map(|op| format!("{}:{}", op.txid_str, op.output_index))
                        .unwrap_or_else(|| "<unknown outpoint>".to_string());
                    let reason = UpdateFailure::try_from(f.reason)
                        .map(|r| r.as_str_name())
                        .unwrap_or("UPDATE_FAILURE_UNKNOWN");
                    format!("{outpoint}: {reason} ({})", f.update_error)
                })
                .collect::<Vec<_>>()
                .join("; ");
            return Err(LightningRpcError::FailedToSetChannelFees {
                failure_reason: format!("update_channel_policy reported failures: {details}"),
            });
        }

        Ok(())
    }

    async fn get_balances(&self) -> Result<GetBalancesResponse, LightningRpcError> {
        let mut client = self.connect().await?;

        let wallet_balance_response = client
            .lightning()
            .wallet_balance(WalletBalanceRequest {
                ..Default::default()
            })
            .await
            .map_err(|e| LightningRpcError::FailedToGetBalances {
                failure_reason: format!("Failed to get on-chain balance {e:?}"),
            })?
            .into_inner();

        let channel_balance_response = client
            .lightning()
            .channel_balance(ChannelBalanceRequest {})
            .await
            .map_err(|e| LightningRpcError::FailedToGetBalances {
                failure_reason: format!("Failed to get lightning balance {e:?}"),
            })?
            .into_inner();
        let total_outbound = channel_balance_response.local_balance.unwrap_or_default();
        let unsettled_outbound = channel_balance_response
            .unsettled_local_balance
            .unwrap_or_default();
        let pending_outbound = channel_balance_response
            .pending_open_local_balance
            .unwrap_or_default();
        let lightning_balance_msats = total_outbound
            .msat
            .saturating_sub(unsettled_outbound.msat)
            .saturating_sub(pending_outbound.msat);

        let total_inbound = channel_balance_response.remote_balance.unwrap_or_default();
        let unsettled_inbound = channel_balance_response
            .unsettled_remote_balance
            .unwrap_or_default();
        let pending_inbound = channel_balance_response
            .pending_open_remote_balance
            .unwrap_or_default();
        let inbound_lightning_liquidity_msats = total_inbound
            .msat
            .saturating_sub(unsettled_inbound.msat)
            .saturating_sub(pending_inbound.msat);

        Ok(GetBalancesResponse {
            onchain_balance_sats: (wallet_balance_response.total_balance
                + wallet_balance_response.reserved_balance_anchor_chan)
                as u64,
            lightning_balance_msats,
            inbound_lightning_liquidity_msats,
        })
    }

    async fn get_invoice(
        &self,
        get_invoice_request: GetInvoiceRequest,
    ) -> Result<Option<GetInvoiceResponse>, LightningRpcError> {
        let mut client = self.connect().await?;
        let invoice = client
            .invoices()
            .lookup_invoice_v2(LookupInvoiceMsg {
                invoice_ref: Some(InvoiceRef::PaymentHash(
                    get_invoice_request.payment_hash.consensus_encode_to_vec(),
                )),
                ..Default::default()
            })
            .await;
        let invoice = match invoice {
            Ok(invoice) => invoice.into_inner(),
            Err(_) => return Ok(None),
        };
        let status = match &invoice.state() {
            InvoiceState::Settled => fedimint_gateway_common::PaymentStatus::Succeeded,
            InvoiceState::Canceled => fedimint_gateway_common::PaymentStatus::Failed,
            _ => fedimint_gateway_common::PaymentStatus::Pending,
        };

        Ok(Some(GetInvoiceResponse {
            preimage: invoice_preimage_hex(&invoice),
            payment_hash: Some(
                sha256::Hash::from_slice(&invoice.r_hash).expect("Could not convert payment hash"),
            ),
            amount: Amount::from_msats(invoice.value_msat as u64),
            created_at: UNIX_EPOCH + Duration::from_secs(invoice.creation_date as u64),
            status,
        }))
    }

    async fn list_transactions(
        &self,
        start_secs: u64,
        end_secs: u64,
    ) -> Result<ListTransactionsResponse, LightningRpcError> {
        let mut client = self.connect().await?;
        let payments = client
            .lightning()
            .list_payments(ListPaymentsRequest {
                // On higher versions on LND, we can filter on the time range directly in the query
                ..Default::default()
            })
            .await
            .map_err(|err| LightningRpcError::FailedToListTransactions {
                failure_reason: err.to_string(),
            })?
            .into_inner();

        let mut payments = payments
            .payments
            .iter()
            .filter_map(|payment| {
                let timestamp_secs = (payment.creation_time_ns / 1_000_000_000) as u64;
                if timestamp_secs < start_secs || timestamp_secs >= end_secs {
                    return None;
                }
                let payment_hash = sha256::Hash::from_str(&payment.payment_hash).ok();
                let preimage = (!payment.payment_preimage.is_empty())
                    .then_some(payment.payment_preimage.clone());
                let status = match &payment.status() {
                    PaymentStatus::Succeeded => fedimint_gateway_common::PaymentStatus::Succeeded,
                    PaymentStatus::Failed => fedimint_gateway_common::PaymentStatus::Failed,
                    _ => fedimint_gateway_common::PaymentStatus::Pending,
                };
                Some(PaymentDetails {
                    payment_hash,
                    preimage,
                    payment_kind: PaymentKind::Bolt11,
                    amount: Amount::from_msats(payment.value_msat as u64),
                    direction: PaymentDirection::Outbound,
                    status,
                    timestamp_secs,
                })
            })
            .collect::<Vec<_>>();

        let invoices = client
            .lightning()
            .list_invoices(ListInvoiceRequest {
                pending_only: false,
                // On higher versions on LND, we can filter on the time range directly in the query
                ..Default::default()
            })
            .await
            .map_err(|err| LightningRpcError::FailedToListTransactions {
                failure_reason: err.to_string(),
            })?
            .into_inner();

        let mut incoming_payments = invoices
            .invoices
            .iter()
            .filter_map(|invoice| {
                let timestamp_secs = invoice.settle_date as u64;
                if timestamp_secs < start_secs || timestamp_secs >= end_secs {
                    return None;
                }
                let status = match &invoice.state() {
                    InvoiceState::Settled => fedimint_gateway_common::PaymentStatus::Succeeded,
                    InvoiceState::Canceled => fedimint_gateway_common::PaymentStatus::Failed,
                    _ => return None,
                };
                let preimage = (!invoice.r_preimage.is_empty())
                    .then_some(invoice.r_preimage.encode_hex::<String>());
                Some(PaymentDetails {
                    payment_hash: Some(
                        sha256::Hash::from_slice(&invoice.r_hash)
                            .expect("Could not convert payment hash"),
                    ),
                    preimage,
                    payment_kind: PaymentKind::Bolt11,
                    amount: Amount::from_msats(invoice.value_msat as u64),
                    direction: PaymentDirection::Inbound,
                    status,
                    timestamp_secs,
                })
            })
            .collect::<Vec<_>>();

        payments.append(&mut incoming_payments);
        payments.sort_by_key(|p| p.timestamp_secs);

        Ok(ListTransactionsResponse {
            transactions: payments,
        })
    }

    fn create_offer(
        &self,
        _amount_msat: Option<Amount>,
        _description: Option<String>,
        _expiry_secs: Option<u32>,
        _quantity: Option<u64>,
    ) -> Result<String, LightningRpcError> {
        Err(LightningRpcError::Bolt12Error {
            failure_reason: "LND Does not support Bolt12".to_string(),
        })
    }

    async fn pay_offer(
        &self,
        _offer: String,
        _quantity: Option<u64>,
        _amount: Option<Amount>,
        _payer_note: Option<String>,
    ) -> Result<Preimage, LightningRpcError> {
        Err(LightningRpcError::Bolt12Error {
            failure_reason: "LND Does not support Bolt12".to_string(),
        })
    }

    fn sync_wallet(&self) -> Result<(), LightningRpcError> {
        // There is nothing explicit needed to do for syncing an LND node
        Ok(())
    }
}

fn route_hints_to_lnd(
    route_hints: &[fedimint_ln_common::route_hints::RouteHint],
) -> Vec<tonic_lnd::lnrpc::RouteHint> {
    route_hints
        .iter()
        .map(|hint| tonic_lnd::lnrpc::RouteHint {
            hop_hints: hint
                .0
                .iter()
                .map(|hop| tonic_lnd::lnrpc::HopHint {
                    node_id: hop.src_node_id.serialize().encode_hex(),
                    chan_id: hop.short_channel_id,
                    fee_base_msat: hop.base_msat,
                    fee_proportional_millionths: hop.proportional_millionths,
                    cltv_expiry_delta: u32::from(hop.cltv_expiry_delta),
                })
                .collect(),
        })
        .collect()
}

fn wire_features_to_lnd_feature_vec(
    features_wire_encoded: &[u8],
) -> Result<Vec<i32>, FeatureVectorTooLargeError> {
    if features_wire_encoded.len() > 1_000 {
        return Err(FeatureVectorTooLargeError);
    }

    let lnd_features = features_wire_encoded
        .iter()
        .rev()
        .enumerate()
        .flat_map(|(byte_idx, &feature_byte)| {
            (0..8).filter_map(move |bit_idx| {
                if (feature_byte & (1u8 << bit_idx)) != 0 {
                    Some(
                        i32::try_from(byte_idx * 8 + bit_idx)
                            .expect("Index will never exceed i32::MAX for feature vectors <8MB"),
                    )
                } else {
                    None
                }
            })
        })
        .collect::<Vec<_>>();

    Ok(lnd_features)
}

/// Utility struct for logging payment hashes. Useful for debugging.
struct PrettyPaymentHash<'a>(&'a Vec<u8>);

impl Display for PrettyPaymentHash<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "payment_hash={}", self.0.encode_hex::<String>())
    }
}

/// An invoice's destination feature bits are too long to convert for LND.
#[derive(Debug, thiserror::Error)]
#[error("Will not process feature bit vectors larger than 1000 byte")]
struct FeatureVectorTooLargeError;

#[cfg(test)]
mod tests;