1#![doc = include_str!("../README.md")]
4
5use std::fmt;
6use std::net::SocketAddr;
7use std::pin::Pin;
8use std::sync::atomic::{AtomicBool, Ordering};
9use std::sync::Arc;
10use std::time::Duration;
11
12use async_trait::async_trait;
13use bip39::Mnemonic;
14use cdk_common::amount::MSAT_IN_SAT;
15use cdk_common::common::FeeReserve;
16use cdk_common::database::DynKVStore;
17use cdk_common::payment::{self, *};
18use cdk_common::redact::url_for_logs;
19use cdk_common::util::{hex, unix_time};
20use cdk_common::{Amount, CurrencyUnit, MeltOptions, MeltQuoteState, QuoteId};
21use futures::{Stream, StreamExt};
22use ldk_node::bitcoin::hashes::Hash;
23use ldk_node::bitcoin::Network;
24use ldk_node::lightning::ln::channelmanager::PaymentId;
25use ldk_node::lightning::ln::msgs::SocketAddress;
26use ldk_node::lightning::routing::router::RouteParametersConfig;
27use ldk_node::lightning_invoice::{Bolt11InvoiceDescription, Description};
28use ldk_node::lightning_types::payment::PaymentHash;
29use ldk_node::logger::{LogLevel, LogWriter};
30use ldk_node::payment::{PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus};
31use ldk_node::{Builder, Event, Node};
32use tokio_stream::wrappers::BroadcastStream;
33use tokio_util::sync::CancellationToken;
34use tracing::instrument;
35
36use crate::error::Error;
37use crate::log::StdoutLogWriter;
38
39mod error;
40mod log;
41mod web;
42
43const LDK_KV_PRIMARY_NAMESPACE: &str = "cdk_ldk_node_lightning_backend";
45const LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE: &str = "bolt12_outgoing_payments";
48const PAYMENT_WAIT_TIMEOUT: Duration = Duration::from_secs(10);
50const PAYMENT_EVENT_CHANNEL_CAPACITY: usize = 64;
52const LDK_KV_BOLT12_CLEANUP_MARKER: &[u8] = b"cleanup-in-progress";
53
54#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56enum Bolt12QuotePaymentIdLookup {
57 Found(PaymentId),
59 Dispatching,
63 Missing,
65 Malformed,
67}
68
69#[derive(Debug, Clone, Copy, PartialEq, Eq)]
70enum Bolt12QuotePaymentIdResolution {
71 PaymentId(PaymentId),
72 Status(MeltQuoteState),
73}
74
75impl Bolt12QuotePaymentIdLookup {
76 fn resolve(self) -> Bolt12QuotePaymentIdResolution {
77 match self {
78 Self::Found(payment_id) => Bolt12QuotePaymentIdResolution::PaymentId(payment_id),
79 Self::Dispatching => Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Pending),
83 Self::Missing => Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Unpaid),
85 Self::Malformed => Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Unknown),
87 }
88 }
89}
90
91fn bolt12_send_error_has_ambiguous_dispatch(err: &ldk_node::NodeError) -> bool {
98 matches!(err, ldk_node::NodeError::PersistenceFailed)
99}
100
101fn bolt11_send_error_is_explicit_terminal_failure(err: &ldk_node::NodeError) -> bool {
110 matches!(
111 err,
112 ldk_node::NodeError::NotRunning
113 | ldk_node::NodeError::InvalidAmount
114 | ldk_node::NodeError::InvalidInvoice
115 | ldk_node::NodeError::PaymentSendingFailed
116 )
117}
118
119fn outgoing_payment_failure_response(
120 unit: &CurrencyUnit,
121 payment_lookup_id: PaymentIdentifier,
122) -> MakePaymentResponse {
123 MakePaymentResponse {
124 payment_lookup_id,
125 payment_proof: None,
126 status: MeltQuoteState::Failed,
127 total_spent: Amount::new(0, unit.clone()),
128 }
129}
130
131#[derive(Clone)]
136pub struct CdkLdkNode {
137 inner: Arc<Node>,
138 fee_reserve: FeeReserve,
139 kv_store: DynKVStore,
140 wait_invoice_cancel_token: CancellationToken,
141 wait_invoice_is_active: Arc<AtomicBool>,
142 sender: tokio::sync::broadcast::Sender<WaitPaymentResponse>,
143 receiver: Arc<tokio::sync::broadcast::Receiver<WaitPaymentResponse>>,
144 outgoing_payment_sender: tokio::sync::broadcast::Sender<PaymentId>,
145 events_cancel_token: CancellationToken,
146 web_addr: Option<SocketAddr>,
147}
148
149impl fmt::Debug for CdkLdkNode {
150 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
151 f.debug_struct("CdkLdkNode")
152 .field("fee_reserve", &self.fee_reserve)
153 .field("web_addr", &self.web_addr)
154 .finish_non_exhaustive()
155 }
156}
157
158#[derive(Clone)]
162pub struct BitcoinRpcConfig {
163 pub host: String,
165 pub port: u16,
167 pub user: String,
169 pub password: String,
171}
172
173impl fmt::Debug for BitcoinRpcConfig {
174 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
175 f.debug_struct("BitcoinRpcConfig")
176 .field("host", &self.host)
177 .field("port", &self.port)
178 .field("user", &self.user)
179 .field("password", &"[REDACTED]")
180 .finish()
181 }
182}
183
184#[derive(Clone)]
189pub enum ChainSource {
190 Esplora(String),
194 Electrum(String),
198 BitcoinRpc(BitcoinRpcConfig),
202}
203
204impl fmt::Debug for ChainSource {
205 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
206 match self {
207 Self::Esplora(url) => f.debug_tuple("Esplora").field(&url_for_logs(url)).finish(),
208 Self::Electrum(url) => f.debug_tuple("Electrum").field(&url_for_logs(url)).finish(),
209 Self::BitcoinRpc(config) => f.debug_tuple("BitcoinRpc").field(config).finish(),
210 }
211 }
212}
213
214#[derive(Clone)]
219pub enum GossipSource {
220 P2P,
224 RapidGossipSync(String),
228}
229
230impl fmt::Debug for GossipSource {
231 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
232 match self {
233 Self::P2P => f.write_str("P2P"),
234 Self::RapidGossipSync(url) => f
235 .debug_tuple("RapidGossipSync")
236 .field(&url_for_logs(url))
237 .finish(),
238 }
239 }
240}
241pub struct CdkLdkNodeBuilder {
243 network: Network,
244 chain_source: ChainSource,
245 gossip_source: GossipSource,
246 log_dir_path: Option<String>,
247 storage_dir_path: String,
248 fee_reserve: FeeReserve,
249 kv_store: DynKVStore,
250 listening_addresses: Vec<SocketAddress>,
251 seed: Option<Mnemonic>,
252 announcement_addresses: Option<Vec<SocketAddress>>,
253}
254
255impl std::fmt::Debug for CdkLdkNodeBuilder {
256 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
257 f.debug_struct("CdkLdkNodeBuilder")
258 .field("network", &self.network)
259 .field("chain_source", &self.chain_source)
260 .field("gossip_source", &self.gossip_source)
261 .field("log_dir_path", &self.log_dir_path)
262 .field("storage_dir_path", &self.storage_dir_path)
263 .field("fee_reserve", &self.fee_reserve)
264 .field("listening_addresses", &self.listening_addresses)
265 .field("announcement_addresses", &self.announcement_addresses)
266 .finish_non_exhaustive()
267 }
268}
269
270impl CdkLdkNodeBuilder {
271 pub fn new(
273 network: Network,
274 chain_source: ChainSource,
275 gossip_source: GossipSource,
276 storage_dir_path: String,
277 fee_reserve: FeeReserve,
278 listening_addresses: Vec<SocketAddress>,
279 kv_store: DynKVStore,
280 ) -> Self {
281 Self {
282 network,
283 chain_source,
284 gossip_source,
285 storage_dir_path,
286 fee_reserve,
287 kv_store,
288 listening_addresses,
289 seed: None,
290 announcement_addresses: None,
291 log_dir_path: None,
292 }
293 }
294
295 pub fn with_seed(mut self, seed: Mnemonic) -> Self {
297 self.seed = Some(seed);
298 self
299 }
300 pub fn with_announcement_address(mut self, announcement_addresses: Vec<SocketAddress>) -> Self {
302 self.announcement_addresses = Some(announcement_addresses);
303 self
304 }
305 pub fn with_log_dir_path(mut self, log_dir_path: String) -> Self {
307 self.log_dir_path = Some(log_dir_path);
308 self
309 }
310
311 pub fn build(self) -> Result<CdkLdkNode, Error> {
316 let mut ldk = Builder::new();
317 ldk.set_network(self.network);
318 tracing::info!("Storage dir of node is {}", self.storage_dir_path);
319 ldk.set_storage_dir_path(self.storage_dir_path);
320
321 match self.chain_source {
322 ChainSource::Esplora(esplora_url) => {
323 ldk.set_chain_source_esplora(esplora_url, None);
324 }
325 ChainSource::Electrum(electrum_url) => {
326 ldk.set_chain_source_electrum(electrum_url, None);
327 }
328 ChainSource::BitcoinRpc(BitcoinRpcConfig {
329 host,
330 port,
331 user,
332 password,
333 }) => {
334 ldk.set_chain_source_bitcoind_rpc(host, port, user, password);
335 }
336 }
337
338 match self.gossip_source {
339 GossipSource::P2P => {
340 ldk.set_gossip_source_p2p();
341 }
342 GossipSource::RapidGossipSync(rgs_url) => {
343 ldk.set_gossip_source_rgs(rgs_url);
344 }
345 }
346
347 ldk.set_listening_addresses(self.listening_addresses)?;
348 if self.log_dir_path.is_some() {
349 ldk.set_filesystem_logger(self.log_dir_path, Some(LogLevel::Info));
350 } else {
351 ldk.set_custom_logger(Arc::new(StdoutLogWriter));
352 }
353
354 ldk.set_node_alias("cdk-ldk-node".to_string())?;
355 if let Some(seed) = self.seed {
357 ldk.set_entropy_bip39_mnemonic(seed, None);
358 }
359 if let Some(announcement_addresses) = self.announcement_addresses {
361 ldk.set_announcement_addresses(announcement_addresses)?;
362 }
363
364 let node = ldk.build()?;
365
366 tracing::info!("Creating tokio channel for payment notifications");
367 let (sender, receiver) = tokio::sync::broadcast::channel(8);
368 let (outgoing_payment_sender, _) =
369 tokio::sync::broadcast::channel(PAYMENT_EVENT_CHANNEL_CAPACITY);
370
371 let id = node.node_id();
372
373 let adr = node.announcement_addresses();
374
375 tracing::info!(
376 "Created node {} with address {:?} on network {}",
377 id,
378 adr,
379 self.network
380 );
381
382 Ok(CdkLdkNode {
383 inner: node.into(),
384 fee_reserve: self.fee_reserve,
385 kv_store: self.kv_store,
386 wait_invoice_cancel_token: CancellationToken::new(),
387 wait_invoice_is_active: Arc::new(AtomicBool::new(false)),
388 sender,
389 receiver: Arc::new(receiver),
390 outgoing_payment_sender,
391 events_cancel_token: CancellationToken::new(),
392 web_addr: None,
393 })
394 }
395}
396
397impl CdkLdkNode {
398 pub fn set_web_addr(&mut self, addr: Option<SocketAddr>) {
403 self.web_addr = addr;
404 }
405
406 pub fn default_web_addr() -> SocketAddr {
411 SocketAddr::from(([127, 0, 0, 1], 8091))
412 }
413
414 async fn cleanup_bolt12_dispatch_binding(
417 &self,
418 quote_id: &QuoteId,
419 payment_id: Option<&PaymentId>,
420 ) {
421 match delete_bolt12_quote_payment_id_if_equals(&self.kv_store, quote_id, payment_id).await {
422 Ok(true) => {}
423 Ok(false) => {
424 tracing::debug!(
425 quote_id = %quote_id,
426 "BOLT12 dispatch binding changed before cleanup"
427 );
428 }
429 Err(err) => {
430 tracing::warn!(
431 quote_id = %quote_id,
432 "Could not release BOLT12 dispatch binding: {err}"
433 );
434 }
435 }
436 }
437
438 fn make_payment_response_from_details(
439 unit: &CurrencyUnit,
440 payment_lookup_id: PaymentIdentifier,
441 payment_details: &PaymentDetails,
442 ) -> Result<MakePaymentResponse, payment::Error> {
443 let status = match payment_details.status {
444 PaymentStatus::Pending => MeltQuoteState::Pending,
445 PaymentStatus::Succeeded => MeltQuoteState::Paid,
446 PaymentStatus::Failed => MeltQuoteState::Failed,
447 };
448
449 let payment_proof = match &payment_details.kind {
450 PaymentKind::Bolt11 { preimage, .. } => preimage.map(|p| p.to_string()),
451 PaymentKind::Bolt12Offer { preimage, .. } => preimage.map(|p| p.to_string()),
452 _ => return Err(Error::UnexpectedPaymentKind.into()),
453 };
454
455 let total_spent = if status == MeltQuoteState::Paid {
456 let total_spent = payment_details
457 .amount_msat
458 .ok_or(Error::CouldNotGetAmountSpent)?
459 + payment_details.fee_paid_msat.unwrap_or_default();
460 match unit {
462 CurrencyUnit::Sat => {
463 Amount::new(total_spent.div_ceil(MSAT_IN_SAT), CurrencyUnit::Sat)
464 }
465 _ => Amount::new(total_spent, CurrencyUnit::Msat).convert_to(unit)?,
466 }
467 } else {
468 Amount::new(0, unit.clone())
469 };
470
471 Ok(MakePaymentResponse {
472 payment_lookup_id,
473 payment_proof,
474 status,
475 total_spent,
476 })
477 }
478
479 fn select_bolt11_payment_details(
480 payment_details: impl IntoIterator<Item = PaymentDetails>,
481 ) -> Option<PaymentDetails> {
482 payment_details.into_iter().min_by_key(|details| {
483 let status_order = match details.status {
484 PaymentStatus::Succeeded => 0_u8,
485 PaymentStatus::Pending => 1,
486 PaymentStatus::Failed => 2,
487 };
488
489 (
490 status_order,
491 std::cmp::Reverse(details.latest_update_timestamp),
492 )
493 })
494 }
495
496 async fn wait_for_terminal_payment_event(
497 receiver: &mut tokio::sync::broadcast::Receiver<PaymentId>,
498 payment_id: PaymentId,
499 ) -> Result<(), tokio::sync::broadcast::error::RecvError> {
500 loop {
501 match receiver.recv().await {
502 Ok(completed_payment_id) if completed_payment_id == payment_id => return Ok(()),
503 Ok(_) => continue,
504 Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
505 tracing::warn!(
506 payment_id = %payment_id,
507 skipped,
508 "Terminal payment event receiver lagged; continuing to wait"
509 );
510 }
511 Err(err) => return Err(err),
512 }
513 }
514 }
515
516 async fn wait_for_payment_terminal_status(
517 &self,
518 payment_id: PaymentId,
519 mut receiver: tokio::sync::broadcast::Receiver<PaymentId>,
520 ) -> Result<PaymentDetails, payment::Error> {
521 let payment_details = self
522 .inner
523 .payment(&payment_id)
524 .ok_or(Error::PaymentNotFound)?;
525
526 if payment_details.status != PaymentStatus::Pending {
527 return Ok(payment_details);
528 }
529
530 match tokio::time::timeout(
531 PAYMENT_WAIT_TIMEOUT,
532 Self::wait_for_terminal_payment_event(&mut receiver, payment_id),
533 )
534 .await
535 {
536 Ok(Ok(())) => {}
537 Ok(Err(err)) => {
538 tracing::warn!(
539 payment_id = %payment_id,
540 "Could not wait for terminal LDK payment event: {err}"
541 );
542 }
543 Err(_) => {
544 tracing::warn!(
545 payment_id = %payment_id,
546 "Payment did not reach a terminal state within {} seconds",
547 PAYMENT_WAIT_TIMEOUT.as_secs()
548 );
549 }
550 }
551
552 let payment_details = self
553 .inner
554 .payment(&payment_id)
555 .ok_or(Error::PaymentNotFound)?;
556
557 if payment_details.status == PaymentStatus::Pending {
558 tracing::debug!(
559 payment_id = %payment_id,
560 "Payment remains pending after waiting for a terminal event"
561 );
562 }
563
564 Ok(payment_details)
565 }
566
567 pub fn start_ldk_node(&self) -> Result<(), Error> {
578 tracing::info!("Starting cdk-ldk node");
579 self.inner.start()?;
580 let node_config = self.inner.config();
581
582 tracing::info!("Starting node with network {}", node_config.network);
583
584 tracing::info!("Node status: {:?}", self.inner.status());
585
586 self.handle_events()?;
587
588 Ok(())
589 }
590
591 pub fn start_web_server(&self, web_addr: SocketAddr) -> Result<(), Error> {
606 let web_server = crate::web::WebServer::new(Arc::new(self.clone()));
607
608 tokio::spawn(async move {
609 if let Err(e) = web_server.serve(web_addr).await {
610 tracing::error!("Web server error: {}", e);
611 }
612 });
613
614 Ok(())
615 }
616
617 pub fn stop_ldk_node(&self) -> Result<(), Error> {
631 tracing::info!("Stopping CdkLdkNode");
632 tracing::info!("Cancelling event handler");
634 self.events_cancel_token.cancel();
635
636 if self.is_payment_event_stream_active() {
638 tracing::info!("Cancelling payment event stream");
639 self.wait_invoice_cancel_token.cancel();
640 }
641
642 tracing::info!("Stopping LDK node");
644 self.inner.stop()?;
645 tracing::info!("CdkLdkNode stopped successfully");
646 Ok(())
647 }
648
649 async fn handle_payment_received(
651 node: &Arc<Node>,
652 sender: &tokio::sync::broadcast::Sender<WaitPaymentResponse>,
653 payment_id: Option<PaymentId>,
654 payment_hash: PaymentHash,
655 amount_msat: u64,
656 ) {
657 tracing::info!(
658 "Received payment for hash={} of amount={} msat",
659 payment_hash,
660 amount_msat
661 );
662
663 let payment_id = match payment_id {
664 Some(id) => id,
665 None => {
666 tracing::warn!("Received payment without payment_id");
667 return;
668 }
669 };
670
671 let payment_id_hex = hex::encode(payment_id.0);
672
673 if amount_msat == 0 {
674 tracing::warn!("Payment of no amount");
675 return;
676 }
677
678 tracing::info!(
679 "Processing payment notification: id={}, amount={} msats",
680 payment_id_hex,
681 amount_msat
682 );
683
684 let payment_details = match node.payment(&payment_id) {
685 Some(details) => details,
686 None => {
687 tracing::error!("Could not find payment details for id={}", payment_id_hex);
688 return;
689 }
690 };
691
692 let (payment_identifier, payment_id) = match payment_details.kind {
693 PaymentKind::Bolt11 { hash, .. } => {
694 (PaymentIdentifier::PaymentHash(hash.0), hash.to_string())
695 }
696 PaymentKind::Bolt12Offer { hash, offer_id, .. } => match hash {
697 Some(h) => (
698 PaymentIdentifier::OfferId(offer_id.to_string()),
699 h.to_string(),
700 ),
701 None => {
702 tracing::error!("Bolt12 payment missing hash");
703 return;
704 }
705 },
706 k => {
707 tracing::warn!("Received payment of kind {:?} which is not supported", k);
708 return;
709 }
710 };
711
712 let wait_payment_response = WaitPaymentResponse {
713 payment_identifier,
714 payment_amount: Amount::new(amount_msat, CurrencyUnit::Msat),
715 payment_id,
716 };
717
718 match sender.send(wait_payment_response) {
719 Ok(_) => tracing::info!("Successfully sent payment notification to stream"),
720 Err(err) => tracing::error!(
721 "Could not send payment received notification on channel: {}",
722 err
723 ),
724 }
725 }
726
727 pub fn handle_events(&self) -> Result<(), Error> {
729 let node = self.inner.clone();
730 let sender = self.sender.clone();
731 let outgoing_payment_sender = self.outgoing_payment_sender.clone();
732 let cancel_token = self.events_cancel_token.clone();
733
734 tracing::info!("Starting event handler task");
735
736 tokio::spawn(async move {
737 tracing::info!("Event handler loop started");
738 loop {
739 tokio::select! {
740 _ = cancel_token.cancelled() => {
741 tracing::info!("Event handler cancelled");
742 break;
743 }
744 event = node.next_event_async() => {
745 match event {
746 Event::PaymentReceived {
747 payment_id,
748 payment_hash,
749 amount_msat,
750 custom_records: _
751 } => {
752 Self::handle_payment_received(
753 &node,
754 &sender,
755 payment_id,
756 payment_hash,
757 amount_msat
758 ).await;
759 }
760 Event::PaymentSuccessful {
761 payment_id,
762 payment_hash,
763 payment_preimage: _,
764 fee_paid_msat: _,
765 } => {
766 tracing::info!(
767 payment_id = ?payment_id,
768 payment_hash = %payment_hash,
769 "LDK node payment succeeded"
770 );
771 if let Some(payment_id) = payment_id {
772 let _ = outgoing_payment_sender.send(payment_id);
773 }
774 }
775 Event::PaymentFailed {
776 payment_id,
777 payment_hash,
778 reason,
779 } => {
780 tracing::error!(
781 payment_id = ?payment_id,
782 payment_hash = ?payment_hash,
783 reason = ?reason,
784 "LDK node payment failed"
785 );
786 if let Some(payment_id) = payment_id {
787 let _ = outgoing_payment_sender.send(payment_id);
788 }
789 }
790 event => {
791 tracing::debug!("Received other ldk node event: {:?}", event);
792 }
793 }
794
795 if let Err(err) = node.event_handled() {
796 tracing::error!("Error handling node event: {}", err);
797 } else {
798 tracing::debug!("Successfully handled node event");
799 }
800 }
801 }
802 }
803 tracing::info!("Event handler loop terminated");
804 });
805
806 tracing::info!("Event handler task spawned");
807 Ok(())
808 }
809
810 pub fn node(&self) -> Arc<Node> {
812 Arc::clone(&self.inner)
813 }
814}
815
816#[async_trait]
818impl MintPayment for CdkLdkNode {
819 type Err = payment::Error;
820
821 async fn start(&self) -> Result<(), Self::Err> {
824 self.start_ldk_node().map_err(|e| {
825 tracing::error!("Failed to start CdkLdkNode: {}", e);
826 e
827 })?;
828
829 tracing::info!("CdkLdkNode payment processor started successfully");
830
831 if let Some(web_addr) = self.web_addr {
833 tracing::info!("Starting LDK Node web interface on {}", web_addr);
834 self.start_web_server(web_addr).map_err(|e| {
835 tracing::error!("Failed to start web server: {}", e);
836 e
837 })?;
838 } else {
839 tracing::info!("No web server address configured, skipping web interface");
840 }
841
842 Ok(())
843 }
844
845 async fn stop(&self) -> Result<(), Self::Err> {
848 self.stop_ldk_node().map_err(|e| {
849 tracing::error!("Failed to stop CdkLdkNode: {}", e);
850 e.into()
851 })
852 }
853
854 async fn get_settings(&self) -> Result<SettingsResponse, Self::Err> {
856 let settings = SettingsResponse {
857 unit: CurrencyUnit::Msat.to_string(),
858 bolt11: Some(payment::Bolt11Settings {
859 mpp: false,
860 amountless: true,
861 invoice_description: true,
862 }),
863 bolt12: Some(payment::Bolt12Settings {
864 amountless: true,
865 invoice_description: true,
866 }),
867 onchain: None,
868 custom: std::collections::HashMap::new(),
869 };
870 Ok(settings)
871 }
872
873 #[instrument(skip(self))]
875 async fn create_incoming_payment_request(
876 &self,
877 options: IncomingPaymentOptions,
878 ) -> Result<CreateIncomingPaymentResponse, Self::Err> {
879 match options {
880 IncomingPaymentOptions::Bolt11(bolt11_options) => {
881 let amount_msat: Amount = bolt11_options
882 .amount
883 .convert_to(&CurrencyUnit::Msat)?
884 .into();
885 let description = bolt11_options.description.unwrap_or_default();
886 let time = match bolt11_options.unix_expiry {
887 Some(t) => t
888 .checked_sub(unix_time())
889 .ok_or(payment::Error::InvalidExpiry)?,
890 None => 36000,
891 };
892
893 let description = Bolt11InvoiceDescription::Direct(
894 Description::new(description).map_err(|_| Error::InvalidDescription)?,
895 );
896
897 let payment = self
898 .inner
899 .bolt11_payment()
900 .receive(amount_msat.into(), &description, time as u32)
901 .map_err(Error::LdkNode)?;
902
903 let payment_hash = payment.payment_hash().to_string();
904 let payment_identifier = PaymentIdentifier::PaymentHash(
905 hex::decode(&payment_hash)?
906 .try_into()
907 .map_err(|_| Error::InvalidPaymentHashLength)?,
908 );
909
910 Ok(CreateIncomingPaymentResponse {
911 request_lookup_id: payment_identifier,
912 request: payment.to_string(),
913 expiry: Some(unix_time() + time),
914 extra_json: None,
915 })
916 }
917 IncomingPaymentOptions::Bolt12(bolt12_options) => {
918 let Bolt12IncomingPaymentOptions {
919 description,
920 amount,
921 unix_expiry,
922 } = *bolt12_options;
923
924 let time = unix_expiry
925 .map(|t| {
926 t.checked_sub(unix_time())
927 .ok_or(payment::Error::InvalidExpiry)
928 .map(|t| t as u32)
929 })
930 .transpose()?;
931
932 let offer = match amount {
933 Some(amount) => {
934 let amount_msat: Amount = amount.convert_to(&CurrencyUnit::Msat)?.into();
935
936 self.inner
937 .bolt12_payment()
938 .receive(
939 amount_msat.into(),
940 &description.unwrap_or("".to_string()),
941 time,
942 None,
943 )
944 .map_err(Error::LdkNode)?
945 }
946 None => self
947 .inner
948 .bolt12_payment()
949 .receive_variable_amount(&description.unwrap_or("".to_string()), time)
950 .map_err(Error::LdkNode)?,
951 };
952 let payment_identifier = PaymentIdentifier::OfferId(offer.id().to_string());
953
954 Ok(CreateIncomingPaymentResponse {
955 request_lookup_id: payment_identifier,
956 request: offer.to_string(),
957 expiry: unix_expiry,
958 extra_json: None,
959 })
960 }
961 IncomingPaymentOptions::Custom(_) | IncomingPaymentOptions::Onchain(_) => {
962 Err(cdk_common::payment::Error::UnsupportedPaymentOption)
963 }
964 }
965 }
966
967 #[instrument(skip_all)]
970 async fn get_payment_quote(
971 &self,
972 unit: &CurrencyUnit,
973 options: OutgoingPaymentOptions,
974 ) -> Result<PaymentQuoteResponse, Self::Err> {
975 match options {
976 cdk_common::payment::OutgoingPaymentOptions::Custom(_) => {
977 Err(cdk_common::payment::Error::UnsupportedPaymentOption)
978 }
979 OutgoingPaymentOptions::Bolt11(bolt11_options) => {
980 let bolt11 = bolt11_options.bolt11;
981
982 let amount_msat = match bolt11_options.melt_options {
983 Some(MeltOptions::Amountless { amountless }) => {
984 let amount_msat = amountless.amount_msat;
985
986 if let Some(invoice_amount) = bolt11.amount_milli_satoshis() {
987 if invoice_amount != u64::from(amount_msat) {
988 return Err(payment::Error::AmountMismatch);
989 }
990 }
991
992 amount_msat
993 }
994 Some(MeltOptions::Mpp { mpp }) => mpp.amount,
995 None => bolt11
996 .amount_milli_satoshis()
997 .ok_or(Error::UnknownInvoiceAmount)?
998 .into(),
999 };
1000
1001 let amount = match unit {
1002 CurrencyUnit::Sat => Amount::new(
1004 u64::from(amount_msat).div_ceil(MSAT_IN_SAT),
1005 CurrencyUnit::Sat,
1006 ),
1007 _ => Amount::new(amount_msat.into(), CurrencyUnit::Msat).convert_to(unit)?,
1008 };
1009
1010 let relative_fee_reserve =
1011 (self.fee_reserve.percent_fee_reserve * amount.value() as f32) as u64;
1012
1013 let absolute_fee_reserve: u64 = self.fee_reserve.min_fee_reserve.into();
1014
1015 let fee = match relative_fee_reserve > absolute_fee_reserve {
1016 true => relative_fee_reserve,
1017 false => absolute_fee_reserve,
1018 };
1019
1020 let payment_hash = bolt11.payment_hash().to_string();
1021 let payment_hash_bytes = hex::decode(&payment_hash)?
1022 .try_into()
1023 .map_err(|_| Error::InvalidPaymentHashLength)?;
1024
1025 Ok(PaymentQuoteResponse {
1026 request_lookup_id: Some(PaymentIdentifier::PaymentHash(payment_hash_bytes)),
1027 amount,
1028 fee: Amount::new(fee, unit.clone()),
1029 state: MeltQuoteState::Unpaid,
1030 extra_json: None,
1031 estimated_blocks: None,
1032 fee_options: None,
1033 })
1034 }
1035 OutgoingPaymentOptions::Bolt12(bolt12_options) => {
1036 let offer = bolt12_options.offer;
1037
1038 let amount_msat = match bolt12_options.melt_options {
1039 Some(melt_options) => melt_options.amount_msat(),
1040 None => {
1041 let amount = offer.amount().ok_or(payment::Error::AmountMismatch)?;
1042
1043 match amount {
1044 ldk_node::lightning::offers::offer::Amount::Bitcoin {
1045 amount_msats,
1046 } => amount_msats.into(),
1047 _ => return Err(payment::Error::AmountMismatch),
1048 }
1049 }
1050 };
1051 let amount = match unit {
1052 CurrencyUnit::Sat => Amount::new(
1054 u64::from(amount_msat).div_ceil(MSAT_IN_SAT),
1055 CurrencyUnit::Sat,
1056 ),
1057 _ => Amount::new(amount_msat.into(), CurrencyUnit::Msat).convert_to(unit)?,
1058 };
1059
1060 let relative_fee_reserve =
1061 (self.fee_reserve.percent_fee_reserve * amount.value() as f32) as u64;
1062
1063 let absolute_fee_reserve: u64 = self.fee_reserve.min_fee_reserve.into();
1064
1065 let fee = match relative_fee_reserve > absolute_fee_reserve {
1066 true => relative_fee_reserve,
1067 false => absolute_fee_reserve,
1068 };
1069
1070 Ok(PaymentQuoteResponse {
1071 request_lookup_id: Some(PaymentIdentifier::QuoteId(
1072 bolt12_options.quote_id.clone(),
1073 )),
1074 amount,
1075 fee: Amount::new(fee, unit.clone()),
1076 state: MeltQuoteState::Unpaid,
1077 extra_json: None,
1078 estimated_blocks: None,
1079 fee_options: None,
1080 })
1081 }
1082 OutgoingPaymentOptions::Onchain(_) => {
1083 Err(cdk_common::payment::Error::UnsupportedPaymentOption)
1084 }
1085 }
1086 }
1087
1088 #[instrument(skip(self, options))]
1090 async fn make_payment(
1091 &self,
1092 unit: &CurrencyUnit,
1093 options: OutgoingPaymentOptions,
1094 ) -> Result<MakePaymentResponse, Self::Err> {
1095 match options {
1096 cdk_common::payment::OutgoingPaymentOptions::Custom(options) => {
1097 Ok(outgoing_payment_failure_response(
1098 unit,
1099 PaymentIdentifier::QuoteId(options.quote_id),
1100 ))
1101 }
1102 OutgoingPaymentOptions::Bolt11(bolt11_options) => {
1103 let bolt11 = bolt11_options.bolt11;
1104 let payment_lookup_id =
1105 PaymentIdentifier::PaymentHash(bolt11.payment_hash().to_byte_array());
1106
1107 let send_params = match bolt11_options
1108 .max_fee_amount
1109 .map(|f| {
1110 f.convert_to(&CurrencyUnit::Msat)
1111 .map(|amount_msat| RouteParametersConfig {
1112 max_total_routing_fee_msat: Some(amount_msat.value()),
1113 ..Default::default()
1114 })
1115 })
1116 .transpose()
1117 {
1118 Ok(params) => params,
1119 Err(err) => {
1120 tracing::error!("Failed to convert fee amount: {}", err);
1121 return Ok(outgoing_payment_failure_response(unit, payment_lookup_id));
1122 }
1123 };
1124
1125 let payment_event_receiver = self.outgoing_payment_sender.subscribe();
1128
1129 let payment_id = match bolt11_options.melt_options {
1130 Some(MeltOptions::Amountless { amountless }) => {
1131 if let Some(invoice_amount) = bolt11.amount_milli_satoshis() {
1132 if invoice_amount != u64::from(amountless.amount_msat) {
1133 return Ok(outgoing_payment_failure_response(
1134 unit,
1135 payment_lookup_id,
1136 ));
1137 }
1138 }
1139
1140 self.inner.bolt11_payment().send_using_amount(
1141 &bolt11,
1142 amountless.amount_msat.into(),
1143 send_params,
1144 )
1145 }
1146 None => self.inner.bolt11_payment().send(&bolt11, send_params),
1147 _ => {
1148 return Ok(outgoing_payment_failure_response(unit, payment_lookup_id));
1149 }
1150 };
1151
1152 let payment_id = match payment_id {
1153 Ok(payment_id) => payment_id,
1154 Err(err) if bolt11_send_error_is_explicit_terminal_failure(&err) => {
1155 tracing::warn!(
1156 payment_hash = %bolt11.payment_hash(),
1157 "LDK rejected BOLT11 payment before dispatch: {err}"
1158 );
1159 return Ok(outgoing_payment_failure_response(unit, payment_lookup_id));
1160 }
1161 Err(err) => {
1162 tracing::warn!(
1163 payment_hash = %bolt11.payment_hash(),
1164 "LDK BOLT11 send outcome is indeterminate: {err}"
1165 );
1166 return Err(Error::LdkNode(err).into());
1167 }
1168 };
1169
1170 let payment_details = self
1171 .wait_for_payment_terminal_status(payment_id, payment_event_receiver)
1172 .await?;
1173
1174 if payment_details.status == PaymentStatus::Failed {
1175 tracing::error!(payment_id = %payment_id, "Bolt11 payment failed");
1176 }
1177
1178 Self::make_payment_response_from_details(unit, payment_lookup_id, &payment_details)
1179 }
1180 OutgoingPaymentOptions::Bolt12(bolt12_options) => {
1181 let offer = bolt12_options.offer;
1182 let quote_id = bolt12_options.quote_id.clone();
1183 let quote_payment_identifier = PaymentIdentifier::QuoteId(quote_id.clone());
1184
1185 let send_params = match bolt12_options
1186 .max_fee_amount
1187 .map(|f| {
1188 f.convert_to(&CurrencyUnit::Msat)
1189 .map(|amount_msat| RouteParametersConfig {
1190 max_total_routing_fee_msat: Some(amount_msat.value()),
1191 ..Default::default()
1192 })
1193 })
1194 .transpose()
1195 {
1196 Ok(params) => params,
1197 Err(err) => {
1198 tracing::error!("Failed to convert fee amount: {}", err);
1199 return Ok(outgoing_payment_failure_response(
1200 unit,
1201 quote_payment_identifier,
1202 ));
1203 }
1204 };
1205
1206 if let Err(err) =
1214 write_bolt12_quote_payment_id(&self.kv_store, "e_id, None).await
1215 {
1216 tracing::error!(
1217 quote_id = %quote_id,
1218 "Could not persist BOLT12 dispatch claim before sending: {err}"
1219 );
1220 if matches!(&err, Error::Bolt12QuoteAlreadyClaimed { .. }) {
1221 let mut response = self
1224 .check_outgoing_payment("e_payment_identifier)
1225 .await?;
1226 response.total_spent = match (response.total_spent.unit(), unit) {
1229 (CurrencyUnit::Msat, CurrencyUnit::Sat) => Amount::new(
1230 response.total_spent.value().div_ceil(MSAT_IN_SAT),
1231 CurrencyUnit::Sat,
1232 ),
1233 _ => response.total_spent.convert_to(unit)?,
1234 };
1235 return Ok(response);
1236 }
1237
1238 return Err(err.into());
1239 }
1240
1241 let payment_event_receiver = self.outgoing_payment_sender.subscribe();
1244
1245 let payment_id = match bolt12_options.melt_options {
1246 Some(MeltOptions::Amountless { amountless }) => {
1247 self.inner.bolt12_payment().send_using_amount(
1248 &offer,
1249 amountless.amount_msat.into(),
1250 None,
1251 None,
1252 send_params,
1253 )
1254 }
1255 None => self
1256 .inner
1257 .bolt12_payment()
1258 .send(&offer, None, None, send_params),
1259 _ => {
1260 self.cleanup_bolt12_dispatch_binding("e_id, None).await;
1261 return Ok(outgoing_payment_failure_response(
1262 unit,
1263 quote_payment_identifier,
1264 ));
1265 }
1266 };
1267
1268 let payment_id = match payment_id {
1269 Ok(payment_id) => payment_id,
1270 Err(err) => {
1271 match bolt12_send_error_has_ambiguous_dispatch(&err) {
1272 true => {
1273 tracing::warn!(
1274 quote_id = %quote_id,
1275 "LDK payment persistence failed after BOLT12 send; retaining \
1276 the dispatch sentinel because the payment may have been dispatched"
1277 );
1278 }
1279 false => {
1280 self.cleanup_bolt12_dispatch_binding("e_id, None).await;
1281 tracing::warn!(
1282 quote_id = %quote_id,
1283 "LDK rejected BOLT12 payment before dispatch: {err}"
1284 );
1285 return Ok(outgoing_payment_failure_response(
1286 unit,
1287 quote_payment_identifier,
1288 ));
1289 }
1290 }
1291 return Err(Error::LdkNode(err).into());
1292 }
1293 };
1294
1295 if let Err(err) =
1301 write_bolt12_quote_payment_id(&self.kv_store, "e_id, Some(&payment_id))
1302 .await
1303 {
1304 tracing::error!(
1305 "Could not record BOLT12 payment id for quote {quote_id}: {err}. \
1306 The payment will remain Pending until manual intervention."
1307 );
1308 }
1309
1310 let payment_details = self
1311 .wait_for_payment_terminal_status(payment_id, payment_event_receiver)
1312 .await?;
1313
1314 if payment_details.status == PaymentStatus::Failed {
1315 tracing::error!(
1316 payment_id = %payment_id,
1317 amount_msat = ?payment_details.amount_msat,
1318 fee_paid_msat = ?payment_details.fee_paid_msat,
1319 payment_kind = ?payment_details.kind,
1320 "Bolt12 payment failed"
1321 );
1322 self.cleanup_bolt12_dispatch_binding("e_id, Some(&payment_id))
1323 .await;
1324 }
1325
1326 Self::make_payment_response_from_details(
1327 unit,
1328 quote_payment_identifier,
1329 &payment_details,
1330 )
1331 }
1332 OutgoingPaymentOptions::Onchain(options) => Ok(outgoing_payment_failure_response(
1333 unit,
1334 PaymentIdentifier::QuoteId(options.quote_id),
1335 )),
1336 }
1337 }
1338
1339 #[instrument(skip(self))]
1342 async fn wait_payment_event(
1343 &self,
1344 ) -> Result<Pin<Box<dyn Stream<Item = cdk_common::payment::Event> + Send>>, Self::Err> {
1345 tracing::info!("Starting stream for invoices - wait_any_incoming_payment called");
1346
1347 self.wait_invoice_is_active.store(true, Ordering::SeqCst);
1349 tracing::debug!("wait_invoice_is_active set to true");
1350
1351 let receiver = self.receiver.clone();
1352
1353 tracing::info!("Receiver obtained successfully, creating response stream");
1354
1355 let response_stream = BroadcastStream::new(receiver.resubscribe());
1357
1358 let response_stream = response_stream.filter_map(|result| async move {
1360 match result {
1361 Ok(payment) => Some(cdk_common::payment::Event::PaymentReceived(payment)),
1362 Err(err) => {
1363 tracing::warn!("Error in broadcast stream: {}", err);
1364 None
1365 }
1366 }
1367 });
1368
1369 let cancel_token = self.wait_invoice_cancel_token.clone();
1371 let is_active = self.wait_invoice_is_active.clone();
1372
1373 let stream = Box::pin(response_stream);
1374
1375 tokio::spawn(async move {
1377 cancel_token.cancelled().await;
1378 tracing::info!("wait_invoice stream cancelled");
1379 is_active.store(false, Ordering::SeqCst);
1380 });
1381
1382 tracing::info!("wait_any_incoming_payment returning stream");
1383 Ok(stream)
1384 }
1385
1386 fn is_payment_event_stream_active(&self) -> bool {
1388 self.wait_invoice_is_active.load(Ordering::SeqCst)
1389 }
1390
1391 fn cancel_payment_event_stream(&self) {
1393 self.wait_invoice_cancel_token.cancel()
1394 }
1395
1396 async fn check_incoming_payment_status(
1398 &self,
1399 payment_identifier: &PaymentIdentifier,
1400 ) -> Result<Vec<WaitPaymentResponse>, Self::Err> {
1401 if let PaymentIdentifier::OfferId(offer_id) = payment_identifier {
1404 let payments = self.inner.list_payments_with_filter(|p| {
1405 p.direction == PaymentDirection::Inbound
1406 && p.status == PaymentStatus::Succeeded
1407 && matches!(
1408 &p.kind,
1409 PaymentKind::Bolt12Offer { offer_id: oid, .. } if oid.to_string() == *offer_id
1410 )
1411 });
1412
1413 return Ok(payments
1414 .into_iter()
1415 .filter_map(|p| {
1416 let payment_id = match &p.kind {
1417 PaymentKind::Bolt12Offer {
1418 hash: Some(hash), ..
1419 } => hash.to_string(),
1420 _ => {
1421 tracing::warn!("Bolt12 payment for offer {} missing hash", offer_id);
1422 return None;
1423 }
1424 };
1425
1426 Some(WaitPaymentResponse {
1427 payment_identifier: payment_identifier.clone(),
1428 payment_amount: Amount::new(p.amount_msat?, CurrencyUnit::Msat),
1429 payment_id,
1430 })
1431 })
1432 .collect());
1433 }
1434
1435 let payment_id_str = match payment_identifier {
1436 PaymentIdentifier::PaymentHash(hash) => hex::encode(hash),
1437 PaymentIdentifier::CustomId(id) => id.clone(),
1438 _ => return Err(Error::UnsupportedPaymentIdentifierType.into()),
1439 };
1440
1441 let payment_id = PaymentId(
1442 hex::decode(&payment_id_str)?
1443 .try_into()
1444 .map_err(|_| Error::InvalidPaymentIdLength)?,
1445 );
1446
1447 let payment_details = self
1448 .inner
1449 .payment(&payment_id)
1450 .ok_or(Error::PaymentNotFound)?;
1451
1452 if payment_details.direction == PaymentDirection::Outbound {
1453 return Err(Error::InvalidPaymentDirection.into());
1454 }
1455
1456 let amount = if payment_details.status == PaymentStatus::Succeeded {
1457 payment_details
1458 .amount_msat
1459 .ok_or(Error::CouldNotGetPaymentAmount)?
1460 } else {
1461 return Ok(vec![]);
1462 };
1463
1464 let response = WaitPaymentResponse {
1465 payment_identifier: payment_identifier.clone(),
1466 payment_amount: Amount::new(amount, CurrencyUnit::Msat),
1467 payment_id: payment_id_str,
1468 };
1469
1470 Ok(vec![response])
1471 }
1472
1473 async fn check_outgoing_payment(
1475 &self,
1476 request_lookup_id: &PaymentIdentifier,
1477 ) -> Result<MakePaymentResponse, Self::Err> {
1478 let payment_details = match request_lookup_id {
1479 PaymentIdentifier::PaymentHash(id_hash) => {
1480 Self::select_bolt11_payment_details(self.inner.list_payments_with_filter(|p| {
1481 p.direction == PaymentDirection::Outbound
1482 && matches!(&p.kind, PaymentKind::Bolt11 { hash, .. } if &hash.0 == id_hash)
1483 }))
1484 }
1485 PaymentIdentifier::PaymentId(id) => self.inner.payment(&PaymentId(*id)),
1486 PaymentIdentifier::QuoteId(quote_id) => {
1487 match read_bolt12_quote_payment_id(&self.kv_store, quote_id)
1488 .await?
1489 .resolve()
1490 {
1491 Bolt12QuotePaymentIdResolution::PaymentId(payment_id) => {
1492 self.inner.payment(&payment_id)
1493 }
1494 Bolt12QuotePaymentIdResolution::Status(status) => {
1495 return Ok(MakePaymentResponse {
1496 payment_lookup_id: request_lookup_id.clone(),
1497 payment_proof: None,
1498 status,
1499 total_spent: Amount::new(0, CurrencyUnit::Msat),
1500 });
1501 }
1502 }
1503 }
1504 _ => {
1505 return Ok(MakePaymentResponse {
1506 payment_lookup_id: request_lookup_id.clone(),
1507 payment_proof: None,
1508 status: MeltQuoteState::Unknown,
1509 total_spent: Amount::new(0, CurrencyUnit::Msat),
1510 });
1511 }
1512 }
1513 .ok_or(Error::PaymentNotFound)?;
1514
1515 if payment_details.direction != PaymentDirection::Outbound {
1516 return Err(Error::InvalidPaymentDirection.into());
1517 }
1518
1519 if payment_details.status == PaymentStatus::Failed {
1520 if let PaymentIdentifier::QuoteId(quote_id) = request_lookup_id {
1521 self.cleanup_bolt12_dispatch_binding(quote_id, Some(&payment_details.id))
1522 .await;
1523 }
1524 }
1525
1526 Self::make_payment_response_from_details(
1527 &CurrencyUnit::Msat,
1528 request_lookup_id.clone(),
1529 &payment_details,
1530 )
1531 }
1532}
1533
1534impl Drop for CdkLdkNode {
1535 fn drop(&mut self) {
1536 tracing::info!("Drop called on CdkLdkNode");
1537 self.wait_invoice_cancel_token.cancel();
1538 tracing::debug!("Cancelled wait_invoice token in drop");
1539 }
1540}
1541
1542fn bolt12_quote_payment_id_key(quote_id: &QuoteId) -> Result<String, Error> {
1544 match quote_id {
1545 QuoteId::UUID(uuid) => Ok(uuid.to_string()),
1546 QuoteId::BASE64(_) => Err(Error::InvalidQuoteId),
1547 }
1548}
1549
1550async fn write_bolt12_quote_payment_id(
1557 kv_store: &DynKVStore,
1558 quote_id: &QuoteId,
1559 payment_id: Option<&PaymentId>,
1560) -> Result<(), Error> {
1561 let key = bolt12_quote_payment_id_key(quote_id)?;
1562 let value = payment_id.map(|id| hex::encode(id.0)).unwrap_or_default();
1563 let mut tx = kv_store
1564 .begin_transaction()
1565 .await
1566 .map_err(|e| Error::Database(e.to_string()))?;
1567
1568 let written = match payment_id {
1569 None => {
1570 tx.kv_write_if_absent(
1571 LDK_KV_PRIMARY_NAMESPACE,
1572 LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
1573 &key,
1574 value.as_bytes(),
1575 )
1576 .await
1577 }
1578 Some(_) => {
1579 tx.kv_write_if_equals(
1580 LDK_KV_PRIMARY_NAMESPACE,
1581 LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
1582 &key,
1583 b"",
1584 value.as_bytes(),
1585 )
1586 .await
1587 }
1588 }
1589 .map_err(|e| Error::Database(e.to_string()))?;
1590
1591 if written {
1592 tx.commit()
1593 .await
1594 .map_err(|e| Error::Database(e.to_string()))?;
1595 return Ok(());
1596 }
1597
1598 let existing = tx
1599 .kv_read(
1600 LDK_KV_PRIMARY_NAMESPACE,
1601 LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
1602 &key,
1603 )
1604 .await
1605 .map_err(|e| Error::Database(e.to_string()))?;
1606 tx.rollback()
1607 .await
1608 .map_err(|e| Error::Database(e.to_string()))?;
1609
1610 match existing {
1611 Some(existing) if payment_id.is_some() && existing.as_slice() == value.as_bytes() => Ok(()),
1612 _ => Err(Error::Bolt12QuoteAlreadyClaimed {
1613 quote_id: quote_id.to_string(),
1614 }),
1615 }
1616}
1617
1618async fn read_bolt12_quote_payment_id(
1620 kv_store: &DynKVStore,
1621 quote_id: &QuoteId,
1622) -> Result<Bolt12QuotePaymentIdLookup, Error> {
1623 let key = bolt12_quote_payment_id_key(quote_id)?;
1624 let Some(stored) = kv_store
1625 .kv_read(
1626 LDK_KV_PRIMARY_NAMESPACE,
1627 LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
1628 &key,
1629 )
1630 .await
1631 .map_err(|e| Error::Database(e.to_string()))?
1632 else {
1633 return Ok(Bolt12QuotePaymentIdLookup::Missing);
1634 };
1635
1636 if stored.is_empty() {
1637 return Ok(Bolt12QuotePaymentIdLookup::Dispatching);
1638 }
1639
1640 let payment_id_hex = match String::from_utf8(stored) {
1641 Ok(payment_id_hex) => payment_id_hex,
1642 Err(err) => {
1643 tracing::warn!(
1644 "LDK: invalid UTF-8 in BOLT12 payment id mapping for quote {quote_id}: {err}"
1645 );
1646 return Ok(Bolt12QuotePaymentIdLookup::Malformed);
1647 }
1648 };
1649
1650 let payment_id_bytes = match hex::decode(&payment_id_hex) {
1651 Ok(bytes) => bytes,
1652 Err(err) => {
1653 tracing::warn!(
1654 "LDK: invalid hex in BOLT12 payment id mapping for quote {quote_id}: {err}"
1655 );
1656 return Ok(Bolt12QuotePaymentIdLookup::Malformed);
1657 }
1658 };
1659
1660 let payment_id: [u8; 32] = match payment_id_bytes.try_into() {
1661 Ok(payment_id) => payment_id,
1662 Err(_) => {
1663 tracing::warn!("LDK: invalid payment id length in BOLT12 mapping for quote {quote_id}");
1664 return Ok(Bolt12QuotePaymentIdLookup::Malformed);
1665 }
1666 };
1667
1668 Ok(Bolt12QuotePaymentIdLookup::Found(PaymentId(payment_id)))
1669}
1670
1671async fn delete_bolt12_quote_payment_id_if_equals(
1674 kv_store: &DynKVStore,
1675 quote_id: &QuoteId,
1676 payment_id: Option<&PaymentId>,
1677) -> Result<bool, Error> {
1678 let key = bolt12_quote_payment_id_key(quote_id)?;
1679 let expected = payment_id.map(|id| hex::encode(id.0)).unwrap_or_default();
1680 let mut tx = kv_store
1681 .begin_transaction()
1682 .await
1683 .map_err(|e| Error::Database(e.to_string()))?;
1684
1685 let claimed = tx
1686 .kv_write_if_equals(
1687 LDK_KV_PRIMARY_NAMESPACE,
1688 LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
1689 &key,
1690 expected.as_bytes(),
1691 LDK_KV_BOLT12_CLEANUP_MARKER,
1692 )
1693 .await
1694 .map_err(|e| Error::Database(e.to_string()))?;
1695
1696 if !claimed {
1697 tx.rollback()
1698 .await
1699 .map_err(|e| Error::Database(e.to_string()))?;
1700 return Ok(false);
1701 }
1702
1703 tx.kv_remove(
1704 LDK_KV_PRIMARY_NAMESPACE,
1705 LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
1706 &key,
1707 )
1708 .await
1709 .map_err(|e| Error::Database(e.to_string()))?;
1710 tx.commit()
1711 .await
1712 .map_err(|e| Error::Database(e.to_string()))?;
1713
1714 Ok(true)
1715}
1716
1717#[cfg(test)]
1718mod tests {
1719 use super::*;
1720
1721 #[test]
1722 fn bitcoin_rpc_debug_redacts_password() {
1723 let source = ChainSource::BitcoinRpc(BitcoinRpcConfig {
1724 host: "127.0.0.1".to_string(),
1725 port: 8332,
1726 user: "rpc-user".to_string(),
1727 password: "rpc-password-secret".to_string(),
1728 });
1729
1730 let debug = format!("{source:?}");
1731
1732 assert!(debug.contains("127.0.0.1"));
1733 assert!(debug.contains("rpc-user"));
1734 assert!(debug.contains("[REDACTED]"));
1735 assert!(!debug.contains("rpc-password-secret"));
1736 }
1737
1738 #[test]
1739 fn chain_source_debug_redacts_url_credentials() {
1740 for source in [
1741 ChainSource::Esplora("https://esplora-user:esplora-secret@example.com/api".to_string()),
1742 ChainSource::Electrum(
1743 "ssl://electrum-user:electrum-secret@example.com:50002".to_string(),
1744 ),
1745 ] {
1746 let debug = format!("{source:?}");
1747
1748 assert!(debug.contains("example.com"));
1749 assert!(!debug.contains("-user"));
1750 assert!(!debug.contains("-secret"));
1751 }
1752 }
1753
1754 #[test]
1755 fn gossip_source_debug_redacts_url_credentials() {
1756 let source = GossipSource::RapidGossipSync(
1757 "https://rgs-user:rgs-secret@example.com/snapshot".to_string(),
1758 );
1759
1760 let debug = format!("{source:?}");
1761
1762 assert!(debug.contains("https://example.com/snapshot"));
1763 assert!(!debug.contains("rgs-user"));
1764 assert!(!debug.contains("rgs-secret"));
1765 }
1766
1767 fn test_payment_details(status: PaymentStatus, amount_msat: Option<u64>) -> PaymentDetails {
1768 PaymentDetails {
1769 id: PaymentId([2; 32]),
1770 kind: PaymentKind::Bolt11 {
1771 hash: PaymentHash([1; 32]),
1772 preimage: None,
1773 secret: None,
1774 },
1775 amount_msat,
1776 fee_paid_msat: None,
1777 direction: PaymentDirection::Outbound,
1778 status,
1779 latest_update_timestamp: 0,
1780 }
1781 }
1782
1783 fn test_payment_details_with_id(
1784 id: [u8; 32],
1785 status: PaymentStatus,
1786 latest_update_timestamp: u64,
1787 ) -> PaymentDetails {
1788 PaymentDetails {
1789 id: PaymentId(id),
1790 latest_update_timestamp,
1791 ..test_payment_details(status, None)
1792 }
1793 }
1794
1795 #[test]
1796 fn failed_payment_response_does_not_require_amount() {
1797 let details = test_payment_details(PaymentStatus::Failed, None);
1798
1799 let response = CdkLdkNode::make_payment_response_from_details(
1800 &CurrencyUnit::Msat,
1801 PaymentIdentifier::PaymentId([2; 32]),
1802 &details,
1803 )
1804 .expect("failed payment details should map without amount");
1805
1806 assert_eq!(response.status, MeltQuoteState::Failed);
1807 assert_eq!(response.total_spent, Amount::new(0, CurrencyUnit::Msat));
1808 }
1809
1810 #[test]
1811 fn pending_payment_response_does_not_require_amount() {
1812 let details = test_payment_details(PaymentStatus::Pending, None);
1813
1814 let response = CdkLdkNode::make_payment_response_from_details(
1815 &CurrencyUnit::Msat,
1816 PaymentIdentifier::PaymentId([2; 32]),
1817 &details,
1818 )
1819 .expect("pending payment details should map without amount");
1820
1821 assert_eq!(response.status, MeltQuoteState::Pending);
1822 assert_eq!(response.total_spent, Amount::new(0, CurrencyUnit::Msat));
1823 }
1824
1825 #[tokio::test]
1826 async fn payment_quotes_round_up_sat_principals() {
1827 let storage = std::env::temp_dir().join(format!("cdk-ldk-rounding-{}", QuoteId::new()));
1828 let node = CdkLdkNodeBuilder::new(
1829 Network::Regtest,
1830 ChainSource::Esplora("http://127.0.0.1:1".to_owned()),
1831 GossipSource::P2P,
1832 storage.to_str().unwrap().to_owned(),
1833 FeeReserve {
1834 min_fee_reserve: Amount::ZERO,
1835 percent_fee_reserve: 0.0,
1836 },
1837 vec!["127.0.0.1:0".parse().unwrap()],
1838 test_kv_store().await,
1839 )
1840 .build()
1841 .unwrap();
1842 for (msat, sat) in [(1, 1), (999, 1), (1_000, 1), (1_999, 2), (2_000, 2)] {
1843 let invoice = node
1844 .inner
1845 .bolt11_payment()
1846 .receive(
1847 msat,
1848 &Bolt11InvoiceDescription::Direct(
1849 Description::new("rounding".to_owned()).unwrap(),
1850 ),
1851 3600,
1852 )
1853 .unwrap();
1854 let offer = ldk_node::lightning::offers::offer::OfferBuilder::new(node.inner.node_id())
1855 .amount_msats(msat)
1856 .build()
1857 .unwrap();
1858 for (unit, expected) in [(CurrencyUnit::Sat, sat), (CurrencyUnit::Msat, msat)] {
1859 for options in [
1860 OutgoingPaymentOptions::Bolt11(Box::new(Bolt11OutgoingPaymentOptions {
1861 bolt11: invoice.clone(),
1862 max_fee_amount: None,
1863 timeout_secs: None,
1864 melt_options: None,
1865 quote_id: QuoteId::new(),
1866 })),
1867 OutgoingPaymentOptions::Bolt12(Box::new(Bolt12OutgoingPaymentOptions {
1868 offer: offer.clone(),
1869 max_fee_amount: None,
1870 timeout_secs: None,
1871 melt_options: None,
1872 quote_id: QuoteId::new(),
1873 })),
1874 ] {
1875 let quote = node.get_payment_quote(&unit, options).await.unwrap();
1876 assert_eq!(quote.amount, Amount::new(expected, unit.clone()));
1877 }
1878 }
1879 }
1880 drop(node);
1881 std::fs::remove_dir_all(storage).unwrap();
1882 }
1883
1884 #[test]
1885 fn paid_payment_response_rounds_total_once_for_sat_quotes() {
1886 for (principal, fee, expected_sat) in
1887 [(1_999, 0, 2), (2_000, 1, 3), (1_999, 1, 2), (2_000, 0, 2)]
1888 {
1889 let mut details = test_payment_details(PaymentStatus::Succeeded, Some(principal));
1890 details.fee_paid_msat = Some(fee);
1891 let payment_id = PaymentIdentifier::PaymentId([2; 32]);
1892 let synchronous = CdkLdkNode::make_payment_response_from_details(
1893 &CurrencyUnit::Sat,
1894 payment_id.clone(),
1895 &details,
1896 )
1897 .unwrap();
1898 let recovered = CdkLdkNode::make_payment_response_from_details(
1899 &CurrencyUnit::Msat,
1900 payment_id,
1901 &details,
1902 )
1903 .unwrap();
1904 assert_eq!(
1905 synchronous.total_spent,
1906 Amount::new(expected_sat, CurrencyUnit::Sat)
1907 );
1908 assert_eq!(
1909 recovered.total_spent,
1910 Amount::new(principal + fee, CurrencyUnit::Msat)
1911 );
1912 }
1913 }
1914
1915 #[test]
1916 fn paid_payment_response_requires_amount() {
1917 let details = test_payment_details(PaymentStatus::Succeeded, None);
1918
1919 let err = CdkLdkNode::make_payment_response_from_details(
1920 &CurrencyUnit::Msat,
1921 PaymentIdentifier::PaymentId([2; 32]),
1922 &details,
1923 )
1924 .expect_err("paid payment details without amount should fail");
1925
1926 assert!(matches!(err, payment::Error::Backend(_)));
1927 }
1928
1929 #[test]
1930 fn bolt11_payment_selection_prefers_pending_over_failed() {
1931 let failed = test_payment_details_with_id([1; 32], PaymentStatus::Failed, 2);
1932 let pending = test_payment_details_with_id([2; 32], PaymentStatus::Pending, 1);
1933
1934 let selected = CdkLdkNode::select_bolt11_payment_details([failed, pending])
1935 .expect("payment details should be selected");
1936
1937 assert_eq!(selected.id, PaymentId([2; 32]));
1938 assert_eq!(selected.status, PaymentStatus::Pending);
1939 }
1940
1941 #[test]
1942 fn bolt11_payment_selection_prefers_succeeded_over_pending() {
1943 let pending = test_payment_details_with_id([1; 32], PaymentStatus::Pending, 2);
1944 let succeeded = PaymentDetails {
1945 amount_msat: Some(1000),
1946 ..test_payment_details_with_id([2; 32], PaymentStatus::Succeeded, 1)
1947 };
1948
1949 let selected = CdkLdkNode::select_bolt11_payment_details([pending, succeeded])
1950 .expect("payment details should be selected");
1951
1952 assert_eq!(selected.id, PaymentId([2; 32]));
1953 assert_eq!(selected.status, PaymentStatus::Succeeded);
1954 }
1955
1956 #[test]
1957 fn bolt11_payment_selection_uses_latest_failed_when_all_failed() {
1958 let older_failed = test_payment_details_with_id([1; 32], PaymentStatus::Failed, 1);
1959 let newer_failed = test_payment_details_with_id([2; 32], PaymentStatus::Failed, 2);
1960
1961 let selected = CdkLdkNode::select_bolt11_payment_details([older_failed, newer_failed])
1962 .expect("payment details should be selected");
1963
1964 assert_eq!(selected.id, PaymentId([2; 32]));
1965 assert_eq!(selected.status, PaymentStatus::Failed);
1966 }
1967
1968 #[tokio::test]
1969 async fn terminal_payment_event_wait_ignores_other_payments() {
1970 let (sender, mut receiver) = tokio::sync::broadcast::channel(4);
1971 let payment_id = PaymentId([2; 32]);
1972
1973 sender
1976 .send(PaymentId([1; 32]))
1977 .expect("receiver should be subscribed");
1978 sender
1979 .send(payment_id)
1980 .expect("receiver should be subscribed");
1981
1982 CdkLdkNode::wait_for_terminal_payment_event(&mut receiver, payment_id)
1983 .await
1984 .expect("matching terminal event should wake the waiter");
1985 }
1986
1987 #[tokio::test]
1988 async fn terminal_payment_event_wait_recovers_from_lagged_channel() {
1989 let (sender, mut receiver) = tokio::sync::broadcast::channel(2);
1990 let payment_id = PaymentId([3; 32]);
1991
1992 sender
1993 .send(PaymentId([1; 32]))
1994 .expect("receiver should be subscribed");
1995 sender
1996 .send(PaymentId([2; 32]))
1997 .expect("receiver should be subscribed");
1998 sender
1999 .send(PaymentId([4; 32]))
2000 .expect("receiver should be subscribed");
2001 sender
2002 .send(payment_id)
2003 .expect("receiver should be subscribed");
2004
2005 CdkLdkNode::wait_for_terminal_payment_event(&mut receiver, payment_id)
2006 .await
2007 .expect("receiver lag should not prevent a matching event from waking the waiter");
2008 }
2009
2010 #[tokio::test]
2011 async fn terminal_payment_event_wait_reports_closed_channel() {
2012 let (sender, mut receiver) = tokio::sync::broadcast::channel(1);
2013 drop(sender);
2014
2015 let err = CdkLdkNode::wait_for_terminal_payment_event(&mut receiver, PaymentId([2; 32]))
2016 .await
2017 .expect_err("a closed event channel should stop the wait");
2018
2019 assert!(matches!(
2020 err,
2021 tokio::sync::broadcast::error::RecvError::Closed
2022 ));
2023 }
2024
2025 #[test]
2026 fn bolt12_persistence_failure_has_ambiguous_dispatch() {
2027 assert!(bolt12_send_error_has_ambiguous_dispatch(
2028 &ldk_node::NodeError::PersistenceFailed
2029 ));
2030
2031 for not_dispatched in [
2032 ldk_node::NodeError::NotRunning,
2033 ldk_node::NodeError::UnsupportedCurrency,
2034 ldk_node::NodeError::InvalidOffer,
2035 ldk_node::NodeError::InvalidAmount,
2036 ldk_node::NodeError::DuplicatePayment,
2037 ldk_node::NodeError::InvoiceRequestCreationFailed,
2038 ldk_node::NodeError::PaymentSendingFailed,
2039 ] {
2040 assert!(
2041 !bolt12_send_error_has_ambiguous_dispatch(¬_dispatched),
2042 "{not_dispatched} must be treated as not dispatched"
2043 );
2044 }
2045 }
2046
2047 #[test]
2048 fn bolt11_send_errors_only_classify_explicit_rejections_as_terminal() {
2049 for terminal_error in [
2050 ldk_node::NodeError::NotRunning,
2051 ldk_node::NodeError::InvalidAmount,
2052 ldk_node::NodeError::InvalidInvoice,
2053 ldk_node::NodeError::PaymentSendingFailed,
2054 ] {
2055 assert!(
2056 bolt11_send_error_is_explicit_terminal_failure(&terminal_error),
2057 "{terminal_error} must be treated as a definitive failure"
2058 );
2059 }
2060
2061 for ambiguous_error in [
2062 ldk_node::NodeError::PersistenceFailed,
2063 ldk_node::NodeError::DuplicatePayment,
2064 ] {
2065 assert!(
2066 !bolt11_send_error_is_explicit_terminal_failure(&ambiguous_error),
2067 "{ambiguous_error} must not authorize proof release"
2068 );
2069 }
2070 }
2071
2072 #[test]
2073 fn authoritative_outgoing_failure_response_is_terminal_and_spends_nothing() {
2074 let payment_lookup_id = PaymentIdentifier::PaymentHash([42; 32]);
2075 let response =
2076 outgoing_payment_failure_response(&CurrencyUnit::Msat, payment_lookup_id.clone());
2077
2078 assert_eq!(response.payment_lookup_id, payment_lookup_id);
2079 assert_eq!(response.status, MeltQuoteState::Failed);
2080 assert_eq!(response.total_spent, Amount::new(0, CurrencyUnit::Msat));
2081 assert!(response.payment_proof.is_none());
2082 }
2083
2084 #[test]
2085 fn bolt12_quote_payment_id_lookup_resolution_is_safe() {
2086 assert_eq!(
2087 Bolt12QuotePaymentIdLookup::Dispatching.resolve(),
2088 Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Pending),
2089 "an indeterminate dispatch must keep melt proofs reserved"
2090 );
2091 assert_eq!(
2092 Bolt12QuotePaymentIdLookup::Missing.resolve(),
2093 Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Unpaid),
2094 "a missing sentinel means the payment was never dispatched"
2095 );
2096 assert_eq!(
2097 Bolt12QuotePaymentIdLookup::Malformed.resolve(),
2098 Bolt12QuotePaymentIdResolution::Status(MeltQuoteState::Unknown),
2099 "corrupt bookkeeping must remain indeterminate"
2100 );
2101 }
2102
2103 async fn test_kv_store() -> DynKVStore {
2104 std::sync::Arc::new(cdk_sqlite::mint::memory::empty().await.unwrap())
2105 }
2106
2107 #[tokio::test]
2108 async fn duplicate_bolt12_claim_resolves_existing_state() {
2109 use ldk_node::lightning::offers::offer::OfferBuilder;
2110
2111 let storage = tempfile::tempdir().unwrap();
2112 let kv_store = test_kv_store().await;
2113 let node = CdkLdkNodeBuilder::new(
2114 Network::Regtest,
2115 ChainSource::Esplora("http://127.0.0.1:1".to_string()),
2116 GossipSource::P2P,
2117 storage.path().to_str().unwrap().to_string(),
2118 FeeReserve {
2119 min_fee_reserve: 0.into(),
2120 percent_fee_reserve: 0.0,
2121 },
2122 vec!["127.0.0.1:0".parse().unwrap()],
2123 kv_store.clone(),
2124 )
2125 .build()
2126 .unwrap();
2127 let offer = OfferBuilder::new(node.inner.node_id())
2128 .amount_msats(1_000)
2129 .build()
2130 .unwrap();
2131
2132 for (stored, expected) in [
2133 (String::new(), Some(MeltQuoteState::Pending)),
2134 ("corrupt".to_string(), Some(MeltQuoteState::Unknown)),
2135 (hex::encode([7; 32]), None),
2136 ] {
2137 let quote_id = QuoteId::new();
2138 let key = bolt12_quote_payment_id_key("e_id).unwrap();
2139 let mut tx = kv_store.begin_transaction().await.unwrap();
2140 tx.kv_write(
2141 LDK_KV_PRIMARY_NAMESPACE,
2142 LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
2143 &key,
2144 stored.as_bytes(),
2145 )
2146 .await
2147 .unwrap();
2148 tx.commit().await.unwrap();
2149
2150 let result = node
2151 .make_payment(
2152 &CurrencyUnit::Sat,
2153 OutgoingPaymentOptions::Bolt12(Box::new(Bolt12OutgoingPaymentOptions {
2154 offer: offer.clone(),
2155 max_fee_amount: None,
2156 timeout_secs: None,
2157 melt_options: None,
2158 quote_id: quote_id.clone(),
2159 })),
2160 )
2161 .await;
2162
2163 match expected {
2164 Some(status) => {
2165 let response = result.unwrap();
2166 assert_eq!(response.status, status);
2167 assert_eq!(response.total_spent, Amount::new(0, CurrencyUnit::Sat));
2168 assert_eq!(
2169 response.payment_lookup_id,
2170 PaymentIdentifier::QuoteId(quote_id)
2171 );
2172 }
2173 None => assert!(result.is_err(), "missing LDK payment must be indeterminate"),
2174 }
2175 assert_eq!(
2176 kv_store
2177 .kv_read(
2178 LDK_KV_PRIMARY_NAMESPACE,
2179 LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
2180 &key,
2181 )
2182 .await
2183 .unwrap(),
2184 Some(stored.into_bytes()),
2185 "a duplicate must preserve the existing dispatch binding"
2186 );
2187 }
2188 }
2189
2190 #[tokio::test]
2191 async fn duplicate_paid_bolt12_claim_rounds_total_once() {
2192 use ldk_node::bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
2193 use ldk_node::io::sqlite_store::{SqliteStore, KV_TABLE_NAME, SQLITE_DB_FILE_NAME};
2194 use ldk_node::lightning::offers::offer::OfferBuilder;
2195 use ldk_node::lightning::util::persist::KVStoreSync;
2196 use ldk_node::lightning::util::ser::Writeable;
2197 use ldk_node::lightning_types::payment::PaymentPreimage;
2198
2199 let storage = tempfile::tempdir().unwrap();
2200 let kv_store = test_kv_store().await;
2201 let signer = PublicKey::from_secret_key(
2202 &Secp256k1::new(),
2203 &SecretKey::from_slice(&[1; 32]).unwrap(),
2204 );
2205 let offer = OfferBuilder::new(signer)
2206 .amount_msats(1_000)
2207 .build()
2208 .unwrap();
2209 let preimage = PaymentPreimage([42; 32]);
2210 let mut cases = Vec::new();
2211 {
2212 let store = SqliteStore::new(
2213 storage.path().to_path_buf(),
2214 Some(SQLITE_DB_FILE_NAME.to_owned()),
2215 Some(KV_TABLE_NAME.to_owned()),
2216 )
2217 .unwrap();
2218 for (index, (principal, fee, expected_sat)) in
2219 [(10_000, 1, 11), (1_999, 1, 2), (1_999, 0, 2), (2_000, 0, 2)]
2220 .into_iter()
2221 .enumerate()
2222 {
2223 let payment_id = PaymentId([index as u8 + 1; 32]);
2224 let details = PaymentDetails {
2225 id: payment_id,
2226 kind: PaymentKind::Bolt12Offer {
2227 hash: Some(PaymentHash(
2228 ldk_node::bitcoin::hashes::sha256::Hash::hash(&preimage.0)
2229 .to_byte_array(),
2230 )),
2231 preimage: Some(preimage),
2232 secret: None,
2233 offer_id: offer.id(),
2234 payer_note: None,
2235 quantity: None,
2236 },
2237 fee_paid_msat: Some(fee),
2238 ..test_payment_details(PaymentStatus::Succeeded, Some(principal))
2239 };
2240 store
2242 .write("payments", "", &hex::encode(payment_id.0), details.encode())
2243 .unwrap();
2244 let quote_id = QuoteId::new();
2245 write_bolt12_quote_payment_id(&kv_store, "e_id, None)
2246 .await
2247 .unwrap();
2248 write_bolt12_quote_payment_id(&kv_store, "e_id, Some(&payment_id))
2249 .await
2250 .unwrap();
2251 cases.push((quote_id, payment_id, principal + fee, expected_sat));
2252 }
2253 }
2254 let node = CdkLdkNodeBuilder::new(
2255 Network::Regtest,
2256 ChainSource::Esplora("http://127.0.0.1:1".to_owned()),
2257 GossipSource::P2P,
2258 storage.path().to_str().unwrap().to_owned(),
2259 FeeReserve {
2260 min_fee_reserve: 0.into(),
2261 percent_fee_reserve: 0.0,
2262 },
2263 vec!["127.0.0.1:0".parse().unwrap()],
2264 kv_store.clone(),
2265 )
2266 .build()
2267 .unwrap();
2268
2269 for (quote_id, payment_id, total_msat, expected_sat) in cases {
2270 for (unit, expected) in [
2271 (CurrencyUnit::Sat, expected_sat),
2272 (CurrencyUnit::Msat, total_msat),
2273 ] {
2274 let response = node
2275 .make_payment(
2276 &unit,
2277 OutgoingPaymentOptions::Bolt12(Box::new(Bolt12OutgoingPaymentOptions {
2278 offer: offer.clone(),
2279 max_fee_amount: None,
2280 timeout_secs: None,
2281 melt_options: None,
2282 quote_id: quote_id.clone(),
2283 })),
2284 )
2285 .await
2286 .unwrap();
2287 assert_eq!(response.status, MeltQuoteState::Paid);
2288 assert_eq!(response.total_spent, Amount::new(expected, unit));
2289 assert_eq!(response.payment_proof, Some(preimage.to_string()));
2290 assert_eq!(
2291 response.payment_lookup_id,
2292 PaymentIdentifier::QuoteId(quote_id.clone())
2293 );
2294 assert_eq!(
2295 read_bolt12_quote_payment_id(&kv_store, "e_id)
2296 .await
2297 .unwrap(),
2298 Bolt12QuotePaymentIdLookup::Found(payment_id),
2299 );
2300 }
2301 }
2302 assert_eq!(node.inner.list_payments().len(), 4);
2303 }
2304
2305 #[tokio::test]
2309 async fn bolt12_quote_payment_id_mapping_lifecycle() {
2310 let kv_store = test_kv_store().await;
2311 let quote_id = QuoteId::new();
2312
2313 assert_eq!(
2314 read_bolt12_quote_payment_id(&kv_store, "e_id)
2315 .await
2316 .unwrap(),
2317 Bolt12QuotePaymentIdLookup::Missing,
2318 "no record must resolve as never dispatched"
2319 );
2320
2321 write_bolt12_quote_payment_id(&kv_store, "e_id, None)
2323 .await
2324 .unwrap();
2325 assert_eq!(
2326 read_bolt12_quote_payment_id(&kv_store, "e_id)
2327 .await
2328 .unwrap(),
2329 Bolt12QuotePaymentIdLookup::Dispatching,
2330 "sentinel must resolve as indeterminate, never terminal"
2331 );
2332
2333 assert!(
2334 delete_bolt12_quote_payment_id_if_equals(&kv_store, "e_id, None)
2335 .await
2336 .unwrap(),
2337 "an unambiguous pre-dispatch failure should release its sentinel"
2338 );
2339 write_bolt12_quote_payment_id(&kv_store, "e_id, None)
2340 .await
2341 .expect("a retry should reclaim the quote after sentinel cleanup");
2342
2343 let payment_id = PaymentId([7; 32]);
2345 write_bolt12_quote_payment_id(&kv_store, "e_id, Some(&payment_id))
2346 .await
2347 .unwrap();
2348 assert_eq!(
2349 read_bolt12_quote_payment_id(&kv_store, "e_id)
2350 .await
2351 .unwrap(),
2352 Bolt12QuotePaymentIdLookup::Found(payment_id)
2353 );
2354
2355 assert!(
2357 delete_bolt12_quote_payment_id_if_equals(&kv_store, "e_id, Some(&payment_id))
2358 .await
2359 .unwrap()
2360 );
2361 assert_eq!(
2362 read_bolt12_quote_payment_id(&kv_store, "e_id)
2363 .await
2364 .unwrap(),
2365 Bolt12QuotePaymentIdLookup::Missing
2366 );
2367 }
2368
2369 #[tokio::test]
2370 async fn bolt12_quote_payment_id_binding_is_write_once() {
2371 let kv_store = test_kv_store().await;
2372 let quote_id = QuoteId::new();
2373 let payment_id = PaymentId([7; 32]);
2374 let conflicting_payment_id = PaymentId([9; 32]);
2375
2376 write_bolt12_quote_payment_id(&kv_store, "e_id, None)
2377 .await
2378 .expect("first dispatch should claim the quote");
2379 write_bolt12_quote_payment_id(&kv_store, "e_id, Some(&payment_id))
2380 .await
2381 .expect("the claim owner should record its payment id");
2382 write_bolt12_quote_payment_id(&kv_store, "e_id, Some(&payment_id))
2383 .await
2384 .expect("repeating the same payment id must be idempotent");
2385
2386 let duplicate_dispatch = write_bolt12_quote_payment_id(&kv_store, "e_id, None)
2387 .await
2388 .expect_err("a dispatched quote must not be claimed again");
2389 assert!(
2390 matches!(duplicate_dispatch, Error::Bolt12QuoteAlreadyClaimed { .. }),
2391 "unexpected error: {duplicate_dispatch}"
2392 );
2393
2394 let conflicting_binding =
2395 write_bolt12_quote_payment_id(&kv_store, "e_id, Some(&conflicting_payment_id))
2396 .await
2397 .expect_err("a conflicting payment id must be rejected");
2398 assert!(
2399 matches!(conflicting_binding, Error::Bolt12QuoteAlreadyClaimed { .. }),
2400 "unexpected error: {conflicting_binding}"
2401 );
2402
2403 assert_eq!(
2404 read_bolt12_quote_payment_id(&kv_store, "e_id)
2405 .await
2406 .expect("payment id should remain readable"),
2407 Bolt12QuotePaymentIdLookup::Found(payment_id),
2408 "a duplicate dispatch must not redirect recovery"
2409 );
2410 }
2411
2412 #[tokio::test]
2413 async fn failed_bolt12_binding_can_be_released_without_removing_a_retry() {
2414 let kv_store = test_kv_store().await;
2415 let quote_id = QuoteId::new();
2416 let failed_payment_id = PaymentId([7; 32]);
2417 let retry_payment_id = PaymentId([9; 32]);
2418
2419 write_bolt12_quote_payment_id(&kv_store, "e_id, None)
2420 .await
2421 .expect("failed dispatch should claim the quote");
2422 write_bolt12_quote_payment_id(&kv_store, "e_id, Some(&failed_payment_id))
2423 .await
2424 .expect("failed payment id should be recorded");
2425 assert!(delete_bolt12_quote_payment_id_if_equals(
2426 &kv_store,
2427 "e_id,
2428 Some(&failed_payment_id),
2429 )
2430 .await
2431 .expect("failed binding should be released"));
2432
2433 write_bolt12_quote_payment_id(&kv_store, "e_id, None)
2434 .await
2435 .expect("retry should claim the released quote");
2436 write_bolt12_quote_payment_id(&kv_store, "e_id, Some(&retry_payment_id))
2437 .await
2438 .expect("retry payment id should be recorded");
2439 assert!(!delete_bolt12_quote_payment_id_if_equals(
2440 &kv_store,
2441 "e_id,
2442 Some(&failed_payment_id),
2443 )
2444 .await
2445 .expect("stale cleanup should be checked atomically"));
2446
2447 assert_eq!(
2448 read_bolt12_quote_payment_id(&kv_store, "e_id)
2449 .await
2450 .expect("retry binding should remain readable"),
2451 Bolt12QuotePaymentIdLookup::Found(retry_payment_id),
2452 "stale failed-payment cleanup must not remove a newer retry"
2453 );
2454 }
2455
2456 #[tokio::test]
2457 async fn bolt12_quote_dispatch_concurrent_claims_have_single_winner() {
2458 let kv_store = test_kv_store().await;
2459 let quote_id = QuoteId::new();
2460
2461 let (first_result, second_result) = tokio::join!(
2462 write_bolt12_quote_payment_id(&kv_store, "e_id, None),
2463 write_bolt12_quote_payment_id(&kv_store, "e_id, None),
2464 );
2465
2466 let outcomes = [first_result, second_result];
2467 let winners = outcomes.iter().filter(|result| result.is_ok()).count();
2468 let conflicts = outcomes
2469 .iter()
2470 .filter(|result| matches!(result, Err(Error::Bolt12QuoteAlreadyClaimed { .. })))
2471 .count();
2472
2473 assert_eq!(winners, 1, "exactly one dispatch may claim the quote");
2474 assert_eq!(conflicts, 1, "the losing dispatch must be rejected");
2475 }
2476
2477 #[tokio::test]
2478 async fn bolt12_quote_payment_id_concurrent_resolution_has_single_winner() {
2479 let kv_store = test_kv_store().await;
2480 let quote_id = QuoteId::new();
2481 let first_payment_id = PaymentId([7; 32]);
2482 let second_payment_id = PaymentId([9; 32]);
2483
2484 write_bolt12_quote_payment_id(&kv_store, "e_id, None)
2485 .await
2486 .expect("dispatch should claim the quote");
2487
2488 let (first_result, second_result) = tokio::join!(
2489 write_bolt12_quote_payment_id(&kv_store, "e_id, Some(&first_payment_id)),
2490 write_bolt12_quote_payment_id(&kv_store, "e_id, Some(&second_payment_id)),
2491 );
2492
2493 let outcomes = [&first_result, &second_result];
2494 let winners = outcomes.iter().filter(|result| result.is_ok()).count();
2495 let conflicts = outcomes
2496 .iter()
2497 .filter(|result| matches!(result, Err(Error::Bolt12QuoteAlreadyClaimed { .. })))
2498 .count();
2499
2500 assert_eq!(winners, 1, "exactly one payment id may resolve the claim");
2501 assert_eq!(conflicts, 1, "the losing resolution must be rejected");
2502
2503 let winner = if first_result.is_ok() {
2504 first_payment_id
2505 } else {
2506 second_payment_id
2507 };
2508 assert_eq!(
2509 read_bolt12_quote_payment_id(&kv_store, "e_id)
2510 .await
2511 .expect("payment id should remain readable"),
2512 Bolt12QuotePaymentIdLookup::Found(winner)
2513 );
2514 }
2515
2516 #[tokio::test]
2519 async fn bolt12_quote_payment_id_mapping_malformed_is_indeterminate() {
2520 let kv_store = test_kv_store().await;
2521 let quote_id = QuoteId::new();
2522 let key = bolt12_quote_payment_id_key("e_id).unwrap();
2523
2524 for corrupt in ["not-hex", "0102", "zz"] {
2525 let mut tx = kv_store.begin_transaction().await.unwrap();
2526 tx.kv_write(
2527 LDK_KV_PRIMARY_NAMESPACE,
2528 LDK_KV_BOLT12_OUTGOING_SECONDARY_NAMESPACE,
2529 &key,
2530 corrupt.as_bytes(),
2531 )
2532 .await
2533 .unwrap();
2534 tx.commit().await.unwrap();
2535
2536 assert_eq!(
2537 read_bolt12_quote_payment_id(&kv_store, "e_id)
2538 .await
2539 .unwrap(),
2540 Bolt12QuotePaymentIdLookup::Malformed,
2541 "corrupt value {corrupt} must be indeterminate"
2542 );
2543 }
2544 }
2545}