Merge pull request #57 from G8XSU/2025-03-14-more-events

Start publishing more payment events using EventPublisher.

Author: G8XSU

ldk-server-protos/src/events.rs

@@ -2,7 +2,7 @@
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct EventEnvelope {
-	#[prost(oneof = "event_envelope::Event", tags = "2")]
+	#[prost(oneof = "event_envelope::Event", tags = "2, 3, 4, 6")]
 	pub event: ::core::option::Option<event_envelope::Event>,
 }
 /// Nested message and enum types in `EventEnvelope`.
@@ -11,9 +11,39 @@ pub mod event_envelope {
 	#[derive(Clone, PartialEq, ::prost::Oneof)]
 	pub enum Event {
 		#[prost(message, tag = "2")]
+		PaymentReceived(super::PaymentReceived),
+		#[prost(message, tag = "3")]
+		PaymentSuccessful(super::PaymentSuccessful),
+		#[prost(message, tag = "4")]
+		PaymentFailed(super::PaymentFailed),
+		#[prost(message, tag = "6")]
 		PaymentForwarded(super::PaymentForwarded),
 	}
 }
+/// PaymentReceived indicates a payment has been received.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct PaymentReceived {
+	/// The payment details for the payment in event.
+	#[prost(message, optional, tag = "1")]
+	pub payment: ::core::option::Option<super::types::Payment>,
+}
+/// PaymentSuccessful indicates a sent payment was successful.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct PaymentSuccessful {
+	/// The payment details for the payment in event.
+	#[prost(message, optional, tag = "1")]
+	pub payment: ::core::option::Option<super::types::Payment>,
+}
+/// PaymentFailed indicates a sent payment has failed.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct PaymentFailed {
+	/// The payment details for the payment in event.
+	#[prost(message, optional, tag = "1")]
+	pub payment: ::core::option::Option<super::types::Payment>,
+}
 /// PaymentForwarded indicates a payment was forwarded through the node.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]

ldk-server-protos/src/proto/events.proto

@@ -5,10 +5,31 @@ package events;
 // EventEnvelope wraps different event types in a single message to be used by EventPublisher.
 message EventEnvelope {
   oneof event {
-    PaymentForwarded payment_forwarded = 2;
+    PaymentReceived payment_received = 2;
+    PaymentSuccessful payment_successful = 3;
+    PaymentFailed payment_failed = 4;
+    PaymentForwarded payment_forwarded = 6;
   }
 }
 
+// PaymentReceived indicates a payment has been received.
+message PaymentReceived {
+  // The payment details for the payment in event.
+  types.Payment payment = 1;
+}
+
+// PaymentSuccessful indicates a sent payment was successful.
+message PaymentSuccessful {
+  // The payment details for the payment in event.
+  types.Payment payment = 1;
+}
+
+// PaymentFailed indicates a sent payment has failed.
+message PaymentFailed {
+  // The payment details for the payment in event.
+  types.Payment payment = 1;
+}
+
 // PaymentForwarded indicates a payment was forwarded through the node.
 message PaymentForwarded {
   types.ForwardedPayment forwarded_payment = 1;

ldk-server/src/io/events/event_publisher.rs

@@ -18,7 +18,7 @@ use ldk_server_protos::events::EventEnvelope;
 /// The underlying messaging system is expected to support durably buffered events,
 /// enabling easy decoupling between the LDK Server and event consumers.
 #[async_trait]
-pub trait EventPublisher {
+pub trait EventPublisher: Send + Sync {
 	/// Publishes an event to the underlying messaging system.
 	///
 	/// # Arguments

ldk-server/src/io/events/mod.rs

@@ -1 +1,13 @@
 pub(crate) mod event_publisher;
+
+use ldk_server_protos::events::event_envelope;
+
+/// Event variant to event name mapping.
+pub(crate) fn get_event_name(event: &event_envelope::Event) -> &'static str {
+	match event {
+		event_envelope::Event::PaymentReceived(_) => "PaymentReceived",
+		event_envelope::Event::PaymentSuccessful(_) => "PaymentSuccessful",
+		event_envelope::Event::PaymentFailed(_) => "PaymentFailed",
+		event_envelope::Event::PaymentForwarded(_) => "PaymentForwarded",
+	}
+}

ldk-server/src/io/persist/paginated_kv_store.rs

@@ -26,7 +26,7 @@ use std::io;
 /// [`KVStore`]: ldk_node::lightning::util::persist::KVStore
 /// [`KVSTORE_NAMESPACE_KEY_ALPHABET`]: ldk_node::lightning::util::persist::KVSTORE_NAMESPACE_KEY_ALPHABET
 /// [`KVSTORE_NAMESPACE_KEY_MAX_LEN`]: ldk_node::lightning::util::persist::KVSTORE_NAMESPACE_KEY_MAX_LEN
-pub trait PaginatedKVStore {
+pub trait PaginatedKVStore: Send + Sync {
 	/// Returns the data stored for the given `primary_namespace`, `secondary_namespace`, and `key`.
 	///
 	/// Returns an [`ErrorKind::NotFound`] if the given `key` could not be found in the given

ldk-server/src/main.rs

@@ -14,6 +14,7 @@ use hyper::server::conn::http1;
 use hyper_util::rt::TokioIo;
 
 use crate::io::events::event_publisher::{EventPublisher, NoopEventPublisher};
+use crate::io::events::get_event_name;
 use crate::io::persist::paginated_kv_store::PaginatedKVStore;
 use crate::io::persist::sqlite_store::SqliteStore;
 use crate::io::persist::{
@@ -29,12 +30,14 @@ use ldk_node::lightning::ln::channelmanager::PaymentId;
 use ldk_node::logger::LogLevel;
 use ldk_server_protos::events;
 use ldk_server_protos::events::{event_envelope, EventEnvelope};
+use ldk_server_protos::types::Payment;
 use prost::Message;
 use rand::Rng;
 use std::fs;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
+use tokio::select;
 
 const USAGE_GUIDE: &str = "Usage: ldk-server <config_path>";
 
@@ -92,7 +95,7 @@ fn main() {
 		},
 	};
 
-	let paginated_store =
+	let paginated_store: Arc<dyn PaginatedKVStore> =
 		Arc::new(match SqliteStore::new(PathBuf::from(config_file.storage_dir_path), None, None) {
 			Ok(store) => store,
 			Err(e) => {
@@ -124,14 +127,14 @@ fn main() {
 			Err(e) => {
 				println!("Failed to register for SIGTERM stream: {}", e);
 				std::process::exit(-1);
-			},
+			}
 		};
 		let event_node = Arc::clone(&node);
 		let rest_svc_listener = TcpListener::bind(config_file.rest_service_addr)
 			.await
 			.expect("Failed to bind listening port");
 		loop {
-			tokio::select! {
+			select! {
 				event = event_node.next_event_async() => {
 					match event {
 						Event::ChannelPending { channel_id, counterparty_node_id, .. } => {
@@ -154,18 +157,44 @@ fn main() {
 								payment_id, payment_hash, amount_msat
 							);
 							let payment_id = payment_id.expect("PaymentId expected for ldk-server >=0.1");
-							upsert_payment_details(&event_node, Arc::clone(&paginated_store) as Arc<dyn PaginatedKVStore>, &payment_id);
+
+							publish_event_and_upsert_payment(&payment_id,
+								|payment_ref| event_envelope::Event::PaymentReceived(events::PaymentReceived {
+									payment: Some(payment_ref.clone()),
+								}),
+								&event_node,
+								Arc::clone(&event_publisher),
+								Arc::clone(&paginated_store)).await;
 						},
 						Event::PaymentSuccessful {payment_id, ..} => {
 							let payment_id = payment_id.expect("PaymentId expected for ldk-server >=0.1");
-							upsert_payment_details(&event_node, Arc::clone(&paginated_store) as Arc<dyn PaginatedKVStore>, &payment_id);
+
+							publish_event_and_upsert_payment(&payment_id,
+								|payment_ref| event_envelope::Event::PaymentSuccessful(events::PaymentSuccessful {
+									payment: Some(payment_ref.clone()),
+								}),
+								&event_node,
+								Arc::clone(&event_publisher),
+								Arc::clone(&paginated_store)).await;
 						},
 						Event::PaymentFailed {payment_id, ..} => {
 							let payment_id = payment_id.expect("PaymentId expected for ldk-server >=0.1");
-							upsert_payment_details(&event_node, Arc::clone(&paginated_store) as Arc<dyn PaginatedKVStore>, &payment_id);
+
+							publish_event_and_upsert_payment(&payment_id,
+								|payment_ref| event_envelope::Event::PaymentFailed(events::PaymentFailed {
+									payment: Some(payment_ref.clone()),
+								}),
+								&event_node,
+								Arc::clone(&event_publisher),
+								Arc::clone(&paginated_store)).await;
 						},
 						Event::PaymentClaimable {payment_id, ..} => {
-							upsert_payment_details(&event_node, Arc::clone(&paginated_store) as Arc<dyn PaginatedKVStore>, &payment_id);
+							if let Some(payment_details) = event_node.payment(&payment_id) {
+								let payment = payment_to_proto(payment_details);
+								upsert_payment_details(&event_node, Arc::clone(&paginated_store), &payment);
+							} else {
+								eprintln!("Unable to find payment with paymentId: {}", payment_id.to_string());
+							}
 						},
 						Event::PaymentForwarded {
 							prev_channel_id,
@@ -234,13 +263,12 @@ fn main() {
 							event_node.event_handled();
 						},
 					}
-
 				},
 				res = rest_svc_listener.accept() => {
 					match res {
 						Ok((stream, _)) => {
 							let io_stream = TokioIo::new(stream);
-							let node_service = NodeService::new(Arc::clone(&node), Arc::clone(&paginated_store) as Arc<dyn PaginatedKVStore + Send + Sync>);
+							let node_service = NodeService::new(Arc::clone(&node), Arc::clone(&paginated_store));
 							runtime.spawn(async move {
 								if let Err(err) = http1::Builder::new().serve_connection(io_stream, node_service).await {
 									eprintln!("Failed to serve connection: {}", err);
@@ -266,30 +294,48 @@ fn main() {
 	println!("Shutdown complete..");
 }
 
-fn upsert_payment_details(
-	event_node: &Node, paginated_store: Arc<dyn PaginatedKVStore>, payment_id: &PaymentId,
+async fn publish_event_and_upsert_payment(
+	payment_id: &PaymentId, payment_to_event: fn(&Payment) -> event_envelope::Event,
+	event_node: &Node, event_publisher: Arc<dyn EventPublisher>,
+	paginated_store: Arc<dyn PaginatedKVStore>,
 ) {
 	if let Some(payment_details) = event_node.payment(payment_id) {
 		let payment = payment_to_proto(payment_details);
-		let time =
-			SystemTime::now().duration_since(UNIX_EPOCH).expect("Time must be > 1970").as_secs()
-				as i64;
-
-		match paginated_store.write(
-			PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE,
-			PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE,
-			&payment_id.0.to_lower_hex_string(),
-			time,
-			&payment.encode_to_vec(),
-		) {
-			Ok(_) => {
-				event_node.event_handled();
-			},
+
+		let event = payment_to_event(&payment);
+		let event_name = get_event_name(&event);
+		match event_publisher.publish(EventEnvelope { event: Some(event) }).await {
+			Ok(_) => {},
 			Err(e) => {
-				eprintln!("Failed to write payment to persistence: {}", e);
+				println!("Failed to publish '{}' event, : {}", event_name, e);
+				return;
 			},
-		}
+		};
+
+		upsert_payment_details(event_node, Arc::clone(&paginated_store), &payment);
 	} else {
-		eprintln!("Unable to find payment with paymentId: {}", payment_id.0.to_lower_hex_string());
+		eprintln!("Unable to find payment with paymentId: {}", payment_id);
+	}
+}
+
+fn upsert_payment_details(
+	event_node: &Node, paginated_store: Arc<dyn PaginatedKVStore>, payment: &Payment,
+) {
+	let time =
+		SystemTime::now().duration_since(UNIX_EPOCH).expect("Time must be > 1970").as_secs() as i64;
+
+	match paginated_store.write(
+		PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE,
+		PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE,
+		&payment.id,
+		time,
+		&payment.encode_to_vec(),
+	) {
+		Ok(_) => {
+			event_node.event_handled();
+		},
+		Err(e) => {
+			eprintln!("Failed to write payment to persistence: {}", e);
+		},
 	}
 }

ldk-server/src/service.rs

@@ -39,20 +39,18 @@ use std::sync::Arc;
 #[derive(Clone)]
 pub struct NodeService {
 	node: Arc<Node>,
-	paginated_kv_store: Arc<dyn PaginatedKVStore + Send + Sync>,
+	paginated_kv_store: Arc<dyn PaginatedKVStore>,
 }
 
 impl NodeService {
-	pub(crate) fn new(
-		node: Arc<Node>, paginated_kv_store: Arc<dyn PaginatedKVStore + Send + Sync>,
-	) -> Self {
+	pub(crate) fn new(node: Arc<Node>, paginated_kv_store: Arc<dyn PaginatedKVStore>) -> Self {
 		Self { node, paginated_kv_store }
 	}
 }
 
 pub(crate) struct Context {
 	pub(crate) node: Arc<Node>,
-	pub(crate) paginated_kv_store: Arc<dyn PaginatedKVStore + Send + Sync>,
+	pub(crate) paginated_kv_store: Arc<dyn PaginatedKVStore>,
 }
 
 impl Service<Request<Incoming>> for NodeService {