Start publishing more payment events using EventPublisher.

Author: G8XSU

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/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>";
 
@@ -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,7 +263,6 @@ fn main() {
 							event_node.event_handled();
 						},
 					}
-
 				},
 				res = rest_svc_listener.accept() => {
 					match res {
@@ -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);
+		},
 	}
 }