Start storing ForwardedPayments using PaginatedKVStore.

Author: G8XSU

ldk-server/Cargo.toml

@@ -16,6 +16,4 @@ ldk-server-protos = { path = "../ldk-server-protos" }
 bytes = "1.4.0"
 hex = { package = "hex-conservative", version = "0.2.1", default-features = false }
 rusqlite = { version = "0.31.0", features = ["bundled"] }
-
-[dev-dependencies]
 rand = "0.8.5"

ldk-server/src/io/mod.rs

@@ -1,3 +1,7 @@
 pub(crate) mod paginated_kv_store;
 pub(crate) mod sqlite_store;
 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 = "";

ldk-server/src/main.rs

@@ -13,11 +13,22 @@ 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::{
+	FORWARDED_PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE,
+	FORWARDED_PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE,
+};
 use crate::util::config::load_config;
+use crate::util::proto_adapter::forwarded_payment_to_proto;
+use hex::DisplayHex;
 use ldk_node::config::Config;
+use prost::Message;
+use rand::Rng;
 use std::fs;
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::sync::Arc;
+use std::time::{SystemTime, UNIX_EPOCH};
 
 const USAGE_GUIDE: &str = "Usage: ldk-server <config_path>";
 
@@ -44,7 +55,7 @@ fn main() {
 	let config_file = load_config(Path::new(arg)).expect("Invalid configuration file.");
 
 	ldk_node_config.log_level = LogLevel::Trace;
-	ldk_node_config.storage_dir_path = config_file.storage_dir_path;
+	ldk_node_config.storage_dir_path = config_file.storage_dir_path.clone();
 	ldk_node_config.listening_addresses = Some(vec![config_file.listening_addr]);
 	ldk_node_config.network = config_file.network;
 
@@ -75,6 +86,15 @@ fn main() {
 		},
 	};
 
+	let paginated_store =
+		Arc::new(match SqliteStore::new(PathBuf::from(config_file.storage_dir_path), None, None) {
+			Ok(store) => store,
+			Err(e) => {
+				eprintln!("Failed to create SqliteStore: {:?}", e);
+				std::process::exit(-1);
+			},
+		});
+
 	println!("Starting up...");
 	match node.start_with_runtime(Arc::clone(&runtime)) {
 		Ok(()) => {},
@@ -111,22 +131,74 @@ fn main() {
 								"CHANNEL_PENDING: {} from counterparty {}",
 								channel_id, counterparty_node_id
 							);
+							event_node.event_handled();
 						},
 						Event::ChannelReady { channel_id, counterparty_node_id, .. } => {
 							println!(
 								"CHANNEL_READY: {} from counterparty {:?}",
 								channel_id, counterparty_node_id
 							);
+							event_node.event_handled();
 						},
 						Event::PaymentReceived { payment_id, payment_hash, amount_msat } => {
 							println!(
 								"PAYMENT_RECEIVED: with id {:?}, hash {}, amount_msat {}",
 								payment_id, payment_hash, amount_msat
 							);
+							event_node.event_handled();
+						},
+						Event::PaymentForwarded {
+							prev_channel_id,
+							next_channel_id,
+							prev_user_channel_id,
+							next_user_channel_id,
+							total_fee_earned_msat,
+							skimmed_fee_msat,
+							claim_from_onchain_tx,
+							outbound_amount_forwarded_msat
+						} => {
+
+							println!("PAYMENT_FORWARDED: with outbound_amount_forwarded_msat {}, total_fee_earned_msat: {}, inbound channel: {}, outbound channel: {}",
+								outbound_amount_forwarded_msat.unwrap_or(0), total_fee_earned_msat.unwrap_or(0), prev_channel_id, next_channel_id
+							);
+
+							let forwarded_payment = forwarded_payment_to_proto(
+								prev_channel_id,
+								next_channel_id,
+								prev_user_channel_id,
+								next_user_channel_id,
+								total_fee_earned_msat,
+								skimmed_fee_msat,
+								claim_from_onchain_tx,
+								outbound_amount_forwarded_msat
+							);
+
+							// We don't expose this payment-id to the user, it is a temporary measure to generate
+							// some unique identifiers until we have forwarded-payment-id available in ldk.
+							// Currently, this is the expected user handling behaviour for forwarded payments.
+							let mut forwarded_payment_id = [0u8;32];
+							rand::thread_rng().fill(&mut forwarded_payment_id);
+
+							let forwarded_payment_creation_time = SystemTime::now().duration_since(UNIX_EPOCH).expect("Time must be > 1970").as_secs() as i64;
+
+							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,
+								&forwarded_payment.encode_to_vec(),
+							) {
+								Ok(_) => {
+										event_node.event_handled();
+								}
+								Err(e) => {
+										println!("Failed to write forwarded payment to persistence: {}", e);
+								}
+							}
+						},
+						_ => {
+							event_node.event_handled();
 						},
-						_ => {},
 					}
-					event_node.event_handled();
+
 				},
 				res = rest_svc_listener.accept() => {
 					match res {

ldk-server/src/util/proto_adapter.rs

@@ -1,8 +1,9 @@
 use bytes::Bytes;
 use hex::prelude::*;
 use ldk_node::config::{ChannelConfig, MaxDustHTLCExposure};
+use ldk_node::lightning::ln::types::ChannelId;
 use ldk_node::payment::{PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus};
-use ldk_node::{ChannelDetails, LightningBalance, PendingSweepBalance};
+use ldk_node::{ChannelDetails, Event, LightningBalance, PendingSweepBalance, UserChannelId};
 use ldk_server_protos::types::lightning_balance::BalanceType::{
 	ClaimableAwaitingConfirmations, ClaimableOnChannelClose, ContentiousClaimable,
 	CounterpartyRevokedOutputClaimable, MaybePreimageClaimableHtlc, MaybeTimeoutClaimableHtlc,
@@ -13,7 +14,7 @@ use ldk_server_protos::types::payment_kind::Kind::{
 use ldk_server_protos::types::pending_sweep_balance::BalanceType::{
 	AwaitingThresholdConfirmations, BroadcastAwaitingConfirmation, PendingBroadcast,
 };
-use ldk_server_protos::types::{Channel, LspFeeLimits, OutPoint, Payment};
+use ldk_server_protos::types::{Channel, ForwardedPayment, LspFeeLimits, OutPoint, Payment};
 
 pub(crate) fn channel_to_proto(channel: ChannelDetails) -> Channel {
 	Channel {
@@ -320,3 +321,21 @@ pub(crate) fn pending_sweep_balance_to_proto(
 		},
 	}
 }
+
+pub(crate) fn forwarded_payment_to_proto(
+	prev_channel_id: ChannelId, next_channel_id: ChannelId,
+	prev_user_channel_id: Option<UserChannelId>, next_user_channel_id: Option<UserChannelId>,
+	total_fee_earned_msat: Option<u64>, skimmed_fee_msat: Option<u64>, claim_from_onchain_tx: bool,
+	outbound_amount_forwarded_msat: Option<u64>,
+) -> ForwardedPayment {
+	ForwardedPayment {
+		prev_channel_id: prev_channel_id.to_string(),
+		next_channel_id: next_channel_id.to_string(),
+		prev_user_channel_id: prev_user_channel_id.expect("").0.to_string(),
+		next_user_channel_id: next_user_channel_id.map(|u| u.0.to_string()),
+		total_fee_earned_msat,
+		skimmed_fee_msat,
+		claim_from_onchain_tx,
+		outbound_amount_forwarded_msat,
+	}
+}