Adopt event handling logic from ldk-sample

Author: tnull

src/event.rs

@@ -1,6 +1,6 @@
 use crate::{
-	ChannelManager, Error, FilesystemPersister, LdkLiteChainAccess, LdkLiteConfig, NetworkGraph,
-	PaymentInfoStorage,
+	hex, ChannelManager, Error, FilesystemPersister, LdkLiteChainAccess, LdkLiteConfig,
+	NetworkGraph, PaymentInfo, PaymentInfoStorage, PaymentStatus,
 };
 
 #[allow(unused_imports)]
@@ -11,15 +11,19 @@ use crate::logger::{
 
 use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator};
 use lightning::chain::keysinterface::KeysManager;
+use lightning::routing::gossip::NodeId;
 use lightning::util::events as ldk_events;
 use lightning::util::persist::KVStorePersister;
 use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer};
 
 use bdk::database::MemoryDatabase;
 
 use bitcoin::secp256k1::Secp256k1;
-use std::collections::VecDeque;
+use rand::{thread_rng, Rng};
+use std::collections::{hash_map, VecDeque};
 use std::sync::{Arc, Condvar, Mutex};
+use std::thread;
+use std::time::Duration;
 
 /// The event queue will be persisted under this key.
 pub(crate) const EVENTS_PERSISTENCE_KEY: &str = "events";
@@ -134,10 +138,10 @@ pub(crate) struct LdkLiteEventHandler {
 	chain_access: Arc<LdkLiteChainAccess<MemoryDatabase>>,
 	event_queue: Arc<LdkLiteEventQueue<FilesystemPersister>>,
 	channel_manager: Arc<ChannelManager>,
-	_network_graph: Arc<NetworkGraph>,
+	network_graph: Arc<NetworkGraph>,
 	keys_manager: Arc<KeysManager>,
-	_inbound_payments: Arc<PaymentInfoStorage>,
-	_outbound_payments: Arc<PaymentInfoStorage>,
+	inbound_payments: Arc<PaymentInfoStorage>,
+	outbound_payments: Arc<PaymentInfoStorage>,
 	logger: Arc<FilesystemLogger>,
 	_config: Arc<LdkLiteConfig>,
 }
@@ -146,19 +150,19 @@ impl LdkLiteEventHandler {
 	pub fn new(
 		chain_access: Arc<LdkLiteChainAccess<MemoryDatabase>>,
 		event_queue: Arc<LdkLiteEventQueue<FilesystemPersister>>,
-		channel_manager: Arc<ChannelManager>, _network_graph: Arc<NetworkGraph>,
-		keys_manager: Arc<KeysManager>, _inbound_payments: Arc<PaymentInfoStorage>,
-		_outbound_payments: Arc<PaymentInfoStorage>, logger: Arc<FilesystemLogger>,
+		channel_manager: Arc<ChannelManager>, network_graph: Arc<NetworkGraph>,
+		keys_manager: Arc<KeysManager>, inbound_payments: Arc<PaymentInfoStorage>,
+		outbound_payments: Arc<PaymentInfoStorage>, logger: Arc<FilesystemLogger>,
 		_config: Arc<LdkLiteConfig>,
 	) -> Self {
 		Self {
 			event_queue,
 			chain_access,
 			channel_manager,
-			_network_graph,
+			network_graph,
 			keys_manager,
-			_inbound_payments,
-			_outbound_payments,
+			inbound_payments,
+			outbound_payments,
 			logger,
 			_config,
 		}
@@ -207,16 +211,113 @@ impl ldk_events::EventHandler for LdkLiteEventHandler {
 					}
 				}
 			}
-			ldk_events::Event::PaymentReceived { .. } => {}
-			ldk_events::Event::PaymentClaimed { .. } => {}
-			ldk_events::Event::PaymentSent { .. } => {}
-			ldk_events::Event::PaymentFailed { .. } => {}
+			ldk_events::Event::PaymentReceived { payment_hash, purpose, amount_msat } => {
+				log_info!(
+					self.logger,
+					"Received payment from payment hash {} of {} millisatoshis",
+					hex::to_string(&payment_hash.0),
+					amount_msat,
+				);
+				let payment_preimage = match purpose {
+					ldk_events::PaymentPurpose::InvoicePayment { payment_preimage, .. } => {
+						*payment_preimage
+					}
+					ldk_events::PaymentPurpose::SpontaneousPayment(preimage) => Some(*preimage),
+				};
+				self.channel_manager.claim_funds(payment_preimage.unwrap());
+			}
+			ldk_events::Event::PaymentClaimed { payment_hash, purpose, amount_msat } => {
+				log_info!(
+					self.logger,
+					"Claimed payment from payment hash {} of {} millisatoshis",
+					hex::to_string(&payment_hash.0),
+					amount_msat,
+				);
+				let (payment_preimage, payment_secret) = match purpose {
+					ldk_events::PaymentPurpose::InvoicePayment {
+						payment_preimage,
+						payment_secret,
+						..
+					} => (*payment_preimage, Some(*payment_secret)),
+					ldk_events::PaymentPurpose::SpontaneousPayment(preimage) => {
+						(Some(*preimage), None)
+					}
+				};
+				let mut payments = self.inbound_payments.lock().unwrap();
+				match payments.entry(*payment_hash) {
+					hash_map::Entry::Occupied(mut e) => {
+						let payment = e.get_mut();
+						payment.status = PaymentStatus::Succeeded;
+						payment.preimage = payment_preimage;
+						payment.secret = payment_secret;
+					}
+					hash_map::Entry::Vacant(e) => {
+						e.insert(PaymentInfo {
+							preimage: payment_preimage,
+							secret: payment_secret,
+							status: PaymentStatus::Succeeded,
+							amount_msat: Some(*amount_msat),
+						});
+					}
+				}
+			}
+			ldk_events::Event::PaymentSent {
+				payment_preimage,
+				payment_hash,
+				fee_paid_msat,
+				..
+			} => {
+				let mut payments = self.outbound_payments.lock().unwrap();
+				for (hash, payment) in payments.iter_mut() {
+					if *hash == *payment_hash {
+						payment.preimage = Some(*payment_preimage);
+						payment.status = PaymentStatus::Succeeded;
+						log_info!(
+							self.logger,
+							"Successfully sent payment of {} millisatoshis{} from \
+								 payment hash {:?} with preimage {:?}",
+							payment.amount_msat.unwrap(),
+							if let Some(fee) = fee_paid_msat {
+								format!(" (fee {} msat)", fee)
+							} else {
+								"".to_string()
+							},
+							hex::to_string(&payment_hash.0),
+							hex::to_string(&payment_preimage.0)
+						);
+					}
+				}
+			}
+			ldk_events::Event::PaymentFailed { payment_hash, .. } => {
+				log_info!(
+					self.logger,
+					"Failed to send payment to payment hash {:?}: exhausted payment retry attempts",
+					hex::to_string(&payment_hash.0)
+				);
+
+				let mut payments = self.outbound_payments.lock().unwrap();
+				if payments.contains_key(&payment_hash) {
+					let payment = payments.get_mut(&payment_hash).unwrap();
+					payment.status = PaymentStatus::Failed;
+				}
+			}
+
 			ldk_events::Event::PaymentPathSuccessful { .. } => {}
 			ldk_events::Event::PaymentPathFailed { .. } => {}
 			ldk_events::Event::ProbeSuccessful { .. } => {}
 			ldk_events::Event::ProbeFailed { .. } => {}
 			ldk_events::Event::HTLCHandlingFailed { .. } => {}
-			ldk_events::Event::PendingHTLCsForwardable { .. } => {}
+			ldk_events::Event::PendingHTLCsForwardable { time_forwardable } => {
+				let forwarding_channel_manager = self.channel_manager.clone();
+				let min = time_forwardable.as_millis() as u64;
+
+				// TODO: any way we still can use tokio here?
+				thread::spawn(move || {
+					let millis_to_sleep = thread_rng().gen_range(min..min * 5) as u64;
+					thread::sleep(Duration::from_millis(millis_to_sleep));
+					forwarding_channel_manager.process_pending_htlc_forwards();
+				});
+			}
 			ldk_events::Event::SpendableOutputs { outputs } => {
 				let destination_address = self.chain_access.get_new_address().unwrap();
 				let output_descriptors = &outputs.iter().map(|a| a).collect::<Vec<_>>();
@@ -235,8 +336,77 @@ impl ldk_events::EventHandler for LdkLiteEventHandler {
 				self.chain_access.broadcast_transaction(&spending_tx);
 			}
 			ldk_events::Event::OpenChannelRequest { .. } => {}
-			ldk_events::Event::PaymentForwarded { .. } => {}
-			ldk_events::Event::ChannelClosed { .. } => {}
+			ldk_events::Event::PaymentForwarded {
+				prev_channel_id,
+				next_channel_id,
+				fee_earned_msat,
+				claim_from_onchain_tx,
+			} => {
+				let read_only_network_graph = self.network_graph.read_only();
+				let nodes = read_only_network_graph.nodes();
+				let channels = self.channel_manager.list_channels();
+
+				let node_str = |channel_id: &Option<[u8; 32]>| match channel_id {
+					None => String::new(),
+					Some(channel_id) => match channels.iter().find(|c| c.channel_id == *channel_id)
+					{
+						None => String::new(),
+						Some(channel) => {
+							match nodes.get(&NodeId::from_pubkey(&channel.counterparty.node_id)) {
+								None => "private node".to_string(),
+								Some(node) => match &node.announcement_info {
+									None => "unnamed node".to_string(),
+									Some(announcement) => {
+										format!("node {}", announcement.alias)
+									}
+								},
+							}
+						}
+					},
+				};
+				let channel_str = |channel_id: &Option<[u8; 32]>| {
+					channel_id
+						.map(|channel_id| format!(" with channel {}", hex::to_string(&channel_id)))
+						.unwrap_or_default()
+				};
+				let from_prev_str =
+					format!(" from {}{}", node_str(prev_channel_id), channel_str(prev_channel_id));
+				let to_next_str =
+					format!(" to {}{}", node_str(next_channel_id), channel_str(next_channel_id));
+
+				let from_onchain_str = if *claim_from_onchain_tx {
+					"from onchain downstream claim"
+				} else {
+					"from HTLC fulfill message"
+				};
+				if let Some(fee_earned) = fee_earned_msat {
+					log_info!(
+						self.logger,
+						"Forwarded payment{}{}, earning {} msat {}",
+						from_prev_str,
+						to_next_str,
+						fee_earned,
+						from_onchain_str
+					);
+				} else {
+					log_info!(
+						self.logger,
+						"Forwarded payment{}{}, claiming onchain {}",
+						from_prev_str,
+						to_next_str,
+						from_onchain_str
+					);
+				}
+			}
+
+			ldk_events::Event::ChannelClosed { channel_id, reason, user_channel_id: _ } => {
+				log_info!(
+					self.logger,
+					"Channel {} closed due to: {:?}",
+					hex::to_string(channel_id),
+					reason
+				);
+			}
 			ldk_events::Event::DiscardFunding { .. } => {}
 		}
 	}