Introduce EventPublisher trait.

ldk-server-protos/build.rs

@@ -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/src/events.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/lib.rs

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

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

@@ -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/Cargo.toml

@@ -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/src/api/list_forwarded_payments.rs

@@ -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_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/io/events/event_publisher.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/mod.rs

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

ldk-server/src/io/mod.rs

@@ -1,11 +1,3 @@
-pub(crate) mod paginated_kv_store;
-pub(crate) mod sqlite_store;
+pub(crate) mod events;
+pub(crate) mod persist;
 pub(crate) mod utils;
-
-/// The forwarded payments will be persisted under this prefix.
-pub(crate) const FORWARDED_PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE: &str = "forwarded_payments";
-pub(crate) const FORWARDED_PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
-
-/// The payments will be persisted under this prefix.
-pub(crate) const PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE: &str = "payments";
-pub(crate) const PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";

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

@@ -0,0 +1,10 @@
+pub(crate) mod paginated_kv_store;
+pub(crate) mod sqlite_store;
+
+/// The forwarded payments will be persisted under this prefix.
+pub(crate) const FORWARDED_PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE: &str = "forwarded_payments";
+pub(crate) const FORWARDED_PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
+
+/// The payments will be persisted under this prefix.
+pub(crate) const PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE: &str = "payments";
+pub(crate) const PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";

ldk-server/src/io/sqlite_store/mod.rs → ldk-server/src/io/persist/sqlite_store/mod.rs

@@ -1,4 +1,4 @@
-use crate::io::paginated_kv_store::{ListResponse, PaginatedKVStore};
+use crate::io::persist::paginated_kv_store::{ListResponse, PaginatedKVStore};
 use crate::io::utils::check_namespace_key_validity;
 use ldk_node::lightning::types::string::PrintableString;
 use rusqlite::{named_params, Connection};

ldk-server/src/main.rs

@@ -13,9 +13,10 @@ use tokio::signal::unix::SignalKind;
 use hyper::server::conn::http1;
 use hyper_util::rt::TokioIo;
 
-use crate::io::paginated_kv_store::PaginatedKVStore;
-use crate::io::sqlite_store::SqliteStore;
-use crate::io::{
+use crate::io::events::event_publisher::{EventPublisher, NoopEventPublisher};
+use crate::io::persist::paginated_kv_store::PaginatedKVStore;
+use crate::io::persist::sqlite_store::SqliteStore;
+use crate::io::persist::{
 	FORWARDED_PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE,
 	FORWARDED_PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE, PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE,
 	PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE,
@@ -26,6 +27,8 @@ use hex::DisplayHex;
 use ldk_node::config::Config;
 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 prost::Message;
 use rand::Rng;
 use std::fs;
@@ -98,6 +101,8 @@ fn main() {
 			},
 		});
 
+	let event_publisher: Arc<dyn EventPublisher> = Arc::new(NoopEventPublisher);
+
 	println!("Starting up...");
 	match node.start_with_runtime(Arc::clone(&runtime)) {
 		Ok(()) => {},
@@ -200,6 +205,18 @@ fn main() {
 
 							let forwarded_payment_creation_time = SystemTime::now().duration_since(UNIX_EPOCH).expect("Time must be > 1970").as_secs() as i64;
 
+							match event_publisher.publish(EventEnvelope {
+									event: Some(event_envelope::Event::PaymentForwarded(events::PaymentForwarded {
+											forwarded_payment: Some(forwarded_payment.clone()),
+									})),
+							}).await {
+								Ok(_) => {},
+								Err(e) => {
+									println!("Failed to publish 'PaymentForwarded' event: {}", e);
+									continue;
+								}
+							};
+
 							match paginated_store.write(FORWARDED_PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE,FORWARDED_PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE,
 								&forwarded_payment_id.to_lower_hex_string(),
 								forwarded_payment_creation_time,

ldk-server/src/service.rs

@@ -30,7 +30,7 @@ use crate::api::open_channel::{handle_open_channel, OPEN_CHANNEL_PATH};
 use crate::api::update_channel_config::{
 	handle_update_channel_config_request, UPDATE_CHANNEL_CONFIG_PATH,
 };
-use crate::io::paginated_kv_store::PaginatedKVStore;
+use crate::io::persist::paginated_kv_store::PaginatedKVStore;
 use crate::util::proto_adapter::to_error_response;
 use std::future::Future;
 use std::pin::Pin;