Merge pull request #49 from G8XSU/payments-pagination-2

Support pagination in ListPayments API.

Author: G8XSU

ldk-server-cli/src/main.rs

@@ -10,8 +10,9 @@ use ldk_server_client::ldk_server_protos::api::{
 	OnchainReceiveRequest, OnchainSendRequest, OpenChannelRequest,
 };
 use ldk_server_client::ldk_server_protos::types::{
-	bolt11_invoice_description, Bolt11InvoiceDescription,
+	bolt11_invoice_description, Bolt11InvoiceDescription, PageToken, Payment,
 };
+use std::fmt::Debug;
 
 #[derive(Parser, Debug)]
 #[command(version, about, long_about = None)]
@@ -87,7 +88,13 @@ enum Commands {
 		announce_channel: bool,
 	},
 	ListChannels,
-	ListPayments,
+	ListPayments {
+		#[arg(short, long)]
+		#[arg(
+			help = "Minimum number of payments to return. If not provided, only the first page of the paginated list is returned."
+		)]
+		number_of_payments: Option<u64>,
+	},
 }
 
 #[tokio::main]
@@ -186,13 +193,33 @@ async fn main() {
 		Commands::ListChannels => {
 			handle_response_result(client.list_channels(ListChannelsRequest {}).await);
 		},
-		Commands::ListPayments => {
-			handle_response_result(client.list_payments(ListPaymentsRequest {}).await);
+		Commands::ListPayments { number_of_payments } => {
+			handle_response_result(list_n_payments(client, number_of_payments).await);
 		},
 	}
 }
 
-fn handle_response_result<Rs: ::prost::Message>(response: Result<Rs, LdkServerError>) {
+async fn list_n_payments(
+	client: LdkServerClient, number_of_payments: Option<u64>,
+) -> Result<Vec<Payment>, LdkServerError> {
+	let mut payments = Vec::new();
+	let mut page_token: Option<PageToken> = None;
+	// If no count is specified, just list the first page.
+	let target_count = number_of_payments.unwrap_or(0);
+
+	loop {
+		let response = client.list_payments(ListPaymentsRequest { page_token }).await?;
+
+		payments.extend(response.payments);
+		if payments.len() >= target_count as usize || response.next_page_token.is_none() {
+			break;
+		}
+		page_token = response.next_page_token;
+	}
+	Ok(payments)
+}
+
+fn handle_response_result<Rs: Debug>(response: Result<Rs, LdkServerError>) {
 	match response {
 		Ok(response) => {
 			println!("Success: {:?}", response);

ldk-server-protos/src/api.rs

@@ -331,7 +331,16 @@ pub struct GetPaymentDetailsResponse {
 /// See more: <https://docs.rs/ldk-node/latest/ldk_node/struct.Node.html#method.list_payments>
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
-pub struct ListPaymentsRequest {}
+pub struct ListPaymentsRequest {
+	/// `page_token` is a pagination token.
+	///
+	/// To query for the first page, `page_token` must not be specified.
+	///
+	/// For subsequent pages, use the value that was returned as `next_page_token` in the previous
+	/// page's response.
+	#[prost(message, optional, tag = "1")]
+	pub page_token: ::core::option::Option<super::types::PageToken>,
+}
 /// The response `content` for the `ListPayments` API, when HttpStatusCode is OK (200).
 /// When HttpStatusCode is not OK (non-200), the response `content` contains a serialized `ErrorResponse`.
 #[allow(clippy::derive_partial_eq_without_eq)]
@@ -340,6 +349,21 @@ pub struct ListPaymentsResponse {
 	/// List of payments.
 	#[prost(message, repeated, tag = "1")]
 	pub payments: ::prost::alloc::vec::Vec<super::types::Payment>,
+	/// `next_page_token` is a pagination token, used to retrieve the next page of results.
+	/// Use this value to query for next-page of paginated operation, by specifying
+	/// this value as the `page_token` in the next request.
+	///
+	/// If `next_page_token` is `None`, then the "last page" of results has been processed and
+	/// there is no more data to be retrieved.
+	///
+	/// If `next_page_token` is not `None`, it does not necessarily mean that there is more data in the
+	/// result set. The only way to know when you have reached the end of the result set is when
+	/// `next_page_token` is `None`.
+	///
+	/// **Caution**: Clients must not assume a specific number of records to be present in a page for
+	/// paginated response.
+	#[prost(message, optional, tag = "2")]
+	pub next_page_token: ::core::option::Option<super::types::PageToken>,
 }
 /// Retrieves list of all forwarded payments.
 /// See more: <https://docs.rs/ldk-node/latest/ldk_node/enum.Event.html#variant.PaymentForwarded>

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

@@ -314,13 +314,36 @@ message GetPaymentDetailsResponse {
 
 // Retrieves list of all payments.
 // See more: https://docs.rs/ldk-node/latest/ldk_node/struct.Node.html#method.list_payments
-message ListPaymentsRequest {}
+message ListPaymentsRequest {
+  // `page_token` is a pagination token.
+  //
+  // To query for the first page, `page_token` must not be specified.
+  //
+  // For subsequent pages, use the value that was returned as `next_page_token` in the previous
+  // page's response.
+  optional types.PageToken page_token = 1;
+}
 
 // The response `content` for the `ListPayments` API, when HttpStatusCode is OK (200).
 // When HttpStatusCode is not OK (non-200), the response `content` contains a serialized `ErrorResponse`.
 message ListPaymentsResponse {
   // List of payments.
   repeated types.Payment payments = 1;
+
+  // `next_page_token` is a pagination token, used to retrieve the next page of results.
+  // Use this value to query for next-page of paginated operation, by specifying
+  // this value as the `page_token` in the next request.
+  //
+  // If `next_page_token` is `None`, then the "last page" of results has been processed and
+  // there is no more data to be retrieved.
+  //
+  // If `next_page_token` is not `None`, it does not necessarily mean that there is more data in the
+  // result set. The only way to know when you have reached the end of the result set is when
+  // `next_page_token` is `None`.
+  //
+  // **Caution**: Clients must not assume a specific number of records to be present in a page for
+  // paginated response.
+  optional types.PageToken next_page_token = 2;
 }
 
 // Retrieves list of all forwarded payments.

ldk-server/src/api/list_payments.rs

@@ -1,15 +1,54 @@
 use crate::api::error::LdkServerError;
+use crate::api::error::LdkServerErrorCode::InternalServerError;
+use crate::io::{PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE};
 use crate::service::Context;
-use crate::util::proto_adapter::payment_to_proto;
+use bytes::Bytes;
 use ldk_server_protos::api::{ListPaymentsRequest, ListPaymentsResponse};
+use ldk_server_protos::types::{PageToken, Payment};
+use prost::Message;
 
 pub(crate) const LIST_PAYMENTS_PATH: &str = "ListPayments";
 
 pub(crate) fn handle_list_payments_request(
-	context: Context, _request: ListPaymentsRequest,
+	context: Context, request: ListPaymentsRequest,
 ) -> Result<ListPaymentsResponse, LdkServerError> {
-	let payments = context.node.list_payments().into_iter().map(|p| payment_to_proto(p)).collect();
+	let page_token = request.page_token.map(|p| (p.token, p.index));
+	let list_response = context
+		.paginated_kv_store
+		.list(
+			PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE,
+			PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE,
+			page_token,
+		)
+		.map_err(|e| {
+			LdkServerError::new(InternalServerError, format!("Failed to list payments: {}", e))
+		})?;
 
-	let response = ListPaymentsResponse { payments };
+	let mut payments: Vec<Payment> = Vec::with_capacity(list_response.keys.len());
+	for key in list_response.keys {
+		let payment_bytes = context
+			.paginated_kv_store
+			.read(
+				PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE,
+				PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE,
+				&key,
+			)
+			.map_err(|e| {
+				LdkServerError::new(
+					InternalServerError,
+					format!("Failed to read payment data: {}", e),
+				)
+			})?;
+		let payment = Payment::decode(Bytes::from(payment_bytes)).map_err(|e| {
+			LdkServerError::new(InternalServerError, format!("Failed to decode payment: {}", e))
+		})?;
+		payments.push(payment);
+	}
+	let response = ListPaymentsResponse {
+		payments,
+		next_page_token: list_response
+			.next_page_token
+			.map(|(token, index)| PageToken { token, index }),
+	};
 	Ok(response)
 }

ldk-server/src/io/mod.rs

@@ -5,3 +5,7 @@ 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/main.rs

@@ -5,7 +5,7 @@ mod util;
 
 use crate::service::NodeService;
 
-use ldk_node::{Builder, Event};
+use ldk_node::{Builder, Event, Node};
 
 use tokio::net::TcpListener;
 use tokio::signal::unix::SignalKind;
@@ -17,12 +17,14 @@ 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,
+	FORWARDED_PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE, PAYMENTS_PERSISTENCE_PRIMARY_NAMESPACE,
+	PAYMENTS_PERSISTENCE_SECONDARY_NAMESPACE,
 };
 use crate::util::config::load_config;
-use crate::util::proto_adapter::forwarded_payment_to_proto;
+use crate::util::proto_adapter::{forwarded_payment_to_proto, payment_to_proto};
 use hex::DisplayHex;
 use ldk_node::config::Config;
+use ldk_node::lightning::ln::channelmanager::PaymentId;
 use ldk_node::logger::LogLevel;
 use prost::Message;
 use rand::Rng;
@@ -146,7 +148,19 @@ fn main() {
 								"PAYMENT_RECEIVED: with id {:?}, hash {}, amount_msat {}",
 								payment_id, payment_hash, amount_msat
 							);
-							event_node.event_handled();
+							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);
+						},
+						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);
+						},
+						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);
+						},
+						Event::PaymentClaimable {payment_id, ..} => {
+							upsert_payment_details(&event_node, Arc::clone(&paginated_store) as Arc<dyn PaginatedKVStore>, &payment_id);
 						},
 						Event::PaymentForwarded {
 							prev_channel_id,
@@ -234,3 +248,31 @@ fn main() {
 	node.stop().expect("Shutdown should always succeed.");
 	println!("Shutdown complete..");
 }
+
+fn upsert_payment_details(
+	event_node: &Node, paginated_store: Arc<dyn PaginatedKVStore>, payment_id: &PaymentId,
+) {
+	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();
+			},
+			Err(e) => {
+				eprintln!("Failed to write payment to persistence: {}", e);
+			},
+		}
+	} else {
+		eprintln!("Unable to find payment with paymentId: {}", payment_id.0.to_lower_hex_string());
+	}
+}