Merge pull request #51 from G8XSU/2025-03-12-events

Introduce EventPublisher trait.

Author: G8XSU

Cargo.lock

@@ -15,7 +15,12 @@ fn generate_protos() {
 	prost_build::Config::new()
 		.bytes(&["."])
 		.compile_protos(
-			&["src/proto/api.proto", "src/proto/types.proto", "src/proto/error.proto"],
+			&[
+				"src/proto/api.proto",
+				"src/proto/types.proto",
+				"src/proto/events.proto",
+				"src/proto/error.proto",
+			],
 			&["src/proto/"],
 		)
 		.expect("protobuf compilation failed");
@@ -24,6 +29,8 @@ fn generate_protos() {
 	fs::copy(from_path, "src/api.rs").unwrap();
 	let from_path = Path::new(&env::var("OUT_DIR").unwrap()).join("types.rs");
 	fs::copy(from_path, "src/types.rs").unwrap();
+	let from_path = Path::new(&env::var("OUT_DIR").unwrap()).join("events.rs");
+	fs::copy(from_path, "src/events.rs").unwrap();
 	let from_path = Path::new(&env::var("OUT_DIR").unwrap()).join("error.rs");
 	fs::copy(from_path, "src/error.rs").unwrap();
 }

ldk-server-protos/build.rs

@@ -0,0 +1,23 @@
+/// EventEnvelope wraps different event types in a single message to be used by EventPublisher.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct EventEnvelope {
+	#[prost(oneof = "event_envelope::Event", tags = "2")]
+	pub event: ::core::option::Option<event_envelope::Event>,
+}
+/// Nested message and enum types in `EventEnvelope`.
+pub mod event_envelope {
+	#[allow(clippy::derive_partial_eq_without_eq)]
+	#[derive(Clone, PartialEq, ::prost::Oneof)]
+	pub enum Event {
+		#[prost(message, tag = "2")]
+		PaymentForwarded(super::PaymentForwarded),
+	}
+}
+/// PaymentForwarded indicates a payment was forwarded through the node.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct PaymentForwarded {
+	#[prost(message, optional, tag = "1")]
+	pub forwarded_payment: ::core::option::Option<super::types::ForwardedPayment>,
+}

ldk-server-protos/src/events.rs

@@ -1,3 +1,4 @@
 pub mod api;
 pub mod error;
+pub mod events;
 pub mod types;

ldk-server-protos/src/lib.rs

@@ -0,0 +1,15 @@
+syntax = "proto3";
+import "types.proto";
+package events;
+
+// EventEnvelope wraps different event types in a single message to be used by EventPublisher.
+message EventEnvelope {
+  oneof event {
+    PaymentForwarded payment_forwarded = 2;
+  }
+}
+
+// PaymentForwarded indicates a payment was forwarded through the node.
+message PaymentForwarded {
+  types.ForwardedPayment forwarded_payment = 1;
+}

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

@@ -17,3 +17,4 @@ bytes = { version = "1.4.0", default-features = false }
 hex = { package = "hex-conservative", version = "0.2.1", default-features = false }
 rusqlite = { version = "0.31.0", features = ["bundled"] }
 rand = { version = "0.8.5", default-features = false }
+async-trait = { version = "0.1.85", default-features = false }

ldk-server/Cargo.toml

@@ -1,6 +1,6 @@
 use crate::api::error::LdkServerError;
 use crate::api::error::LdkServerErrorCode::InternalServerError;
-use crate::io::{
+use crate::io::persist::{
 	FORWARDED_PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE,
 	FORWARDED_PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE,
 };

ldk-server/src/api/list_forwarded_payments.rs

@@ -1,6 +1,8 @@
 use crate::api::error::LdkServerError;
 use crate::api::error::LdkServerErrorCode::InternalServerError;
-use crate::io::{PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE};
+use crate::io::persist::{
+	PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE,
+};
 use crate::service::Context;
 use bytes::Bytes;
 use ldk_server_protos::api::{ListPaymentsRequest, ListPaymentsResponse};

ldk-server/src/api/list_payments.rs

@@ -0,0 +1,55 @@
+use crate::api::error::LdkServerError;
+use async_trait::async_trait;
+use ldk_server_protos::events::EventEnvelope;
+
+/// A trait for publishing events or notifications from the LDK Server.
+///
+/// Implementors of this trait define how events are sent to various messaging
+/// systems. It provides a consistent, asynchronous interface for event publishing, while allowing
+/// each implementation to manage  its own initialization and configuration, typically sourced from
+/// the `ldk-server.config` file. A no-op implementation is included by default,
+/// with specific implementations enabled via feature flags.
+///
+/// Events are represented as [`EventEnvelope`] messages, which are Protocol Buffers
+/// ([protobuf](https://protobuf.dev/)) objects defined in [`ldk_server_protos::events`].
+/// These events are serialized to bytes by the publisher before transmission, and consumers can
+/// deserialize them using the protobuf definitions.
+///
+/// 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 {
+	/// Publishes an event to the underlying messaging system.
+	///
+	/// # Arguments
+	/// * `event` - The event message to publish, provided as an [`EventEnvelope`]
+	///             defined in [`ldk_server_protos::events`]. Implementors must serialize
+	///             the whole [`EventEnvelope`] to bytes before publishing.
+	///
+	/// In order to ensure no events are lost, implementors of this trait must publish events
+	/// durably to underlying messaging system. An event is considered published when
+	/// [`EventPublisher::publish`] returns `Ok(())`, thus implementors MUST durably persist/publish events *before*
+	/// returning `Ok(())`.
+	///
+	/// # Errors
+	/// May return an [`LdkServerErrorCode::InternalServerError`] if the event cannot be published,
+	/// such as due to network failures, misconfiguration, or transport-specific issues.
+	/// If event publishing fails, the LDK Server will retry publishing the event indefinitely, which
+	/// may degrade performance until the underlying messaging system is operational again.
+	///
+	/// [`LdkServerErrorCode::InternalServerError`]: crate::api::error::LdkServerErrorCode
+	async fn publish(&self, event: EventEnvelope) -> Result<(), LdkServerError>;
+}
+
+pub(crate) struct NoopEventPublisher;
+
+#[async_trait]
+impl EventPublisher for NoopEventPublisher {
+	/// Publishes an event to a no-op sink, effectively discarding it.
+	///
+	/// This implementation does nothing and always returns `Ok(())`, serving as a
+	/// default when no messaging system is configured.
+	async fn publish(&self, _event: EventEnvelope) -> Result<(), LdkServerError> {
+		Ok(())
+	}
+}

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

@@ -0,0 +1 @@
+pub(crate) mod event_publisher;