breez_sdk_core/greenlight/
node_api.rs

1use std::cmp::{min, Reverse};
2use std::collections::{HashMap, HashSet};
3use std::iter::Iterator;
4use std::pin::Pin;
5use std::str::FromStr;
6use std::sync::atomic::{AtomicU16, Ordering};
7use std::sync::Arc;
8use std::time::{Duration, SystemTime, UNIX_EPOCH};
9
10use anyhow::{anyhow, Result};
11use ecies::symmetric::{sym_decrypt, sym_encrypt};
12use futures::{Future, Stream};
13use gl_client::credentials::{Device, Nobody};
14use gl_client::node;
15use gl_client::node::ClnClient;
16use gl_client::pb::cln::delinvoice_request::DelinvoiceStatus;
17use gl_client::pb::cln::listinvoices_invoices::ListinvoicesInvoicesStatus;
18use gl_client::pb::cln::listinvoices_request::ListinvoicesIndex;
19use gl_client::pb::cln::listpays_pays::ListpaysPaysStatus;
20use gl_client::pb::cln::listpeerchannels_channels::ListpeerchannelsChannelsState::*;
21use gl_client::pb::cln::listsendpays_request::ListsendpaysIndex;
22use gl_client::pb::cln::{
23    self, Amount, DelinvoiceRequest, GetrouteRequest, GetrouteRoute, ListchannelsRequest,
24    ListclosedchannelsClosedchannels, ListpaysPays, ListpeerchannelsChannels, ListsendpaysPayments,
25    PreapproveinvoiceRequest, SendpayRequest, SendpayRoute, WaitsendpayRequest,
26};
27use gl_client::pb::{incoming_payment, TrampolinePayRequest};
28use gl_client::scheduler::Scheduler;
29use gl_client::signer::model::greenlight::{amount, scheduler};
30use gl_client::signer::Signer;
31use sdk_common::prelude::*;
32use serde::{Deserialize, Serialize};
33use serde_json::{json, Map, Value};
34use strum_macros::{Display, EnumString};
35use tokio::join;
36use tokio::sync::{mpsc, watch, Mutex};
37use tokio::time::{sleep, Instant, MissedTickBehavior};
38use tokio_stream::StreamExt;
39
40use crate::bitcoin::bech32::{u5, ToBase32};
41use crate::bitcoin::blockdata::constants::WITNESS_SCALE_FACTOR;
42use crate::bitcoin::hashes::Hash;
43use crate::bitcoin::secp256k1::ecdsa::{RecoverableSignature, RecoveryId};
44use crate::bitcoin::secp256k1::PublicKey;
45use crate::bitcoin::secp256k1::Secp256k1;
46use crate::bitcoin::util::bip32::{ChildNumber, ExtendedPrivKey};
47use crate::bitcoin::{
48    Address, OutPoint, Script, Sequence, Transaction, TxIn, TxOut, Txid, Witness,
49};
50use crate::lightning::util::message_signing::verify;
51use crate::lightning_invoice::{RawBolt11Invoice, SignedRawBolt11Invoice};
52use crate::node_api::{
53    CreateInvoiceRequest, FetchBolt11Result, IncomingPayment, NodeAPI, NodeError, NodeResult,
54};
55use crate::persist::cache::NodeStateStorage;
56use crate::persist::db::SqliteStorage;
57use crate::persist::send_pays::{SendPay, SendPayStatus};
58use crate::{models::*, LspInformation};
59use crate::{NodeConfig, PrepareRedeemOnchainFundsRequest, PrepareRedeemOnchainFundsResponse};
60
61const KEY_GL_CREDENTIALS: &str = "gl_credentials";
62const MAX_PAYMENT_AMOUNT_MSAT: u64 = 4294967000;
63const MAX_INBOUND_LIQUIDITY_MSAT: u64 = 4000000000;
64const TRAMPOLINE_BASE_FEE_MSAT: u64 = 4000;
65const TRAMPOLINE_FEE_PPM: u64 = 5000;
66const PAYMENT_STATE_PENDING: u8 = 1;
67const PAYMENT_STATE_COMPLETE: u8 = 2;
68const PAYMENT_STATE_FAILED: u8 = 4;
69
70pub(crate) struct Greenlight {
71    sdk_config: Config,
72    signer: Mutex<Arc<Signer>>,
73    device: Device,
74    gl_client: Mutex<Option<node::Client>>,
75    node_client: Mutex<Option<ClnClient>>,
76    persister: Arc<SqliteStorage>,
77    inprogress_payments: AtomicU16,
78}
79
80#[derive(Serialize, Deserialize)]
81struct InvoiceLabel {
82    pub unix_milli: u128,
83    pub payer_amount_msat: Option<u64>,
84}
85
86#[derive(Serialize, Deserialize)]
87struct PaymentLabel {
88    pub unix_nano: u128,
89    pub trampoline: bool,
90    pub client_label: Option<String>,
91    pub amount_msat: u64,
92}
93
94impl Greenlight {
95    /// Connects to a live node using the provided seed and config.
96    /// If the node is not registered, it will try to recover it using the seed.
97    /// If the node is not created, it will register it using the provided partner credentials
98    /// or invite code
99    /// If the node is already registered and an existing credentials were found, it will try to
100    /// connect to the node using these credentials.
101    pub async fn connect(
102        config: Config,
103        seed: Vec<u8>,
104        restore_only: Option<bool>,
105        persister: Arc<SqliteStorage>,
106    ) -> NodeResult<Self> {
107        // Derive the encryption key from the seed
108        let temp_signer = Arc::new(Signer::new(
109            seed.clone(),
110            config.network.into(),
111            Nobody::new(),
112        )?);
113        let encryption_key = Self::derive_bip32_key(
114            config.network,
115            &temp_signer,
116            vec![ChildNumber::from_hardened_idx(140)?, ChildNumber::from(0)],
117        )?
118        .to_priv()
119        .to_bytes();
120        let encryption_key_slice = encryption_key.as_slice();
121
122        let register_credentials = match config.node_config.clone() {
123            NodeConfig::Greenlight { config } => config,
124        };
125
126        // Query for the existing credentials
127        let mut parsed_credentials =
128            Self::get_node_credentials(config.network, &temp_signer, persister.clone())?
129                .ok_or(NodeError::credentials("No credentials found"));
130        if parsed_credentials.is_err() {
131            info!("No credentials found, trying to recover existing node");
132            parsed_credentials = match Self::recover(config.network, seed.clone()).await {
133                Ok(creds) => Ok(creds),
134                Err(_) => {
135                    match restore_only.unwrap_or(false) {
136                        false => {
137                            // If we got here it means we failed to recover so we need to register a new node
138                            info!("Failed to recover node, registering new one");
139                            let credentials = Self::register(
140                                config.clone().network,
141                                seed.clone(),
142                                register_credentials.partner_credentials,
143                                register_credentials.invite_code,
144                            )
145                            .await?;
146                            Ok(credentials)
147                        }
148                        true => {
149                            return Err(NodeError::RestoreOnly("Node does not exist".to_string()));
150                        }
151                    }
152                }
153            }
154        }
155
156        // Persist the connection credentials for future use and return the node instance
157        match parsed_credentials {
158            Ok(creds) => {
159                let temp_scheduler = Scheduler::new(config.network.into(), creds.clone()).await?;
160                debug!("upgrading credentials");
161                let creds = creds.upgrade(&temp_scheduler, &temp_signer).await?;
162                debug!("upgrading credentials succeeded");
163                let encrypted_creds = sym_encrypt(encryption_key_slice, &creds.to_bytes());
164                match encrypted_creds {
165                    Some(c) => {
166                        persister.update_cached_item(KEY_GL_CREDENTIALS, hex::encode(c))?;
167                        Greenlight::new(config, seed, creds.clone(), persister)
168                    }
169                    None => Err(NodeError::generic("Failed to encrypt credentials")),
170                }
171            }
172            Err(_) => Err(NodeError::credentials("Failed to get gl credentials")),
173        }
174    }
175
176    fn new(
177        sdk_config: Config,
178        seed: Vec<u8>,
179        device: Device,
180        persister: Arc<SqliteStorage>,
181    ) -> NodeResult<Greenlight> {
182        let greenlight_network = sdk_config.network.into();
183        let signer = Signer::new(seed.clone(), greenlight_network, device.clone())?;
184        Ok(Greenlight {
185            sdk_config,
186            signer: Mutex::new(Arc::new(signer)),
187            device,
188            gl_client: Mutex::new(None),
189            node_client: Mutex::new(None),
190            persister,
191            inprogress_payments: AtomicU16::new(0),
192        })
193    }
194
195    async fn get_signer(&self) -> Arc<Signer> {
196        Arc::clone(&*self.signer.lock().await)
197    }
198
199    fn derive_bip32_key(
200        network: Network,
201        signer: &Arc<Signer>,
202        path: Vec<ChildNumber>,
203    ) -> NodeResult<ExtendedPrivKey> {
204        Ok(
205            ExtendedPrivKey::new_master(network.into(), &signer.bip32_ext_key())?
206                .derive_priv(&Secp256k1::new(), &path)?,
207        )
208    }
209
210    fn legacy_derive_bip32_key(
211        network: Network,
212        signer: &Arc<Signer>,
213        path: Vec<ChildNumber>,
214    ) -> NodeResult<ExtendedPrivKey> {
215        Ok(
216            ExtendedPrivKey::new_master(network.into(), &signer.legacy_bip32_ext_key())?
217                .derive_priv(&Secp256k1::new(), &path)?,
218        )
219    }
220
221    async fn register(
222        network: Network,
223        seed: Vec<u8>,
224        register_credentials: Option<GreenlightCredentials>,
225        invite_code: Option<String>,
226    ) -> Result<Device> {
227        if invite_code.is_some() && register_credentials.is_some() {
228            return Err(anyhow!("Cannot specify both invite code and credentials"));
229        }
230        let greenlight_network = network.into();
231        let creds = match register_credentials {
232            Some(creds) => {
233                debug!("registering with credentials");
234                Nobody {
235                    cert: creds.developer_cert,
236                    key: creds.developer_key,
237                    ..Default::default()
238                }
239            }
240            None => Nobody::new(),
241        };
242
243        let signer = Signer::new(seed, greenlight_network, creds.clone())?;
244        let scheduler = Scheduler::new(greenlight_network, creds).await?;
245
246        let register_res: scheduler::RegistrationResponse =
247            scheduler.register(&signer, invite_code).await?;
248
249        Ok(Device::from_bytes(register_res.creds))
250    }
251
252    async fn recover(network: Network, seed: Vec<u8>) -> Result<Device> {
253        let greenlight_network = network.into();
254        let credentials = Nobody::new();
255        let signer = Signer::new(seed, greenlight_network, credentials.clone())?;
256        let scheduler = Scheduler::new(greenlight_network, credentials).await?;
257        let recover_res: scheduler::RecoveryResponse = scheduler.recover(&signer).await?;
258
259        Ok(Device::from_bytes(recover_res.creds))
260    }
261
262    async fn get_client(&self) -> NodeResult<node::Client> {
263        let mut gl_client = self.gl_client.lock().await;
264        if gl_client.is_none() {
265            let scheduler = Scheduler::new(self.sdk_config.network.into(), self.device.clone())
266                .await
267                .map_err(|e| NodeError::ServiceConnectivity(e.to_string()))?;
268            *gl_client = Some(scheduler.node().await?);
269        }
270        Ok(gl_client.clone().unwrap())
271    }
272
273    pub(crate) async fn get_node_client(&self) -> NodeResult<node::ClnClient> {
274        let mut node_client = self.node_client.lock().await;
275        if node_client.is_none() {
276            let scheduler = Scheduler::new(self.sdk_config.network.into(), self.device.clone())
277                .await
278                .map_err(|e| NodeError::ServiceConnectivity(e.to_string()))?;
279            *node_client = Some(scheduler.node().await?);
280        }
281        Ok(node_client.clone().unwrap())
282    }
283
284    fn get_node_credentials(
285        network: Network,
286        signer: &Arc<Signer>,
287        persister: Arc<SqliteStorage>,
288    ) -> NodeResult<Option<Device>> {
289        // Derive the encryption key from the seed
290        let encryption_key = Self::derive_bip32_key(
291            network,
292            signer,
293            vec![ChildNumber::from_hardened_idx(140)?, ChildNumber::from(0)],
294        )?
295        .to_priv()
296        .to_bytes();
297        let encryption_key_slice = encryption_key.as_slice();
298
299        let legacy_encryption_key = Self::legacy_derive_bip32_key(
300            network,
301            signer,
302            vec![ChildNumber::from_hardened_idx(140)?, ChildNumber::from(0)],
303        )?
304        .to_priv()
305        .to_bytes();
306        let legacy_encryption_key_slice = legacy_encryption_key.as_slice();
307
308        match persister.get_cached_item(KEY_GL_CREDENTIALS)? {
309            Some(encrypted_creds) => {
310                let encrypted_creds = hex::decode(encrypted_creds)?;
311                let mut decrypted_credentials =
312                    sym_decrypt(encryption_key_slice, encrypted_creds.as_slice());
313                if decrypted_credentials.is_none() {
314                    info!("Failed to decrypt credentials, trying legacy key");
315                    decrypted_credentials =
316                        sym_decrypt(legacy_encryption_key_slice, encrypted_creds.as_slice());
317                }
318                match decrypted_credentials {
319                    Some(decrypted_creds) => {
320                        let credentials = Device::from_bytes(decrypted_creds.as_slice());
321                        if credentials.cert.is_empty() {
322                            Err(NodeError::credentials("Unable to parse credentials"))
323                        } else {
324                            Ok(Some(credentials))
325                        }
326                    }
327                    None => Err(NodeError::credentials(
328                        "Failed to decrypt credentials, seed doesn't match existing node",
329                    )),
330                }
331            }
332            None => Ok(None),
333        }
334    }
335
336    async fn fetch_outgoing_payment_with_retry(
337        client: node::ClnClient,
338        payment_hash: Vec<u8>,
339    ) -> Result<cln::ListpaysPays> {
340        let mut response = cln::ListpaysResponse::default();
341        let mut retry = 0;
342        let max_retries = 20;
343
344        while response.pays.is_empty() && retry < max_retries {
345            let req = cln::ListpaysRequest {
346                payment_hash: Some(payment_hash.clone()),
347                status: Some(cln::listpays_request::ListpaysStatus::Complete.into()),
348                ..cln::ListpaysRequest::default()
349            };
350            let mut client = client.clone();
351            response = with_connection_retry!(client.list_pays(req.clone()))
352                .await?
353                .into_inner();
354            if response.pays.is_empty() {
355                debug!("fetch outgoing payment failed, retrying in 100ms...");
356                sleep(Duration::from_millis(100)).await;
357            }
358            retry += 1;
359        }
360
361        // CLN also returns failed ListpaysPays along with the complete one
362        debug!("list_pays: {:?}", response.pays);
363        let pays: Vec<ListpaysPays> = response
364            .pays
365            .into_iter()
366            .filter(|pay| pay.status() == cln::listpays_pays::ListpaysPaysStatus::Complete)
367            .collect();
368
369        if pays.is_empty() {
370            return Err(anyhow!("Payment not found"));
371        }
372        Ok(pays[0].clone())
373    }
374
375    async fn fetch_channels_and_balance_with_retry(
376        cln_client: node::ClnClient,
377        persister: Arc<SqliteStorage>,
378        match_local_balance: bool,
379    ) -> NodeResult<(
380        Vec<cln::ListpeerchannelsChannels>,
381        Vec<cln::ListpeerchannelsChannels>,
382        Vec<String>,
383        u64,
384    )> {
385        let (mut all_channels, mut opened_channels, mut connected_peers, mut channels_balance) =
386            Greenlight::fetch_channels_and_balance(cln_client.clone()).await?;
387        if match_local_balance {
388            let node_state = persister.get_node_state()?;
389            if let Some(state) = node_state {
390                let mut retry_count = 0;
391                while state.channels_balance_msat != channels_balance && retry_count < 10 {
392                    warn!("balance matching local state is required and not yet satisfied, retrying in 100ms...");
393                    sleep(Duration::from_millis(100)).await;
394                    (
395                        all_channels,
396                        opened_channels,
397                        connected_peers,
398                        channels_balance,
399                    ) = Greenlight::fetch_channels_and_balance(cln_client.clone()).await?;
400                    retry_count += 1;
401                }
402            }
403        }
404        Ok((
405            all_channels,
406            opened_channels,
407            connected_peers,
408            channels_balance,
409        ))
410    }
411
412    async fn fetch_channels_and_balance(
413        mut client: node::ClnClient,
414    ) -> NodeResult<(
415        Vec<cln::ListpeerchannelsChannels>,
416        Vec<cln::ListpeerchannelsChannels>,
417        Vec<String>,
418        u64,
419    )> {
420        // list all channels
421        let req = cln::ListpeerchannelsRequest::default();
422        let peerchannels = with_connection_retry!(client.list_peer_channels(req.clone()))
423            .await?
424            .into_inner();
425
426        // filter only connected peers
427        let connected_peers: Vec<String> = peerchannels
428            .channels
429            .iter()
430            .filter(|channel| channel.peer_connected())
431            .filter_map(|channel| channel.peer_id.clone())
432            .map(hex::encode)
433            .collect::<HashSet<_>>()
434            .into_iter()
435            .collect();
436
437        // filter only opened channels
438        let opened_channels: Vec<cln::ListpeerchannelsChannels> = peerchannels
439            .channels
440            .iter()
441            .filter(|c| c.state() == ChanneldNormal)
442            .cloned()
443            .collect();
444
445        // calculate channels balance only from opened channels
446        let channels_balance = opened_channels
447            .iter()
448            .map(|c| Channel::from(c.clone()))
449            .map(|c| c.spendable_msat)
450            .sum::<u64>();
451        Ok((
452            peerchannels.channels,
453            opened_channels,
454            connected_peers,
455            channels_balance,
456        ))
457    }
458
459    async fn list_funds(&self) -> Result<cln::ListfundsResponse> {
460        let mut client = self.get_node_client().await?;
461        let req = cln::ListfundsRequest::default();
462        let funds: cln::ListfundsResponse = with_connection_retry!(client.list_funds(req.clone()))
463            .await?
464            .into_inner();
465        Ok(funds)
466    }
467
468    async fn on_chain_balance(&self, funds: &cln::ListfundsResponse) -> Result<u64> {
469        let on_chain_balance = funds.outputs.iter().fold(0, |a, b| {
470            if b.reserved {
471                return a;
472            }
473            a + b.amount_msat.clone().unwrap_or_default().msat
474        });
475        Ok(on_chain_balance)
476    }
477
478    async fn pending_onchain_balance(
479        &self,
480        peer_channels: &[cln::ListpeerchannelsChannels],
481    ) -> Result<u64> {
482        let pending_onchain_balance = peer_channels.iter().fold(0, |a, b| match b.state() {
483            ChanneldShuttingDown | ClosingdSigexchange | ClosingdComplete | AwaitingUnilateral
484            | FundingSpendSeen => a + b.to_us_msat.clone().unwrap_or_default().msat,
485
486            // When we  unilaterally close the channel it will get status as `AwaitingUnilateral`
487            // first, but when the closing transaction is confirmed onchain the funds receive status
488            // as `Onchain`. Though if we closed the channel we'll have to wait for the timelock to
489            // pass before the funds can be spent.
490            Onchain => {
491                if b.closer() == cln::ChannelSide::Local
492                    && b.status
493                        .last()
494                        .is_some_and(|status| status.contains("DELAYED_OUTPUT_TO_US"))
495                {
496                    a + b.to_us_msat.clone().unwrap_or_default().msat
497                } else {
498                    a
499                }
500            }
501            _ => a,
502        });
503        info!("pending_onchain_balance is {pending_onchain_balance}");
504        Ok(pending_onchain_balance)
505    }
506
507    // Collect utxos from onchain funds
508    async fn utxos(&self, funds: cln::ListfundsResponse) -> Result<Vec<UnspentTransactionOutput>> {
509        let utxos: Vec<UnspentTransactionOutput> = funds
510            .outputs
511            .iter()
512            .map(|output| UnspentTransactionOutput {
513                txid: output.txid.clone(),
514                outnum: output.output,
515                amount_millisatoshi: output
516                    .amount_msat
517                    .as_ref()
518                    .map(|a| a.msat)
519                    .unwrap_or_default(),
520                address: output.address.clone().unwrap_or_default(),
521                reserved: output.reserved,
522            })
523            .collect();
524        Ok(utxos)
525    }
526
527    async fn build_payment_path(
528        &self,
529        route: &Vec<GetrouteRoute>,
530        first_edge: PaymentPathEdge,
531    ) -> NodeResult<PaymentPath> {
532        let client = self.get_node_client().await?;
533        let mut hops = vec![first_edge];
534
535        for hop in route {
536            let req = ListchannelsRequest {
537                short_channel_id: Some(hop.channel.clone()),
538                source: None,
539                destination: None,
540            };
541            let mut client = client.clone();
542            let hopchannels = with_connection_retry!(client.list_channels(req.clone()))
543                .await?
544                .into_inner()
545                .channels;
546
547            let first_channel = hopchannels.first().ok_or(NodeError::RouteNotFound(format!(
548                "Channel not found {}",
549                hop.channel.clone()
550            )))?;
551
552            info!("found channel in route: {first_channel:?}");
553            hops.push(PaymentPathEdge {
554                base_fee_msat: first_channel.base_fee_millisatoshi as u64,
555                fee_per_millionth: first_channel.fee_per_millionth as u64,
556                node_id: hop.id.clone(),
557                short_channel_id: hop.channel.clone(),
558                channel_delay: first_channel.delay as u64,
559            });
560        }
561        Ok(PaymentPath { edges: hops })
562    }
563
564    async fn max_sendable_amount_from_peer(
565        &self,
566        via_peer_id: Vec<u8>,
567        via_peer_channels: Vec<ListpeerchannelsChannels>,
568        payee_node_id: Option<Vec<u8>>,
569        max_hops: u32,
570        last_hop_hint: Option<&RouteHintHop>,
571    ) -> NodeResult<Vec<MaxChannelAmount>> {
572        let mut client = self.get_node_client().await?;
573
574        // Consider the hints as part of the route. If there is a routing hint we will
575        // attempt to calculate the path until the last hop in the hint and then add
576        // the last hop to the path.
577        let (last_node, max_hops) = match last_hop_hint {
578            Some(hop) => (hex::decode(&hop.src_node_id)?, max_hops - 1),
579            None => match payee_node_id.clone() {
580                Some(node_id) => (node_id, max_hops),
581                None => {
582                    return Err(NodeError::RouteNotFound(
583                        "No payee node id or last hop hints provided, cannot calculate max amount"
584                            .to_string(),
585                    ));
586                }
587            },
588        };
589
590        // fetch a route from greenlight
591        info!(
592            "calling get_route for peer {} to node {}, max_hops: {}",
593            hex::encode(via_peer_id.clone()),
594            hex::encode(last_node.clone()),
595            max_hops - 1
596        );
597        let req = GetrouteRequest {
598            id: last_node.clone(),
599            amount_msat: Some(Amount { msat: 0 }),
600            riskfactor: 0,
601            cltv: None,
602            fromid: Some(via_peer_id.clone()),
603            fuzzpercent: Some(0),
604            exclude: vec![],
605            // we deduct the first hop that we calculate manually
606            maxhops: Some(max_hops - 1),
607        };
608        let route_result = with_connection_retry!(client.get_route(req.clone())).await;
609
610        // In case we have no route better to return no amounts for this peer's channels.
611        if let Err(e) = route_result {
612            error!(
613                "Failed to get route for peer {}: {}",
614                hex::encode(via_peer_id.clone()),
615                e
616            );
617            return Ok(vec![]);
618        }
619
620        let route_response = route_result?.into_inner();
621        info!(
622            "max_sendable_amount: route response = {:?}",
623            route_response
624                .route
625                .iter()
626                .map(|r| format!(
627                    "{{node_id: {}, channel: {}}}",
628                    hex::encode(&r.id),
629                    r.channel
630                ))
631                .collect::<Vec<_>>()
632        );
633
634        // We fetch the opened channels so can calculate max amount to send for each channel
635        let opened_channels: Vec<cln::ListpeerchannelsChannels> = via_peer_channels
636            .iter()
637            .filter(|c| c.state() == ChanneldNormal)
638            .cloned()
639            .collect();
640
641        let mut max_per_channel = vec![];
642        for c in opened_channels {
643            let chan_id = c
644                .clone()
645                .channel_id
646                .ok_or(NodeError::generic("Empty channel id"))?;
647
648            // First hop is forwarding so no fees and delays.
649            let first_edge = PaymentPathEdge {
650                base_fee_msat: 0,
651                fee_per_millionth: 0,
652                node_id: via_peer_id.clone(),
653                short_channel_id: c.clone().short_channel_id.unwrap_or_default(),
654                channel_delay: 0,
655            };
656
657            // convert the route to a payment path so we can calculate the amount to forward for each hop
658            let mut payment_path = self
659                .build_payment_path(&route_response.route, first_edge)
660                .await?;
661
662            // Add the last hop hints (if any) to the route
663            if let Some(hint) = last_hop_hint {
664                payment_path.edges.extend(vec![PaymentPathEdge {
665                    base_fee_msat: hint.fees_base_msat as u64,
666                    fee_per_millionth: hint.fees_proportional_millionths as u64,
667                    node_id: payee_node_id.clone().unwrap_or_default(),
668                    short_channel_id: hint.short_channel_id.clone(),
669                    channel_delay: hint.cltv_expiry_delta,
670                }])
671            }
672
673            info!(
674                "max_sendable_amount: route_hops = {:?}",
675                payment_path
676                    .edges
677                    .iter()
678                    .map(|e| format!(
679                        "{{node_id: {}, channel: {}}}",
680                        hex::encode(&e.node_id),
681                        e.short_channel_id
682                    ))
683                    .collect::<Vec<_>>()
684            );
685
686            // go over each hop and calculate the amount to forward.
687            let max_payment_amount =
688                payment_path.final_hop_amount(c.clone().spendable_msat.unwrap_or_default().msat);
689            max_per_channel.push(MaxChannelAmount {
690                channel_id: hex::encode(chan_id),
691                amount_msat: max_payment_amount,
692                path: payment_path,
693            });
694        }
695
696        Ok(max_per_channel)
697    }
698
699    /// Get open peer channels (private and public) as raw protobuf structs, indexed by peer pubkey
700    async fn get_open_peer_channels_pb(
701        &self,
702    ) -> NodeResult<HashMap<Vec<u8>, cln::ListpeerchannelsChannels>> {
703        let mut client = self.get_node_client().await?;
704        // Get the peer channels
705        let req = cln::ListpeerchannelsRequest::default();
706        let peer_channels = with_connection_retry!(client.list_peer_channels(req.clone()))
707            .await?
708            .into_inner();
709
710        let open_peer_channels: HashMap<Vec<u8>, cln::ListpeerchannelsChannels> = peer_channels
711            .channels
712            .into_iter()
713            .filter(|c| {
714                c.state == Some(cln::ChannelState::ChanneldNormal as i32) && c.peer_id.is_some()
715            })
716            .map(|c| (c.peer_id.clone().unwrap(), c))
717            .collect();
718        Ok(open_peer_channels)
719    }
720
721    async fn with_keep_alive<T, F>(&self, f: F) -> T
722    where
723        F: Future<Output = T>,
724    {
725        _ = self.inprogress_payments.fetch_add(1, Ordering::Relaxed);
726        let res = f.await;
727        _ = self.inprogress_payments.fetch_sub(1, Ordering::Relaxed);
728        res
729    }
730
731    // pulls transactions from greenlight based on last sync timestamp.
732    // greenlight gives us the payments via API and for received payments we are looking for settled invoices.
733    async fn pull_transactions(
734        &self,
735        sync_state: &SyncState,
736        htlc_list: Vec<Htlc>,
737    ) -> NodeResult<(SyncState, Vec<Payment>)> {
738        let (receive_payments_res, send_payments_res) = join!(
739            self.pull_receive_payments(&sync_state.list_invoices_index),
740            self.pull_send_payments(&sync_state.send_pays_index, htlc_list),
741        );
742
743        let (receive_payments, list_invoices_index) = receive_payments_res?;
744        let (send_payments, send_pays_index) = send_payments_res?;
745        let mut new_state = sync_state.clone();
746        new_state.list_invoices_index = list_invoices_index;
747        new_state.send_pays_index = send_pays_index;
748
749        let mut payments: Vec<Payment> = Vec::new();
750        payments.extend(receive_payments);
751        payments.extend(send_payments);
752
753        Ok((new_state, payments))
754    }
755
756    async fn pull_receive_payments(
757        &self,
758        state: &SyncIndex,
759    ) -> NodeResult<(Vec<Payment>, SyncIndex)> {
760        let mut client = self.get_node_client().await?;
761
762        let req = cln::ListinvoicesRequest {
763            index: Some(ListinvoicesIndex::Created.into()),
764            start: Some(state.created),
765            ..Default::default()
766        };
767        let mut clone = client.clone();
768        let created_invoices = with_connection_retry!(clone.list_invoices(req.clone()))
769            .await?
770            .into_inner();
771        let req = cln::ListinvoicesRequest {
772            index: Some(ListinvoicesIndex::Updated.into()),
773            start: Some(state.updated),
774            ..Default::default()
775        };
776
777        let updated_invoices = with_connection_retry!(client.list_invoices(req.clone()))
778            .await?
779            .into_inner();
780        let mut new_state = state.clone();
781        if let Some(last) = created_invoices.invoices.last() {
782            new_state.created = last.created_index()
783        }
784        if let Some(last) = updated_invoices.invoices.last() {
785            new_state.updated = last.updated_index()
786        }
787
788        let received_payments: NodeResult<Vec<Payment>> = created_invoices
789            .invoices
790            .into_iter()
791            .chain(updated_invoices.invoices.into_iter())
792            .filter(|i| i.status() == ListinvoicesInvoicesStatus::Paid)
793            .map(TryInto::try_into)
794            .collect();
795
796        Ok((received_payments?, new_state))
797    }
798
799    async fn pull_send_payments(
800        &self,
801        state: &SyncIndex,
802        htlc_list: Vec<Htlc>,
803    ) -> NodeResult<(Vec<Payment>, SyncIndex)> {
804        let mut client = self.get_node_client().await?;
805        let req = cln::ListsendpaysRequest {
806            index: Some(ListsendpaysIndex::Created.into()),
807            start: Some(state.created),
808            ..Default::default()
809        };
810        let mut clone = client.clone();
811        let created_send_pays = with_connection_retry!(clone.list_send_pays(req.clone()))
812            .await?
813            .into_inner();
814        let req = cln::ListsendpaysRequest {
815            index: Some(ListsendpaysIndex::Updated.into()),
816            start: Some(state.updated),
817            ..Default::default()
818        };
819        let updated_send_pays = with_connection_retry!(client.list_send_pays(req.clone()))
820            .await?
821            .into_inner();
822
823        let mut new_state = state.clone();
824        if let Some(last) = created_send_pays.payments.last() {
825            new_state.created = last.created_index()
826        }
827        if let Some(last) = updated_send_pays.payments.last() {
828            new_state.updated = last.updated_index()
829        }
830
831        let hash_groups: HashMap<_, _> = created_send_pays
832            .payments
833            .iter()
834            .chain(updated_send_pays.payments.iter())
835            .map(|p| {
836                let mut key = hex::encode(&p.payment_hash);
837                key.push('|');
838                key.push_str(&p.groupid.to_string());
839                (key, (p.payment_hash.clone(), p.groupid.to_string()))
840            })
841            .collect();
842        let hash_group_values: Vec<_> = hash_groups.values().cloned().collect();
843
844        self.persister.insert_send_pays(
845            &created_send_pays
846                .payments
847                .into_iter()
848                .map(TryInto::try_into)
849                .collect::<Result<Vec<_>, _>>()?,
850        )?;
851        self.persister.insert_send_pays(
852            &updated_send_pays
853                .payments
854                .into_iter()
855                .map(TryInto::try_into)
856                .collect::<Result<Vec<_>, _>>()?,
857        )?;
858
859        // Now all new send_pays are persisted. Retrieve the send_pays for the
860        // payment hashes, to ensure any send_pays belonging to the same payment
861        // that were not fetched in this round are also included.
862        let send_pays = self.persister.list_send_pays(&hash_group_values)?;
863
864        // Now that all send_pays belonging to all payments are here, aggregate
865        // the send_pays into payments. This is a copy of what core lightning's
866        // listpays function does under the hood.
867        let mut outbound_payments: HashMap<String, SendPayAgg> = HashMap::new();
868        for send_pay in send_pays {
869            let mut key = hex::encode(&send_pay.payment_hash);
870            key.push('|');
871            key.push_str(&send_pay.groupid);
872            let payment = outbound_payments.entry(key).or_insert(SendPayAgg {
873                state: 0,
874                created_at: send_pay.created_at,
875                payment_hash: send_pay.payment_hash,
876                bolt11: None,
877                destination: None,
878                label: None,
879                description: None,
880                preimage: None,
881                amount_sent: 0,
882                amount: Some(0),
883                num_nonfailed_parts: 0,
884            });
885            if payment.bolt11.is_none() {
886                payment.bolt11 = send_pay.bolt11;
887            }
888            if payment.destination.is_none() {
889                payment.destination = send_pay.destination;
890            }
891            if payment.description.is_none() {
892                payment.description = send_pay.description;
893            }
894            if payment.label.is_none() {
895                payment.label = send_pay.label;
896            }
897            if payment.preimage.is_none() {
898                payment.preimage = send_pay.payment_preimage;
899            }
900            if send_pay.created_at < payment.created_at {
901                payment.created_at = send_pay.created_at;
902            }
903
904            match send_pay.status {
905                SendPayStatus::Pending => {
906                    add_amount_sent(payment, send_pay.amount_sent_msat, send_pay.amount_msat);
907                    payment.num_nonfailed_parts += 1;
908                    payment.state |= PAYMENT_STATE_PENDING;
909                }
910                SendPayStatus::Failed => {
911                    payment.state |= PAYMENT_STATE_FAILED;
912                }
913                SendPayStatus::Complete => {
914                    add_amount_sent(payment, send_pay.amount_sent_msat, send_pay.amount_msat);
915                    payment.num_nonfailed_parts += 1;
916                    payment.state |= PAYMENT_STATE_COMPLETE;
917                }
918            }
919        }
920
921        let outbound_payments: Vec<Payment> = outbound_payments
922            .into_values()
923            .map(TryInto::try_into)
924            .collect::<Result<Vec<_>, _>>()?;
925        let outbound_payments = update_payment_expirations(outbound_payments, htlc_list)?;
926        Ok((outbound_payments, new_state))
927    }
928
929    async fn wait_channel_reestablished(&self, path: &PaymentPath) -> NodeResult<()> {
930        let deadline =
931            Instant::now()
932                .checked_add(Duration::from_secs(20))
933                .ok_or(NodeError::generic(
934                    "Failed to set channel establishment deadline",
935                ))?;
936
937        while Instant::now().le(&deadline) && !self.poll_channel_reestablished(path).await? {
938            tokio::time::sleep(Duration::from_millis(50)).await
939        }
940
941        Ok(())
942    }
943
944    async fn poll_channel_reestablished(&self, path: &PaymentPath) -> NodeResult<bool> {
945        let edge = match path.edges.first() {
946            Some(edge) => edge,
947            None => return Err(NodeError::generic("Channel not found")),
948        };
949        let mut client = self.get_node_client().await?;
950        let req = cln::ListpeerchannelsRequest {
951            id: Some(edge.node_id.clone()),
952        };
953        let res = with_connection_retry!(client.list_peer_channels(req.clone()))
954            .await?
955            .into_inner();
956        let channel = match res.channels.iter().find(|c| {
957            match (
958                c.alias.as_ref().and_then(|a| a.local.as_ref()),
959                c.short_channel_id.as_ref(),
960            ) {
961                (Some(alias), Some(short_channel_id)) => {
962                    *alias == edge.short_channel_id || *short_channel_id == edge.short_channel_id
963                }
964                (Some(alias), None) => *alias == edge.short_channel_id,
965                (None, Some(short_channel_id)) => *short_channel_id == edge.short_channel_id,
966                (None, None) => false,
967            }
968        }) {
969            Some(channel) => channel,
970            None => return Err(NodeError::generic("Channel not found")),
971        };
972
973        if let Some(peer_connected) = channel.peer_connected {
974            if !peer_connected {
975                return Ok(false);
976            }
977        }
978
979        if !channel
980            .status
981            .iter()
982            .any(|s| s.contains("Channel ready") || s.contains("Reconnected, and reestablished"))
983        {
984            return Ok(false);
985        }
986
987        Ok(true)
988    }
989}
990
991fn add_amount_sent(
992    agg: &mut SendPayAgg,
993    send_pay_amount_sent_msat: Option<u64>,
994    send_pay_amount_msat: Option<u64>,
995) {
996    if let Some(amount_sent_msat) = send_pay_amount_sent_msat {
997        agg.amount_sent += amount_sent_msat;
998    }
999
1000    let amount_msat = match send_pay_amount_msat {
1001        Some(amount_msat) => amount_msat,
1002        None => {
1003            agg.amount = None;
1004            return;
1005        }
1006    };
1007
1008    if let Some(amount) = agg.amount {
1009        agg.amount = Some(amount + amount_msat);
1010    }
1011}
1012
1013#[derive(Clone, Debug, Default, Deserialize, Serialize)]
1014struct SyncIndex {
1015    pub created: u64,
1016    pub updated: u64,
1017}
1018
1019#[derive(Clone, Debug, Default, Deserialize, Serialize)]
1020struct SyncState {
1021    pub send_pays_index: SyncIndex,
1022    pub list_invoices_index: SyncIndex,
1023}
1024
1025#[tonic::async_trait]
1026impl NodeAPI for Greenlight {
1027    async fn node_credentials(&self) -> NodeResult<Option<NodeCredentials>> {
1028        Ok(Self::get_node_credentials(
1029            self.sdk_config.network,
1030            &self.get_signer().await,
1031            self.persister.clone(),
1032        )?
1033        .map(|credentials| NodeCredentials::Greenlight {
1034            credentials: GreenlightDeviceCredentials {
1035                device: credentials.to_bytes(),
1036            },
1037        }))
1038    }
1039
1040    async fn configure_node(&self, close_to_address: Option<String>) -> NodeResult<()> {
1041        match close_to_address {
1042            Some(close_to_addr) => {
1043                let mut client = self.get_client().await?;
1044                let req = gl_client::pb::GlConfig { close_to_addr };
1045                with_connection_retry!(client.configure(req.clone()))
1046                    .await
1047                    .map_err(|e| NodeError::Generic(format!("Unable to set node config: {e}")))?;
1048            }
1049            None => {
1050                let mut client = self.get_node_client().await?;
1051                let req = cln::DeldatastoreRequest {
1052                    key: vec!["glconf".to_string(), "request".to_string()],
1053                    generation: None,
1054                };
1055                with_connection_retry!(client.del_datastore(req.clone()))
1056                    .await
1057                    .map_err(|e| {
1058                        NodeError::Generic(format!("Unable to delete node config: {e}"))
1059                    })?;
1060            }
1061        }
1062        Ok(())
1063    }
1064
1065    async fn create_invoice(&self, request: CreateInvoiceRequest) -> NodeResult<String> {
1066        let mut client = self.get_node_client().await?;
1067        let label = serde_json::to_string(&InvoiceLabel {
1068            unix_milli: SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis(),
1069            payer_amount_msat: request.payer_amount_msat,
1070        })?;
1071        let cln_request = cln::InvoiceRequest {
1072            amount_msat: Some(cln::AmountOrAny {
1073                value: Some(cln::amount_or_any::Value::Amount(cln::Amount {
1074                    msat: request.amount_msat,
1075                })),
1076            }),
1077            label,
1078            description: request.description,
1079            preimage: request.preimage,
1080            deschashonly: request.use_description_hash,
1081            expiry: request.expiry.map(|e| e as u64),
1082            fallbacks: vec![],
1083            cltv: request.cltv,
1084        };
1085
1086        let res = with_connection_retry!(client.invoice(cln_request.clone()))
1087            .await?
1088            .into_inner();
1089        Ok(res.bolt11)
1090    }
1091
1092    async fn delete_invoice(&self, bolt11: String) -> NodeResult<()> {
1093        let mut client = self.get_node_client().await?;
1094        let invoice_request = cln::ListinvoicesRequest {
1095            invstring: Some(bolt11),
1096            ..Default::default()
1097        };
1098        let invoice_result = with_connection_retry!(client.list_invoices(invoice_request.clone()))
1099            .await?
1100            .into_inner();
1101        let invoice_result = invoice_result.invoices.first();
1102        let result = match invoice_result {
1103            Some(result) => result,
1104            None => return Ok(()),
1105        };
1106
1107        let status = match result.status() {
1108            ListinvoicesInvoicesStatus::Unpaid => DelinvoiceStatus::Unpaid,
1109            ListinvoicesInvoicesStatus::Paid => return Err(NodeError::InvoiceAlreadyPaid),
1110            ListinvoicesInvoicesStatus::Expired => DelinvoiceStatus::Expired,
1111        };
1112        with_connection_retry!(client.del_invoice(DelinvoiceRequest {
1113            label: result.label.clone(),
1114            status: status.into(),
1115            desconly: Some(false),
1116        }))
1117        .await?;
1118        Ok(())
1119    }
1120
1121    async fn fetch_bolt11(&self, payment_hash: Vec<u8>) -> NodeResult<Option<FetchBolt11Result>> {
1122        let mut client = self.get_node_client().await?;
1123        let request = cln::ListinvoicesRequest {
1124            payment_hash: Some(payment_hash),
1125            ..Default::default()
1126        };
1127
1128        let result = with_connection_retry!(client.list_invoices(request.clone()))
1129            .await?
1130            .into_inner()
1131            .invoices
1132            .first()
1133            .cloned()
1134            .and_then(|invoice| {
1135                invoice.bolt11.map(|bolt11| FetchBolt11Result {
1136                    bolt11,
1137                    payer_amount_msat: serde_json::from_str::<InvoiceLabel>(&invoice.label)
1138                        .map(|label| label.payer_amount_msat)
1139                        .ok()
1140                        .flatten(),
1141                })
1142            });
1143
1144        Ok(result)
1145    }
1146
1147    // implement pull changes from greenlight
1148    async fn pull_changed(
1149        &self,
1150        sync_state: Option<Value>,
1151        match_local_balance: bool,
1152    ) -> NodeResult<SyncResponse> {
1153        let sync_state: SyncState = match sync_state {
1154            Some(sync_state) => serde_json::from_value(sync_state)?,
1155            None => SyncState::default(),
1156        };
1157
1158        let client = self.get_node_client().await?;
1159
1160        // get node info
1161        let mut client_clone1 = client.clone();
1162        let node_info_future =
1163            with_connection_retry!(client_clone1.getinfo(cln::GetinfoRequest::default()));
1164
1165        // list both off chain funds and on chain fudns
1166        let funds_future = self.list_funds();
1167
1168        // Fetch closed channels from greenlight
1169        let mut client_clone2 = client.clone();
1170        let closed_channels_future = with_connection_retry!(
1171            client_clone2.list_closed_channels(cln::ListclosedchannelsRequest { id: None })
1172        );
1173
1174        // calculate the node new balance and in case the caller signals balance has changed
1175        // keep polling until the balance is updated
1176        let balance_future = Greenlight::fetch_channels_and_balance_with_retry(
1177            client.clone(),
1178            self.persister.clone(),
1179            match_local_balance,
1180        );
1181
1182        let (node_info_res, funds_res, closed_channels_res, balance_res) = tokio::join!(
1183            node_info_future,
1184            funds_future,
1185            closed_channels_future,
1186            balance_future
1187        );
1188
1189        let node_info = node_info_res?.into_inner();
1190        let funds = funds_res?;
1191        let closed_channels = closed_channels_res?.into_inner().closedchannels;
1192        let (all_channels, opened_channels, connected_peers, channels_balance) = balance_res?;
1193        let forgotten_closed_channels: NodeResult<Vec<Channel>> = closed_channels
1194            .into_iter()
1195            .filter(|cc| {
1196                all_channels
1197                    .iter()
1198                    .all(|ac| ac.funding_txid != Some(cc.funding_txid.clone()))
1199            })
1200            .map(TryInto::try_into)
1201            .collect();
1202        info!("forgotten_closed_channels {:?}", forgotten_closed_channels);
1203
1204        let mut all_channel_models: Vec<Channel> =
1205            all_channels.clone().into_iter().map(|c| c.into()).collect();
1206        all_channel_models.extend(forgotten_closed_channels?);
1207
1208        // calculate onchain balance
1209        let onchain_balance = self.on_chain_balance(&funds).await?;
1210        let pending_onchain_balance = self.pending_onchain_balance(&all_channels).await?;
1211        let utxos: Vec<UnspentTransactionOutput> = self.utxos(funds).await?;
1212
1213        // calculate payment limits and inbound liquidity
1214        let mut max_payable: u64 = 0;
1215        let mut max_receivable_single_channel: u64 = 0;
1216        let mut total_inbound_liquidity_msats: u64 = 0;
1217        opened_channels.iter().try_for_each(|c| -> Result<()> {
1218            max_payable += c
1219                .spendable_msat
1220                .as_ref()
1221                .map(|a| a.msat)
1222                .unwrap_or_default();
1223            let receivable_amount = c
1224                .receivable_msat
1225                .as_ref()
1226                .map(|a| a.msat)
1227                .unwrap_or_default();
1228            total_inbound_liquidity_msats += receivable_amount;
1229            if receivable_amount > max_receivable_single_channel {
1230                max_receivable_single_channel = receivable_amount;
1231            }
1232            Ok(())
1233        })?;
1234
1235        let max_allowed_to_receive_msats =
1236            MAX_INBOUND_LIQUIDITY_MSAT.saturating_sub(channels_balance);
1237        let node_pubkey = hex::encode(node_info.id);
1238        // construct the node state
1239        let node_state = NodeState {
1240            id: node_pubkey.clone(),
1241            block_height: node_info.blockheight,
1242            channels_balance_msat: channels_balance,
1243            onchain_balance_msat: onchain_balance,
1244            pending_onchain_balance_msat: pending_onchain_balance,
1245            utxos,
1246            max_payable_msat: max_payable,
1247            max_receivable_msat: max_allowed_to_receive_msats,
1248            max_single_payment_amount_msat: MAX_PAYMENT_AMOUNT_MSAT,
1249            max_chan_reserve_msats: channels_balance - min(max_payable, channels_balance),
1250            connected_peers,
1251            max_receivable_single_payment_amount_msat: max_receivable_single_channel,
1252            total_inbound_liquidity_msats,
1253        };
1254        let mut htlc_list: Vec<Htlc> = Vec::new();
1255        for channel in all_channel_models.clone() {
1256            htlc_list.extend(channel.htlcs);
1257        }
1258
1259        let (new_sync_state, payments) = self.pull_transactions(&sync_state, htlc_list).await?;
1260
1261        Ok(SyncResponse {
1262            sync_state: serde_json::to_value(new_sync_state)?,
1263            node_state,
1264            payments,
1265            channels: all_channel_models,
1266        })
1267    }
1268
1269    async fn send_pay(&self, bolt11: String, max_hops: u32) -> NodeResult<PaymentResponse> {
1270        let invoice = parse_invoice(&bolt11)?;
1271        let last_hop = invoice.routing_hints.first().and_then(|rh| rh.hops.first());
1272        let mut client = self.get_node_client().await?;
1273
1274        // Valid the invoice network against the config network
1275        validate_network(invoice.clone(), self.sdk_config.network)?;
1276
1277        // We first calculate for each channel the max amount to pay (at the receiver)
1278        let mut max_amount_per_channel = self
1279            .max_sendable_amount(Some(hex::decode(invoice.payee_pubkey)?), max_hops, last_hop)
1280            .await?;
1281        info!("send_pay: routes: {:?}", max_amount_per_channel);
1282
1283        // Calculate the total amount to pay
1284        let total_msat: u64 = max_amount_per_channel.iter().map(|m| m.amount_msat).sum();
1285
1286        // Sort the channels by max amount descending so we can build the route in a way that it
1287        // drains the largest channels first
1288        max_amount_per_channel.sort_by_key(|m| Reverse(m.amount_msat));
1289
1290        let amount_to_pay_msat = match invoice.amount_msat {
1291            Some(amount) => Ok(amount),
1292            None => Err(NodeError::generic("Invoice has no amount")),
1293        }?;
1294
1295        if amount_to_pay_msat > total_msat {
1296            return Err(NodeError::RouteNotFound(format!(
1297                "Amount too high, max amount is {total_msat} msat"
1298            )));
1299        }
1300
1301        // This is needed in greenlight for the signer to recognize this invoice.
1302        client
1303            .pre_approve_invoice(PreapproveinvoiceRequest {
1304                bolt11: Some(bolt11.clone()),
1305            })
1306            .await?;
1307
1308        // We need to allocate a part id for each part that we are sending.
1309        let mut part_id = 1;
1310        // The total amount we sent. i.e. what the recipient received + fees
1311        let mut amount_sent_msat = 0;
1312        // The total amount received by the recipient
1313        let mut amount_received_msat = 0;
1314        // Generate a random group_id for the payment
1315        let group_id = rand::random::<u64>();
1316
1317        // The algorithm goes over each channel and drains it until the received amount
1318        // equals to the amount to pay defined in the bolt11 invoice.
1319        for max in max_amount_per_channel {
1320            // calculating the incoming amount for the remaining amount to pay.
1321            let left_to_pay_msat = amount_to_pay_msat - amount_received_msat;
1322            // Whether we draining the whole channel balance or only what is left to pay
1323            let to_pay_msat = std::cmp::min(left_to_pay_msat, max.amount_msat);
1324
1325            // We convert our payment path to an actual route that can be sent to the node.
1326            // This requires calculating the right fees and cltv delta in each hop.
1327            let (route, sent_msat) = convert_to_send_pay_route(
1328                max.path.clone(),
1329                to_pay_msat,
1330                invoice.min_final_cltv_expiry_delta,
1331            );
1332            info!("send_pay route to pay: {route:?}, received_amount = {to_pay_msat}");
1333            self.wait_channel_reestablished(&max.path).await?;
1334            // We send the part using the node API
1335            let req = SendpayRequest {
1336                route,
1337                payment_hash: hex::decode(invoice.payment_hash.clone())?,
1338                label: None,
1339                amount_msat: Some(Amount {
1340                    msat: amount_to_pay_msat,
1341                }),
1342                bolt11: Some(bolt11.clone()),
1343                payment_secret: Some(invoice.payment_secret.clone()),
1344                partid: Some(part_id),
1345                localinvreqid: None,
1346                groupid: Some(group_id),
1347            };
1348            let mut client = client.clone();
1349            with_connection_retry!(client.send_pay(req.clone())).await?;
1350            part_id += 1;
1351            amount_sent_msat += sent_msat;
1352            amount_received_msat += to_pay_msat;
1353            if amount_received_msat == amount_to_pay_msat {
1354                break;
1355            }
1356        }
1357
1358        // Now we wait for the first part to be completed as a way to wait for the payment
1359        // to complete.
1360        let req = WaitsendpayRequest {
1361            payment_hash: hex::decode(invoice.payment_hash.clone())?,
1362            partid: Some(1),
1363            timeout: Some(self.sdk_config.payment_timeout_sec),
1364            groupid: Some(group_id),
1365        };
1366        let response = self
1367            .with_keep_alive(with_connection_retry!(client.wait_send_pay(req.clone())))
1368            .await?
1369            .into_inner();
1370        Ok(PaymentResponse {
1371            payment_time: response.completed_at.unwrap_or(response.created_at as f64) as i64,
1372            amount_msat: amount_received_msat,
1373            fee_msat: amount_sent_msat - amount_received_msat,
1374            payment_hash: invoice.payment_hash,
1375            payment_preimage: hex::encode(response.payment_preimage.unwrap_or_default()),
1376        })
1377    }
1378
1379    async fn send_payment(
1380        &self,
1381        bolt11: String,
1382        amount_msat: Option<u64>,
1383        label: Option<String>,
1384    ) -> NodeResult<Payment> {
1385        let mut description = None;
1386        if !bolt11.is_empty() {
1387            let invoice = parse_invoice(&bolt11)?;
1388            validate_network(invoice.clone(), self.sdk_config.network)?;
1389            description = invoice.description;
1390        }
1391
1392        let mut client = self.get_node_client().await?;
1393        let request = cln::PayRequest {
1394            bolt11,
1395            amount_msat: amount_msat.map(|amt| cln::Amount { msat: amt }),
1396            maxfeepercent: Some(self.sdk_config.maxfee_percent),
1397            retry_for: Some(self.sdk_config.payment_timeout_sec),
1398            label,
1399            maxdelay: None,
1400            riskfactor: None,
1401            localinvreqid: None,
1402            exclude: vec![],
1403            maxfee: None,
1404            description,
1405            exemptfee: Some(cln::Amount {
1406                msat: self.sdk_config.exemptfee_msat,
1407            }),
1408        };
1409        let result: cln::PayResponse = self
1410            .with_keep_alive(with_connection_retry!(client.pay(request.clone())))
1411            .await?
1412            .into_inner();
1413
1414        // Before returning from send_payment we need to make sure it is persisted in the backend node.
1415        // We do so by polling for the payment.
1416        let payment = Self::fetch_outgoing_payment_with_retry(client, result.payment_hash).await?;
1417        payment.try_into()
1418    }
1419
1420    async fn send_trampoline_payment(
1421        &self,
1422        bolt11: String,
1423        amount_msat: u64,
1424        label: Option<String>,
1425        trampoline_node_id: Vec<u8>,
1426    ) -> NodeResult<Payment> {
1427        let invoice = parse_invoice(&bolt11)?;
1428        validate_network(invoice.clone(), self.sdk_config.network)?;
1429        let label = serde_json::to_string(&PaymentLabel {
1430            trampoline: true,
1431            client_label: label,
1432            unix_nano: SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos(),
1433            amount_msat,
1434        })?;
1435        let fee_msat =
1436            (amount_msat.saturating_mul(TRAMPOLINE_FEE_PPM) / 1_000_000) + TRAMPOLINE_BASE_FEE_MSAT;
1437        let fee_percent = ((fee_msat as f64 / amount_msat as f64) * 100.) as f32;
1438        debug!("using fee msat {} fee percent {}", fee_msat, fee_percent);
1439        let mut client = self.get_client().await?;
1440        let request = TrampolinePayRequest {
1441            bolt11,
1442            trampoline_node_id,
1443            amount_msat,
1444            label,
1445            maxdelay: u32::default(),
1446            description: String::default(),
1447            maxfeepercent: fee_percent,
1448        };
1449        let result = self
1450            .with_keep_alive(with_connection_retry!(
1451                client.trampoline_pay(request.clone())
1452            ))
1453            .await?
1454            .into_inner();
1455
1456        let client = self.get_node_client().await?;
1457
1458        // Before returning from send_payment we need to make sure it is
1459        // persisted in the backend node. We do so by polling for the payment.
1460        // TODO: Ensure this works with trampoline payments
1461        // NOTE: If this doesn't work with trampoline payments, the sync also
1462        // needs updating.
1463        let payment = Self::fetch_outgoing_payment_with_retry(client, result.payment_hash).await?;
1464        payment.try_into()
1465    }
1466
1467    async fn send_spontaneous_payment(
1468        &self,
1469        node_id: String,
1470        amount_msat: u64,
1471        extra_tlvs: Option<Vec<TlvEntry>>,
1472        label: Option<String>,
1473    ) -> NodeResult<Payment> {
1474        let mut client: node::ClnClient = self.get_node_client().await?;
1475        let request = cln::KeysendRequest {
1476            destination: hex::decode(node_id)?,
1477            amount_msat: Some(cln::Amount { msat: amount_msat }),
1478            label: label.or(Some(format!(
1479                "breez-{}",
1480                SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis()
1481            ))),
1482            extratlvs: extra_tlvs.map(|tlvs| cln::TlvStream {
1483                entries: tlvs
1484                    .into_iter()
1485                    .map(|tlv| cln::TlvEntry {
1486                        r#type: tlv.field_number,
1487                        value: tlv.value,
1488                    })
1489                    .collect(),
1490            }),
1491            routehints: None,
1492            maxfeepercent: Some(self.sdk_config.maxfee_percent),
1493            exemptfee: None,
1494            retry_for: Some(self.sdk_config.payment_timeout_sec),
1495            maxdelay: None,
1496        };
1497
1498        // Not wrapped with connection retry, in case it causes to send twice.
1499        let result = self
1500            .with_keep_alive(client.key_send(request))
1501            .await?
1502            .into_inner();
1503
1504        // Before returning from send_payment we need to make sure it is persisted in the backend node.
1505        // We do so by polling for the payment.
1506        let payment = Self::fetch_outgoing_payment_with_retry(client, result.payment_hash).await?;
1507        payment.try_into()
1508    }
1509
1510    async fn node_id(&self) -> NodeResult<String> {
1511        Ok(hex::encode(self.get_signer().await.node_id()))
1512    }
1513
1514    async fn redeem_onchain_funds(
1515        &self,
1516        to_address: String,
1517        sat_per_vbyte: u32,
1518    ) -> NodeResult<Vec<u8>> {
1519        let mut client = self.get_node_client().await?;
1520
1521        let request = cln::WithdrawRequest {
1522            feerate: Some(cln::Feerate {
1523                style: Some(cln::feerate::Style::Perkw(sat_per_vbyte * 250)),
1524            }),
1525            satoshi: Some(cln::AmountOrAll {
1526                value: Some(cln::amount_or_all::Value::All(true)),
1527            }),
1528            destination: to_address,
1529            minconf: None,
1530            utxos: vec![],
1531        };
1532
1533        Ok(with_connection_retry!(client.withdraw(request.clone()))
1534            .await?
1535            .into_inner()
1536            .txid)
1537    }
1538
1539    async fn prepare_redeem_onchain_funds(
1540        &self,
1541        req: PrepareRedeemOnchainFundsRequest,
1542    ) -> NodeResult<PrepareRedeemOnchainFundsResponse> {
1543        let funds = self.list_funds().await?;
1544        let utxos = self.utxos(funds).await?;
1545
1546        let mut amount_msat: u64 = 0;
1547        let txins: Vec<TxIn> = utxos
1548            .iter()
1549            .map(|utxo| {
1550                amount_msat += utxo.amount_millisatoshi;
1551                TxIn {
1552                    previous_output: OutPoint {
1553                        txid: Txid::from_slice(&utxo.txid).unwrap(),
1554                        vout: 0,
1555                    },
1556                    script_sig: Script::new(),
1557                    sequence: Sequence(0),
1558                    witness: Witness::default(),
1559                }
1560            })
1561            .collect();
1562
1563        let amount_sat = amount_msat / 1_000;
1564        let btc_address = Address::from_str(&req.to_address)?;
1565        let tx_out: Vec<TxOut> = vec![TxOut {
1566            value: amount_sat,
1567            script_pubkey: btc_address.payload.script_pubkey(),
1568        }];
1569        let tx = Transaction {
1570            version: 2,
1571            lock_time: crate::bitcoin::PackedLockTime(0),
1572            input: txins.clone(),
1573            output: tx_out,
1574        };
1575
1576        let witness_input_size: u64 = 110;
1577        let tx_weight = tx.strippedsize() as u64 * WITNESS_SCALE_FACTOR as u64
1578            + witness_input_size * txins.len() as u64;
1579        let fee: u64 = tx_weight * req.sat_per_vbyte as u64 / WITNESS_SCALE_FACTOR as u64;
1580        if fee >= amount_sat {
1581            return Err(NodeError::InsufficientFunds(
1582                "Insufficient funds to pay fees".to_string(),
1583            ));
1584        }
1585
1586        return Ok(PrepareRedeemOnchainFundsResponse {
1587            tx_weight,
1588            tx_fee_sat: fee,
1589        });
1590    }
1591
1592    /// Starts the signer that listens in a loop until the shutdown signal is received
1593    async fn start(&self, shutdown: mpsc::Receiver<()>) {
1594        match self.get_signer().await.run_forever(shutdown).await {
1595            Ok(_) => info!("signer exited gracefully"),
1596            Err(e) => error!("signer exited with error: {e}"),
1597        }
1598    }
1599
1600    async fn start_keep_alive(&self, mut shutdown: watch::Receiver<()>) {
1601        info!("keep alive started");
1602        let mut interval = tokio::time::interval(Duration::from_secs(15));
1603        interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
1604        loop {
1605            tokio::select! {
1606                  _ = shutdown.changed() => {
1607                    info!("keep alive exited");
1608                    break;
1609                  }
1610                  _ = interval.tick() => {
1611                    let inprogress_payments = self.inprogress_payments.load(Ordering::Relaxed);
1612                    if inprogress_payments == 0 {
1613                      continue
1614                    }
1615                    let client_res = self.get_node_client().await;
1616                    match client_res {
1617                      Ok(mut client) => {
1618                        let res = client.getinfo(cln::GetinfoRequest {}).await;
1619                        match res {
1620                          Ok(_) => {
1621                            info!("keep alive ping sent, in progress payments: {inprogress_payments}");
1622                          }
1623                          Err(e) => {
1624                            error!("keep alive ping failed: {e}");
1625                          }
1626                        }
1627                      }
1628                      Err(e) => {
1629                        error!("keep alive ping failed to create client: {e}");
1630                      }
1631                    }
1632                  }
1633            }
1634        }
1635    }
1636
1637    async fn connect_peer(&self, id: String, addr: String) -> NodeResult<()> {
1638        let mut client = self.get_node_client().await?;
1639        let connect_req = cln::ConnectRequest {
1640            id: format!("{id}@{addr}"),
1641            host: None,
1642            port: None,
1643        };
1644        with_connection_retry!(client.connect_peer(connect_req.clone())).await?;
1645        Ok(())
1646    }
1647
1648    async fn sign_message(&self, message: &str) -> NodeResult<String> {
1649        let (sig, recovery_id) = self
1650            .get_signer()
1651            .await
1652            .sign_message(message.as_bytes().to_vec())?;
1653        let mut complete_signature = vec![31 + recovery_id];
1654        complete_signature.extend_from_slice(&sig);
1655        Ok(zbase32::encode_full_bytes(&complete_signature))
1656    }
1657
1658    async fn check_message(
1659        &self,
1660        message: &str,
1661        pubkey: &str,
1662        signature: &str,
1663    ) -> NodeResult<bool> {
1664        let pk = PublicKey::from_str(pubkey)?;
1665        Ok(verify(message.as_bytes(), signature, &pk))
1666    }
1667
1668    async fn sign_invoice(&self, invoice: RawBolt11Invoice) -> NodeResult<String> {
1669        let hrp_bytes = invoice.hrp.to_string().as_bytes().to_vec();
1670        let data_bytes = invoice.data.to_base32();
1671
1672        // create the message for the signer
1673        let msg_type: u16 = 8;
1674        let data_len: u16 = data_bytes.len().try_into()?;
1675        let mut data_len_bytes = data_len.to_be_bytes().to_vec();
1676        let mut data_buf = data_bytes.iter().copied().map(u5::to_u8).collect();
1677
1678        let hrp_len: u16 = hrp_bytes.len().try_into()?;
1679        let mut hrp_len_bytes = hrp_len.to_be_bytes().to_vec();
1680        let mut hrp_buf = hrp_bytes.to_vec();
1681
1682        let mut buf = msg_type.to_be_bytes().to_vec();
1683        buf.append(&mut data_len_bytes);
1684        buf.append(&mut data_buf);
1685        buf.append(&mut hrp_len_bytes);
1686        buf.append(&mut hrp_buf);
1687        // Sign the invoice using the signer
1688        let raw_result = self.get_signer().await.sign_invoice(buf)?;
1689        info!(
1690            "recover id: {:?} raw = {:?}",
1691            raw_result, raw_result[64] as i32
1692        );
1693        // contruct the RecoveryId
1694        let rid = RecoveryId::from_i32(raw_result[64] as i32).expect("recovery ID");
1695        let sig = &raw_result[0..64];
1696        let recoverable_sig = RecoverableSignature::from_compact(sig, rid)?;
1697
1698        let signed_invoice: Result<SignedRawBolt11Invoice> = invoice.sign(|_| Ok(recoverable_sig));
1699        Ok(signed_invoice?.to_string())
1700    }
1701
1702    async fn close_peer_channels(&self, node_id: String) -> NodeResult<Vec<String>> {
1703        let mut client = self.get_node_client().await?;
1704        let req = cln::ListpeerchannelsRequest {
1705            id: Some(hex::decode(node_id)?),
1706        };
1707        let closed_channels = with_connection_retry!(client.list_peer_channels(req.clone()))
1708            .await?
1709            .into_inner();
1710        let mut tx_ids = vec![];
1711        for channel in closed_channels.channels {
1712            let mut should_close = false;
1713            if let Some(state) = channel.state {
1714                match cln::ChannelState::from_i32(state) {
1715                    Some(cln::ChannelState::Openingd) => should_close = true,
1716                    Some(cln::ChannelState::ChanneldAwaitingLockin) => should_close = true,
1717                    Some(cln::ChannelState::ChanneldNormal) => should_close = true,
1718                    Some(cln::ChannelState::ChanneldShuttingDown) => should_close = true,
1719                    Some(cln::ChannelState::FundingSpendSeen) => should_close = true,
1720                    Some(cln::ChannelState::DualopendOpenInit) => should_close = true,
1721                    Some(cln::ChannelState::DualopendAwaitingLockin) => should_close = true,
1722                    Some(_) => should_close = false,
1723                    None => should_close = false,
1724                }
1725            }
1726
1727            if should_close {
1728                let chan_id = channel.channel_id.ok_or(anyhow!("Empty channel id"))?;
1729                let req = cln::CloseRequest {
1730                    id: hex::encode(chan_id),
1731                    unilateraltimeout: None,
1732                    destination: None,
1733                    fee_negotiation_step: None,
1734                    wrong_funding: None,
1735                    force_lease_closed: None,
1736                    feerange: vec![],
1737                };
1738                let response = with_connection_retry!(client.close(req.clone())).await;
1739                match response {
1740                    Ok(res) => {
1741                        tx_ids.push(hex::encode(
1742                            res.into_inner()
1743                                .txid
1744                                .ok_or(anyhow!("Empty txid in close response"))?,
1745                        ));
1746                    }
1747                    Err(e) => Err(anyhow!("Empty closing channel: {e}"))?,
1748                };
1749            }
1750        }
1751        Ok(tx_ids)
1752    }
1753
1754    async fn stream_incoming_payments(
1755        &self,
1756    ) -> NodeResult<Pin<Box<dyn Stream<Item = IncomingPayment> + Send>>> {
1757        let mut client = self.get_client().await?;
1758        let req = gl_client::signer::model::greenlight::StreamIncomingFilter {};
1759        let stream = with_connection_retry!(client.stream_incoming(req.clone()))
1760            .await?
1761            .into_inner();
1762        Ok(Box::pin(stream.filter_map(|msg| match msg {
1763            Ok(msg) => match msg.details {
1764                Some(incoming_payment::Details::Offchain(p)) => Some(IncomingPayment {
1765                    label: p.label,
1766                    payment_hash: p.payment_hash,
1767                    preimage: p.preimage,
1768                    amount_msat: amount_to_msat(&p.amount.unwrap_or_default()),
1769                    bolt11: p.bolt11,
1770                }),
1771                _ => None,
1772            },
1773            Err(e) => {
1774                debug!("failed to receive message: {e}");
1775                None
1776            }
1777        })))
1778    }
1779
1780    async fn stream_log_messages(&self) -> NodeResult<Pin<Box<dyn Stream<Item = String> + Send>>> {
1781        let mut client = self.get_client().await?;
1782        let req = gl_client::signer::model::greenlight::StreamLogRequest {};
1783        let stream = with_connection_retry!(client.stream_log(req.clone()))
1784            .await?
1785            .into_inner();
1786        Ok(Box::pin(stream.filter_map(|msg| match msg {
1787            Ok(msg) => Some(msg.line),
1788            Err(e) => {
1789                debug!("failed to receive log message: {e}");
1790                None
1791            }
1792        })))
1793    }
1794
1795    async fn static_backup(&self) -> NodeResult<Vec<String>> {
1796        let mut client = self.get_node_client().await?;
1797        let req = cln::StaticbackupRequest {};
1798        let res = with_connection_retry!(client.static_backup(req.clone()))
1799            .await?
1800            .into_inner();
1801        let hex_vec: Vec<String> = res.scb.into_iter().map(hex::encode).collect();
1802        Ok(hex_vec)
1803    }
1804
1805    async fn generate_diagnostic_data(&self) -> NodeResult<Value> {
1806        let all_commands = vec![
1807            NodeCommand::GetInfo.to_string(),
1808            NodeCommand::ListPeerChannels.to_string(),
1809            NodeCommand::ListFunds.to_string(),
1810            NodeCommand::ListPayments.to_string(),
1811            NodeCommand::ListInvoices.to_string(),
1812        ];
1813
1814        let mut result = Map::new();
1815        for command in all_commands {
1816            let command_name = command.clone();
1817            let res = self
1818                .execute_command(command)
1819                .await
1820                .unwrap_or_else(|e| json!({ "error": e.to_string() }));
1821            result.insert(command_name, res);
1822        }
1823        Ok(Value::Object(result))
1824    }
1825
1826    async fn execute_command(&self, command: String) -> NodeResult<Value> {
1827        let node_cmd =
1828            NodeCommand::from_str(&command).map_err(|_| anyhow!("Command not found: {command}"))?;
1829
1830        let mut client = self.get_node_client().await?;
1831        match node_cmd {
1832            NodeCommand::ListPeers => {
1833                let req = cln::ListpeersRequest::default();
1834                let resp = with_connection_retry!(client.list_peers(req.clone()))
1835                    .await?
1836                    .into_inner();
1837
1838                Ok(crate::serializer::value::to_value(&resp)?)
1839            }
1840            NodeCommand::ListPeerChannels => {
1841                let req = cln::ListpeerchannelsRequest::default();
1842                let resp = with_connection_retry!(client.list_peer_channels(req.clone()))
1843                    .await?
1844                    .into_inner();
1845                Ok(crate::serializer::value::to_value(&resp)?)
1846            }
1847            NodeCommand::ListFunds => {
1848                let req = cln::ListfundsRequest::default();
1849                let resp = with_connection_retry!(client.list_funds(req.clone()))
1850                    .await?
1851                    .into_inner();
1852                Ok(crate::serializer::value::to_value(&resp)?)
1853            }
1854            NodeCommand::ListPayments => {
1855                let req = cln::ListpaysRequest::default();
1856                let resp = with_connection_retry!(client.list_pays(req.clone()))
1857                    .await?
1858                    .into_inner();
1859                Ok(crate::serializer::value::to_value(&resp)?)
1860            }
1861            NodeCommand::ListInvoices => {
1862                let req = cln::ListinvoicesRequest::default();
1863                let resp = with_connection_retry!(client.list_invoices(req.clone()))
1864                    .await?
1865                    .into_inner();
1866
1867                Ok(crate::serializer::value::to_value(&resp)?)
1868            }
1869            NodeCommand::CloseAllChannels => {
1870                let req = cln::ListpeersRequest::default();
1871                let resp = with_connection_retry!(client.list_peers(req.clone()))
1872                    .await?
1873                    .into_inner();
1874                for p in resp.peers {
1875                    self.close_peer_channels(hex::encode(p.id)).await?;
1876                }
1877
1878                Ok(Value::String("All channels were closed".to_string()))
1879            }
1880            NodeCommand::GetInfo => {
1881                let req = cln::GetinfoRequest::default();
1882                let resp = with_connection_retry!(client.getinfo(req.clone()))
1883                    .await?
1884                    .into_inner();
1885                Ok(crate::serializer::value::to_value(&resp)?)
1886            }
1887            NodeCommand::Stop => {
1888                let req = cln::StopRequest::default();
1889                let resp = with_connection_retry!(client.stop(req.clone()))
1890                    .await?
1891                    .into_inner();
1892                Ok(crate::serializer::value::to_value(&resp)?)
1893            }
1894        }
1895    }
1896
1897    async fn max_sendable_amount<'a>(
1898        &self,
1899        payee_node_id: Option<Vec<u8>>,
1900        max_hops: u32,
1901        last_hop_hint: Option<&'a RouteHintHop>,
1902    ) -> NodeResult<Vec<MaxChannelAmount>> {
1903        let mut client = self.get_node_client().await?;
1904
1905        let mut peers = HashMap::new();
1906        let req = cln::ListpeerchannelsRequest::default();
1907        with_connection_retry!(client.list_peer_channels(req.clone()))
1908            .await?
1909            .into_inner()
1910            .channels
1911            .into_iter()
1912            .for_each(|channel| {
1913                peers
1914                    .entry(channel.peer_id().to_vec())
1915                    .or_insert(Vec::new())
1916                    .push(channel)
1917            });
1918
1919        let mut max_channel_amounts = vec![];
1920        for (peer, channels) in peers {
1921            let max_amounts_for_peer = self
1922                .max_sendable_amount_from_peer(
1923                    peer,
1924                    channels,
1925                    payee_node_id.clone(),
1926                    max_hops,
1927                    last_hop_hint,
1928                )
1929                .await?;
1930            max_channel_amounts.extend_from_slice(max_amounts_for_peer.as_slice());
1931        }
1932        Ok(max_channel_amounts)
1933    }
1934
1935    async fn derive_bip32_key(&self, path: Vec<ChildNumber>) -> NodeResult<ExtendedPrivKey> {
1936        Self::derive_bip32_key(self.sdk_config.network, &self.get_signer().await, path)
1937    }
1938
1939    async fn legacy_derive_bip32_key(&self, path: Vec<ChildNumber>) -> NodeResult<ExtendedPrivKey> {
1940        Self::legacy_derive_bip32_key(self.sdk_config.network, &self.get_signer().await, path)
1941    }
1942
1943    async fn stream_custom_messages(
1944        &self,
1945    ) -> NodeResult<Pin<Box<dyn Stream<Item = Result<CustomMessage>> + Send>>> {
1946        let stream = {
1947            let mut client = self.get_client().await?;
1948            let req = gl_client::signer::model::greenlight::StreamCustommsgRequest {};
1949            with_connection_retry!(client.stream_custommsg(req.clone()))
1950                .await?
1951                .into_inner()
1952        };
1953
1954        Ok(Box::pin(stream.filter_map(|msg| {
1955            let msg = match msg {
1956                Ok(msg) => msg,
1957                Err(e) => return Some(Err(anyhow!("failed to receive message: {e}"))),
1958            };
1959
1960            if msg.payload.len() < 2 {
1961                debug!(
1962                    "received too short custom message payload: {:?}",
1963                    &msg.payload
1964                );
1965                return None;
1966            }
1967
1968            let msg_type = u16::from_be_bytes([msg.payload[0], msg.payload[1]]);
1969
1970            Some(Ok(CustomMessage {
1971                peer_id: msg.peer_id,
1972                message_type: msg_type,
1973                payload: msg.payload[2..].to_vec(),
1974            }))
1975        })))
1976    }
1977
1978    async fn send_custom_message(&self, message: CustomMessage) -> NodeResult<()> {
1979        let mut client = self.get_node_client().await?;
1980
1981        let mut msg = message.message_type.to_be_bytes().to_vec();
1982        msg.extend(message.payload);
1983        let req = cln::SendcustommsgRequest {
1984            msg,
1985            node_id: message.peer_id,
1986        };
1987        let resp = with_connection_retry!(client.send_custom_msg(req.clone()))
1988            .await?
1989            .into_inner();
1990        debug!("send_custom_message returned status {:?}", resp.status);
1991        Ok(())
1992    }
1993
1994    // Gets the routing hints related to all private channels that the node has
1995    async fn get_routing_hints(
1996        &self,
1997        lsp_info: &LspInformation,
1998    ) -> NodeResult<(Vec<RouteHint>, bool)> {
1999        let mut client = self.get_node_client().await?;
2000
2001        let open_peer_channels = self.get_open_peer_channels_pb().await?;
2002        let (open_peer_channels_private, open_peer_channels_public): (
2003            HashMap<Vec<u8>, ListpeerchannelsChannels>,
2004            HashMap<Vec<u8>, ListpeerchannelsChannels>,
2005        ) = open_peer_channels
2006            .into_iter()
2007            .partition(|(_, c)| c.private.unwrap_or_default());
2008        let has_public_channel = !open_peer_channels_public.is_empty();
2009
2010        let mut hints: Vec<RouteHint> = vec![];
2011
2012        // Get channels where our node is the destination
2013        let pubkey = self
2014            .persister
2015            .get_node_state()?
2016            .map(|n| n.id)
2017            .ok_or(NodeError::generic("Node info not found"))?;
2018        let req = cln::ListchannelsRequest {
2019            destination: Some(hex::decode(pubkey)?),
2020            ..Default::default()
2021        };
2022        let channels: HashMap<Vec<u8>, cln::ListchannelsChannels> =
2023            with_connection_retry!(client.list_channels(req.clone()))
2024                .await?
2025                .into_inner()
2026                .channels
2027                .into_iter()
2028                .map(|c| (c.source.clone(), c))
2029                .collect();
2030
2031        // Create a routing hint from each private channel.
2032        for (peer_id, peer_channel) in open_peer_channels_private {
2033            let peer_id_str = hex::encode(&peer_id);
2034            let optional_channel_id = peer_channel
2035                .alias
2036                .and_then(|a| a.remote)
2037                .or(peer_channel.short_channel_id);
2038
2039            if let Some(channel_id) = optional_channel_id {
2040                // The remote fee policy
2041                let maybe_policy = match channels.get(&peer_id) {
2042                    Some(channel) => Some((
2043                        channel.base_fee_millisatoshi,
2044                        channel.fee_per_millionth,
2045                        channel.delay,
2046                    )),
2047                    None if peer_id_str == lsp_info.pubkey => Some((
2048                        lsp_info.base_fee_msat as u32,
2049                        (lsp_info.fee_rate * 1000000.0) as u32,
2050                        lsp_info.time_lock_delta,
2051                    )),
2052                    _ => None,
2053                };
2054                match maybe_policy {
2055                    Some((fees_base_msat, fees_proportional_millionths, cltv_delta)) => {
2056                        debug!(
2057                            "For peer {peer_id_str}: remote base {fees_base_msat} proportional {fees_proportional_millionths} cltv_delta {cltv_delta}",
2058                        );
2059                        let hint = RouteHint {
2060                            hops: vec![RouteHintHop {
2061                                src_node_id: peer_id_str,
2062                                short_channel_id: channel_id,
2063                                fees_base_msat,
2064                                fees_proportional_millionths,
2065                                cltv_expiry_delta: cltv_delta as u64,
2066                                htlc_minimum_msat: Some(
2067                                    peer_channel
2068                                        .minimum_htlc_in_msat
2069                                        .clone()
2070                                        .unwrap_or_default()
2071                                        .msat,
2072                                ),
2073                                htlc_maximum_msat: None,
2074                            }],
2075                        };
2076                        info!("Generating hint hop as routing hint: {hint:?}");
2077                        hints.push(hint);
2078                    }
2079                    _ => debug!("No source channel found for peer: {peer_id_str:?}"),
2080                };
2081            }
2082        }
2083        Ok((hints, has_public_channel))
2084    }
2085
2086    async fn get_open_peers(&self) -> NodeResult<HashSet<Vec<u8>>> {
2087        let open_peer_channels = self.get_open_peer_channels_pb().await?;
2088        let open_peers: HashSet<Vec<u8>> = open_peer_channels.into_keys().collect();
2089        Ok(open_peers)
2090    }
2091}
2092
2093#[derive(Clone, PartialEq, Eq, Debug, EnumString, Display, Deserialize, Serialize)]
2094enum NodeCommand {
2095    /// Closes all channels of all peers.
2096    #[strum(serialize = "closeallchannels")]
2097    CloseAllChannels,
2098
2099    /// See <https://docs.corelightning.org/reference/lightning-getinfo>
2100    #[strum(serialize = "getinfo")]
2101    GetInfo,
2102
2103    /// See <https://docs.corelightning.org/reference/lightning-listfunds>
2104    #[strum(serialize = "listfunds")]
2105    ListFunds,
2106
2107    /// See <https://docs.corelightning.org/reference/lightning-listinvoices>
2108    #[strum(serialize = "listinvoices")]
2109    ListInvoices,
2110
2111    /// See <https://docs.corelightning.org/reference/lightning-listpays>
2112    #[strum(serialize = "listpayments")]
2113    ListPayments,
2114
2115    /// See <https://docs.corelightning.org/reference/lightning-listpeers>
2116    #[strum(serialize = "listpeers")]
2117    ListPeers,
2118
2119    /// See <https://docs.corelightning.org/reference/lightning-listpeerchannels>
2120    #[strum(serialize = "listpeerchannels")]
2121    ListPeerChannels,
2122
2123    /// Stops the node.
2124    ///
2125    /// Note that this command will return an error, as the node is stopped before it can reply.
2126    ///
2127    /// See <https://docs.corelightning.org/reference/lightning-stop>
2128    #[strum(serialize = "stop")]
2129    Stop,
2130}
2131
2132struct SendPayAgg {
2133    state: u8,
2134    created_at: u64,
2135    payment_hash: Vec<u8>,
2136    bolt11: Option<String>,
2137    destination: Option<Vec<u8>>,
2138    label: Option<String>,
2139    description: Option<String>,
2140    preimage: Option<Vec<u8>>,
2141    amount_sent: u64,
2142    amount: Option<u64>,
2143    num_nonfailed_parts: u64,
2144}
2145
2146fn update_payment_expirations(
2147    payments: Vec<Payment>,
2148    htlc_list: Vec<Htlc>,
2149) -> NodeResult<Vec<Payment>> {
2150    if htlc_list.is_empty() {
2151        return Ok(payments);
2152    }
2153
2154    let mut payments_res: Vec<Payment> = Vec::new();
2155    for mut payment in payments {
2156        if payment.status == PaymentStatus::Pending {
2157            let new_data = payment.clone().details;
2158            if let PaymentDetails::Ln { data } = new_data {
2159                for htlc in &htlc_list {
2160                    let payment_hash = hex::encode(htlc.clone().payment_hash);
2161                    if payment_hash == data.payment_hash
2162                        && data.pending_expiration_block < Some(htlc.expiry)
2163                    {
2164                        payment.details.add_pending_expiration_block(htlc.clone())
2165                    }
2166                }
2167            }
2168        }
2169        payments_res.push(payment);
2170    }
2171    info!("pending htlc payments {payments_res:?}");
2172    Ok(payments_res)
2173}
2174
2175impl TryFrom<ListsendpaysPayments> for SendPay {
2176    type Error = NodeError;
2177
2178    fn try_from(value: ListsendpaysPayments) -> std::result::Result<Self, Self::Error> {
2179        Ok(SendPay {
2180            created_index: value
2181                .created_index
2182                .ok_or(NodeError::generic("missing created index"))?,
2183            updated_index: value.updated_index,
2184            groupid: value.groupid.to_string(),
2185            partid: value.partid,
2186            payment_hash: value.payment_hash,
2187            status: value.status.try_into()?,
2188            amount_msat: value.amount_msat.map(|a| a.msat),
2189            destination: value.destination,
2190            created_at: value.created_at,
2191            amount_sent_msat: value.amount_sent_msat.map(|a| a.msat),
2192            label: value.label,
2193            bolt11: value.bolt11,
2194            description: value.description,
2195            bolt12: value.bolt12,
2196            payment_preimage: value.payment_preimage,
2197            erroronion: value.erroronion,
2198        })
2199    }
2200}
2201
2202impl TryFrom<i32> for SendPayStatus {
2203    type Error = NodeError;
2204
2205    fn try_from(value: i32) -> std::result::Result<Self, Self::Error> {
2206        match value {
2207            0 => Ok(Self::Pending),
2208            1 => Ok(Self::Failed),
2209            2 => Ok(Self::Complete),
2210            _ => Err(NodeError::generic("invalid send_pay status")),
2211        }
2212    }
2213}
2214
2215impl TryFrom<SendPayAgg> for Payment {
2216    type Error = NodeError;
2217
2218    fn try_from(value: SendPayAgg) -> std::result::Result<Self, Self::Error> {
2219        let ln_invoice = value
2220            .bolt11
2221            .as_ref()
2222            .ok_or(InvoiceError::generic("No bolt11 invoice"))
2223            .and_then(|b| parse_invoice(b));
2224
2225        // For trampoline payments the amount_msat doesn't match the actual
2226        // amount. If it's a trampoline payment, take the amount from the label.
2227        let (payment_amount, client_label) =
2228            serde_json::from_str::<PaymentLabel>(&value.label.clone().unwrap_or_default())
2229                .ok()
2230                .and_then(|label| {
2231                    label
2232                        .trampoline
2233                        .then_some((label.amount_msat, label.client_label))
2234                })
2235                .unwrap_or((value.amount.unwrap_or_default(), value.label));
2236        let fee_msat = value.amount_sent.saturating_sub(payment_amount);
2237        let status = if value.state & PAYMENT_STATE_COMPLETE > 0 {
2238            PaymentStatus::Complete
2239        } else if value.state & PAYMENT_STATE_PENDING > 0 {
2240            PaymentStatus::Pending
2241        } else {
2242            PaymentStatus::Failed
2243        };
2244        Ok(Self {
2245            id: hex::encode(&value.payment_hash),
2246            payment_type: PaymentType::Sent,
2247            payment_time: value.created_at as i64,
2248            amount_msat: match status {
2249                PaymentStatus::Complete => payment_amount,
2250                _ => ln_invoice
2251                    .as_ref()
2252                    .map_or(0, |i| i.amount_msat.unwrap_or_default()),
2253            },
2254            fee_msat,
2255            status,
2256            error: None,
2257            description: ln_invoice
2258                .as_ref()
2259                .map(|i| i.description.clone())
2260                .unwrap_or_default(),
2261            details: PaymentDetails::Ln {
2262                data: LnPaymentDetails {
2263                    payment_hash: hex::encode(&value.payment_hash),
2264                    label: client_label.unwrap_or_default(),
2265                    destination_pubkey: ln_invoice.map_or(
2266                        value.destination.map(hex::encode).unwrap_or_default(),
2267                        |i| i.payee_pubkey,
2268                    ),
2269                    payment_preimage: value.preimage.map(hex::encode).unwrap_or_default(),
2270                    keysend: value.bolt11.is_none(),
2271                    bolt11: value.bolt11.unwrap_or_default(),
2272                    open_channel_bolt11: None,
2273                    lnurl_success_action: None,
2274                    lnurl_pay_domain: None,
2275                    lnurl_pay_comment: None,
2276                    ln_address: None,
2277                    lnurl_metadata: None,
2278                    lnurl_withdraw_endpoint: None,
2279                    swap_info: None,
2280                    reverse_swap_info: None,
2281                    pending_expiration_block: None,
2282                },
2283            },
2284            metadata: None,
2285        })
2286    }
2287}
2288
2289/// Construct a lightning transaction from an invoice
2290impl TryFrom<cln::ListinvoicesInvoices> for Payment {
2291    type Error = NodeError;
2292
2293    fn try_from(invoice: cln::ListinvoicesInvoices) -> std::result::Result<Self, Self::Error> {
2294        let ln_invoice = invoice
2295            .bolt11
2296            .as_ref()
2297            .ok_or(InvoiceError::generic("No bolt11 invoice"))
2298            .and_then(|b| parse_invoice(b))?;
2299        Ok(Payment {
2300            id: hex::encode(invoice.payment_hash.clone()),
2301            payment_type: PaymentType::Received,
2302            payment_time: invoice.paid_at.map(|i| i as i64).unwrap_or_default(),
2303            amount_msat: invoice
2304                .amount_received_msat
2305                .or(invoice.amount_msat)
2306                .map(|a| a.msat)
2307                .unwrap_or_default(),
2308            fee_msat: 0,
2309            status: PaymentStatus::Complete,
2310            error: None,
2311            description: ln_invoice.description,
2312            details: PaymentDetails::Ln {
2313                data: LnPaymentDetails {
2314                    payment_hash: hex::encode(invoice.payment_hash),
2315                    label: invoice.label,
2316                    destination_pubkey: ln_invoice.payee_pubkey,
2317                    payment_preimage: invoice
2318                        .payment_preimage
2319                        .map(hex::encode)
2320                        .unwrap_or_default(),
2321                    keysend: false,
2322                    bolt11: invoice.bolt11.unwrap_or_default(),
2323                    lnurl_success_action: None, // For received payments, this is None
2324                    lnurl_pay_domain: None,     // For received payments, this is None
2325                    lnurl_pay_comment: None,    // For received payments, this is None
2326                    lnurl_metadata: None,       // For received payments, this is None
2327                    ln_address: None,
2328                    lnurl_withdraw_endpoint: None,
2329                    swap_info: None,
2330                    reverse_swap_info: None,
2331                    pending_expiration_block: None,
2332                    open_channel_bolt11: None,
2333                },
2334            },
2335            metadata: None,
2336        })
2337    }
2338}
2339
2340impl From<ListpaysPaysStatus> for PaymentStatus {
2341    fn from(value: ListpaysPaysStatus) -> Self {
2342        match value {
2343            ListpaysPaysStatus::Pending => PaymentStatus::Pending,
2344            ListpaysPaysStatus::Complete => PaymentStatus::Complete,
2345            ListpaysPaysStatus::Failed => PaymentStatus::Failed,
2346        }
2347    }
2348}
2349
2350impl TryFrom<cln::ListpaysPays> for Payment {
2351    type Error = NodeError;
2352
2353    fn try_from(payment: cln::ListpaysPays) -> NodeResult<Self, Self::Error> {
2354        let ln_invoice = payment
2355            .bolt11
2356            .as_ref()
2357            .ok_or(InvoiceError::generic("No bolt11 invoice"))
2358            .and_then(|b| parse_invoice(b));
2359        let payment_amount_sent = payment
2360            .amount_sent_msat
2361            .clone()
2362            .map(|a| a.msat)
2363            .unwrap_or_default();
2364
2365        // For trampoline payments the amount_msat doesn't match the actual
2366        // amount. If it's a trampoline payment, take the amount from the label.
2367        let (payment_amount, client_label) = serde_json::from_str::<PaymentLabel>(payment.label())
2368            .ok()
2369            .and_then(|label| {
2370                label
2371                    .trampoline
2372                    .then_some((label.amount_msat, label.client_label))
2373            })
2374            .unwrap_or((
2375                payment
2376                    .amount_msat
2377                    .clone()
2378                    .map(|a| a.msat)
2379                    .unwrap_or_default(),
2380                payment.label.clone(),
2381            ));
2382        let status = payment.status().into();
2383
2384        Ok(Payment {
2385            id: hex::encode(payment.payment_hash.clone()),
2386            payment_type: PaymentType::Sent,
2387            payment_time: payment.completed_at.unwrap_or(payment.created_at) as i64,
2388            amount_msat: match status {
2389                PaymentStatus::Complete => payment_amount,
2390                _ => ln_invoice
2391                    .as_ref()
2392                    .map_or(0, |i| i.amount_msat.unwrap_or_default()),
2393            },
2394            fee_msat: payment_amount_sent.saturating_sub(payment_amount),
2395            status,
2396            error: None,
2397            description: ln_invoice
2398                .as_ref()
2399                .map(|i| i.description.clone())
2400                .unwrap_or_default(),
2401            details: PaymentDetails::Ln {
2402                data: LnPaymentDetails {
2403                    payment_hash: hex::encode(payment.payment_hash),
2404                    label: client_label.unwrap_or_default(),
2405                    destination_pubkey: ln_invoice.map_or(
2406                        payment.destination.map(hex::encode).unwrap_or_default(),
2407                        |i| i.payee_pubkey,
2408                    ),
2409                    payment_preimage: payment.preimage.map(hex::encode).unwrap_or_default(),
2410                    keysend: payment.bolt11.is_none(),
2411                    bolt11: payment.bolt11.unwrap_or_default(),
2412                    lnurl_success_action: None,
2413                    lnurl_pay_domain: None,
2414                    lnurl_pay_comment: None,
2415                    lnurl_metadata: None,
2416                    ln_address: None,
2417                    lnurl_withdraw_endpoint: None,
2418                    swap_info: None,
2419                    reverse_swap_info: None,
2420                    pending_expiration_block: None,
2421                    open_channel_bolt11: None,
2422                },
2423            },
2424            metadata: None,
2425        })
2426    }
2427}
2428
2429impl TryFrom<cln::PayResponse> for PaymentResponse {
2430    type Error = NodeError;
2431
2432    fn try_from(payment: cln::PayResponse) -> std::result::Result<Self, Self::Error> {
2433        let payment_amount = payment.amount_msat.unwrap_or_default().msat;
2434        let payment_amount_sent = payment.amount_sent_msat.unwrap_or_default().msat;
2435
2436        Ok(PaymentResponse {
2437            payment_time: payment.created_at as i64,
2438            amount_msat: payment_amount,
2439            fee_msat: payment_amount_sent - payment_amount,
2440            payment_hash: hex::encode(payment.payment_hash),
2441            payment_preimage: hex::encode(payment.payment_preimage),
2442        })
2443    }
2444}
2445
2446impl TryFrom<cln::KeysendResponse> for PaymentResponse {
2447    type Error = NodeError;
2448
2449    fn try_from(payment: cln::KeysendResponse) -> std::result::Result<Self, Self::Error> {
2450        let payment_amount = payment.amount_msat.unwrap_or_default().msat;
2451        let payment_amount_sent = payment.amount_sent_msat.unwrap_or_default().msat;
2452
2453        Ok(PaymentResponse {
2454            payment_time: payment.created_at as i64,
2455            amount_msat: payment_amount,
2456            fee_msat: payment_amount_sent - payment_amount,
2457            payment_hash: hex::encode(payment.payment_hash),
2458            payment_preimage: hex::encode(payment.payment_preimage),
2459        })
2460    }
2461}
2462
2463fn amount_to_msat(amount: &gl_client::pb::greenlight::Amount) -> u64 {
2464    match amount.unit {
2465        Some(amount::Unit::Millisatoshi(val)) => val,
2466        Some(amount::Unit::Satoshi(val)) => val * 1000,
2467        Some(amount::Unit::Bitcoin(val)) => val * 100000000,
2468        Some(_) => 0,
2469        None => 0,
2470    }
2471}
2472
2473/// Conversion for an open channel
2474impl From<cln::ListpeerchannelsChannels> for Channel {
2475    fn from(c: cln::ListpeerchannelsChannels) -> Self {
2476        let state = match c.state() {
2477            Openingd | ChanneldAwaitingLockin | DualopendOpenInit | DualopendAwaitingLockin => {
2478                ChannelState::PendingOpen
2479            }
2480            ChanneldNormal => ChannelState::Opened,
2481            _ => ChannelState::PendingClose,
2482        };
2483
2484        let (alias_remote, alias_local) = match c.alias {
2485            Some(a) => (a.remote, a.local),
2486            None => (None, None),
2487        };
2488
2489        Channel {
2490            short_channel_id: c.short_channel_id,
2491            state,
2492            funding_txid: c.funding_txid.map(hex::encode).unwrap_or_default(),
2493            spendable_msat: c.spendable_msat.unwrap_or_default().msat,
2494            local_balance_msat: c.to_us_msat.unwrap_or_default().msat,
2495            receivable_msat: c.receivable_msat.unwrap_or_default().msat,
2496            closed_at: None,
2497            funding_outnum: c.funding_outnum,
2498            alias_remote,
2499            alias_local,
2500            closing_txid: None,
2501            htlcs: c
2502                .htlcs
2503                .into_iter()
2504                .map(|c| Htlc::from(c.expiry.unwrap_or(0), c.payment_hash.unwrap_or_default()))
2505                .collect(),
2506        }
2507    }
2508}
2509
2510fn convert_to_send_pay_route(
2511    route: PaymentPath,
2512    to_pay_msat: u64,
2513    final_cltv_delta: u64,
2514) -> (Vec<SendpayRoute>, u64) {
2515    let mut sendpay_route = vec![];
2516    let mut to_forward = to_pay_msat;
2517    let mut cltv_delay = 0;
2518    let hops_arr = route.edges.as_slice();
2519
2520    let reverse_hops: Vec<&PaymentPathEdge> = hops_arr.iter().rev().collect();
2521
2522    // Iterating over the path in a reverse order so we can calculate
2523    // the cltv deltas and fees.
2524    for (reverse_index, hop) in reverse_hops.iter().enumerate() {
2525        //let hop = h.clone();
2526        (to_forward, cltv_delay) = match reverse_index == 0 {
2527            // last hop should not take any fees and should use the final_cltv_delta.
2528            true => (to_forward, final_cltv_delta),
2529
2530            // all other hops are forwarding therefore should take fees and increase the cltv delta.
2531            false => (
2532                reverse_hops[reverse_index - 1].amount_from_forward(to_forward),
2533                cltv_delay + reverse_hops[reverse_index - 1].channel_delay,
2534            ),
2535        };
2536
2537        sendpay_route.insert(
2538            0,
2539            SendpayRoute {
2540                amount_msat: Some(gl_client::pb::cln::Amount { msat: to_forward }),
2541                id: hop.node_id.clone(),
2542                delay: cltv_delay as u32,
2543                channel: hop.short_channel_id.clone(),
2544            },
2545        );
2546    }
2547
2548    (sendpay_route, to_forward)
2549}
2550
2551impl TryFrom<ListclosedchannelsClosedchannels> for Channel {
2552    type Error = NodeError;
2553
2554    fn try_from(
2555        c: cln::ListclosedchannelsClosedchannels,
2556    ) -> std::result::Result<Self, Self::Error> {
2557        let (alias_remote, alias_local) = match c.alias {
2558            Some(a) => (a.remote, a.local),
2559            None => (None, None),
2560        };
2561
2562        // To keep the conversion simple and fast, some closing-related fields (closed_at, closing_txid)
2563        // are left empty here in the conversion, but populated later (via chain service lookup, or DB lookup)
2564        let local_balance_msat = c
2565            .final_to_us_msat
2566            .ok_or(anyhow!("final_to_us_msat is missing"))?
2567            .msat;
2568        Ok(Channel {
2569            short_channel_id: c.short_channel_id,
2570            state: ChannelState::Closed,
2571            funding_txid: hex::encode(c.funding_txid),
2572            spendable_msat: local_balance_msat,
2573            local_balance_msat,
2574            receivable_msat: 0,
2575            closed_at: None,
2576            funding_outnum: Some(c.funding_outnum),
2577            alias_remote,
2578            alias_local,
2579            closing_txid: None,
2580            htlcs: Vec::new(),
2581        })
2582    }
2583}
2584
2585#[cfg(test)]
2586mod tests {
2587    use anyhow::Result;
2588    use gl_client::pb::cln::listpeerchannels_channels::{
2589        ListpeerchannelsChannelsState, ListpeerchannelsChannelsState::*,
2590    };
2591    use gl_client::pb::cln::Amount;
2592    use gl_client::pb::{self, cln};
2593
2594    use crate::greenlight::node_api::convert_to_send_pay_route;
2595    use crate::{models, PaymentPath, PaymentPathEdge};
2596
2597    #[test]
2598    fn test_convert_route() -> Result<()> {
2599        let path = PaymentPath {
2600            edges: vec![
2601                PaymentPathEdge {
2602                    node_id: vec![1],
2603                    short_channel_id: "807189x2048x0".into(),
2604                    channel_delay: 34,
2605                    base_fee_msat: 1000,
2606                    fee_per_millionth: 10,
2607                },
2608                PaymentPathEdge {
2609                    node_id: vec![2],
2610                    short_channel_id: "811871x2726x1".into(),
2611                    channel_delay: 34,
2612                    base_fee_msat: 0,
2613                    fee_per_millionth: 0,
2614                },
2615                PaymentPathEdge {
2616                    node_id: vec![3],
2617                    short_channel_id: "16000000x0x18087".into(),
2618                    channel_delay: 40,
2619                    base_fee_msat: 1000,
2620                    fee_per_millionth: 1,
2621                },
2622            ],
2623        };
2624
2625        let (r, sent) = convert_to_send_pay_route(path, 50000000, 144);
2626        assert_eq!(
2627            r,
2628            vec![
2629                pb::cln::SendpayRoute {
2630                    amount_msat: Some(gl_client::pb::cln::Amount { msat: 50001050 }),
2631                    id: vec![1],
2632                    delay: 218,
2633                    channel: "807189x2048x0".into(),
2634                },
2635                pb::cln::SendpayRoute {
2636                    amount_msat: Some(gl_client::pb::cln::Amount { msat: 50001050 }),
2637                    id: vec![2],
2638                    delay: 184,
2639                    channel: "811871x2726x1".into(),
2640                },
2641                pb::cln::SendpayRoute {
2642                    amount_msat: Some(gl_client::pb::cln::Amount { msat: 50000000 }),
2643                    id: vec![3],
2644                    delay: 144,
2645                    channel: "16000000x0x18087".into(),
2646                }
2647            ]
2648        );
2649        assert_eq!(sent, 50001050);
2650
2651        let path = PaymentPath {
2652            edges: vec![
2653                PaymentPathEdge {
2654                    node_id: vec![1],
2655                    short_channel_id: "807189x2048x0".into(),
2656                    channel_delay: 34,
2657                    base_fee_msat: 1000,
2658                    fee_per_millionth: 10,
2659                },
2660                PaymentPathEdge {
2661                    node_id: vec![2],
2662                    short_channel_id: "811871x2726x1".into(),
2663                    channel_delay: 34,
2664                    base_fee_msat: 0,
2665                    fee_per_millionth: 0,
2666                },
2667                PaymentPathEdge {
2668                    node_id: vec![3],
2669                    short_channel_id: "16000000x0x18087".into(),
2670                    channel_delay: 40,
2671                    base_fee_msat: 0,
2672                    fee_per_millionth: 2000,
2673                },
2674            ],
2675        };
2676        let (r, sent) = convert_to_send_pay_route(path, 50000000, 144);
2677        assert_eq!(
2678            r,
2679            vec![
2680                pb::cln::SendpayRoute {
2681                    amount_msat: Some(gl_client::pb::cln::Amount { msat: 50100000 }),
2682                    id: vec![1],
2683                    delay: 218,
2684                    channel: "807189x2048x0".into(),
2685                },
2686                pb::cln::SendpayRoute {
2687                    amount_msat: Some(gl_client::pb::cln::Amount { msat: 50100000 }),
2688                    id: vec![2],
2689                    delay: 184,
2690                    channel: "811871x2726x1".into(),
2691                },
2692                pb::cln::SendpayRoute {
2693                    amount_msat: Some(gl_client::pb::cln::Amount { msat: 50000000 }),
2694                    id: vec![3],
2695                    delay: 144,
2696                    channel: "16000000x0x18087".into(),
2697                }
2698            ]
2699        );
2700        assert_eq!(sent, 50100000);
2701
2702        Ok(())
2703    }
2704
2705    #[test]
2706    fn test_channel_states() -> Result<()> {
2707        for s in &[Openingd, ChanneldAwaitingLockin] {
2708            let c: models::Channel = cln_channel(s).into();
2709            assert_eq!(c.state, models::ChannelState::PendingOpen);
2710        }
2711
2712        let s = ChanneldNormal;
2713        let c: models::Channel = cln_channel(&s).into();
2714        assert_eq!(c.state, models::ChannelState::Opened);
2715
2716        for s in &[
2717            ChanneldShuttingDown,
2718            ClosingdSigexchange,
2719            ClosingdComplete,
2720            AwaitingUnilateral,
2721            FundingSpendSeen,
2722        ] {
2723            let c: models::Channel = cln_channel(s).into();
2724            assert_eq!(c.state, models::ChannelState::PendingClose);
2725        }
2726
2727        let c: models::Channel = cln_channel(&Onchain).into();
2728        assert_eq!(c.state, models::ChannelState::PendingClose);
2729
2730        Ok(())
2731    }
2732
2733    fn cln_channel(state: &ListpeerchannelsChannelsState) -> cln::ListpeerchannelsChannels {
2734        cln::ListpeerchannelsChannels {
2735            state: Some((*state).into()),
2736            scratch_txid: None,
2737            feerate: None,
2738            owner: None,
2739            short_channel_id: None,
2740            channel_id: None,
2741            funding_txid: None,
2742            funding_outnum: None,
2743            initial_feerate: None,
2744            last_feerate: None,
2745            next_feerate: None,
2746            next_fee_step: None,
2747            inflight: vec![],
2748            close_to: None,
2749            private: Some(true),
2750            opener: Some(0),
2751            closer: None,
2752            funding: None,
2753            to_us_msat: None,
2754            min_to_us_msat: None,
2755            max_to_us_msat: None,
2756            total_msat: Some(Amount { msat: 1_000 }),
2757            fee_base_msat: None,
2758            fee_proportional_millionths: None,
2759            dust_limit_msat: Some(Amount { msat: 10 }),
2760            max_total_htlc_in_msat: None,
2761            their_reserve_msat: None,
2762            our_reserve_msat: None,
2763            spendable_msat: Some(Amount { msat: 20_000 }),
2764            receivable_msat: Some(Amount { msat: 960_000 }),
2765            minimum_htlc_in_msat: None,
2766            minimum_htlc_out_msat: None,
2767            maximum_htlc_out_msat: None,
2768            their_to_self_delay: Some(144),
2769            our_to_self_delay: Some(144),
2770            max_accepted_htlcs: None,
2771            alias: None,
2772            status: vec![],
2773            in_payments_offered: None,
2774            in_offered_msat: None,
2775            in_payments_fulfilled: None,
2776            in_fulfilled_msat: None,
2777            out_payments_offered: None,
2778            out_offered_msat: None,
2779            out_payments_fulfilled: None,
2780            out_fulfilled_msat: None,
2781            htlcs: vec![],
2782            close_to_addr: None,
2783            peer_id: None,
2784            peer_connected: None,
2785            updates: None,
2786            ignore_fee_limits: None,
2787            lost_state: None,
2788            last_stable_connection: None,
2789        }
2790    }
2791}