Add `LSPS5ServiceHandler` persistence

diff --git a/lightning-liquidity/src/lsps5/service.rs b/lightning-liquidity/src/lsps5/service.rs
index e9a8eff..ff1a27c 100644
--- a/lightning-liquidity/src/lsps5/service.rs
+++ b/lightning-liquidity/src/lsps5/service.rs
@@ -17,6 +17,9 @@ use crate::lsps5::msgs::{
 	SetWebhookRequest, SetWebhookResponse, WebhookNotification, WebhookNotificationMethod,
 };
 use crate::message_queue::MessageQueue;
+use crate::persist::{
+	LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+};
 use crate::prelude::*;
 use crate::sync::{Arc, Mutex, RwLock, RwLockWriteGuard};
 use crate::utils::time::TimeProvider;
@@ -28,6 +31,8 @@ use lightning::ln::channelmanager::AChannelManager;
 use lightning::ln::msgs::{ErrorAction, LightningError};
 use lightning::sign::NodeSigner;
 use lightning::util::logger::Level;
+use lightning::util::persist::KVStore;
+use lightning::util::ser::Writeable;
 
 use core::ops::Deref;
 use core::time::Duration;
@@ -118,10 +123,11 @@ impl Default for LSPS5ServiceConfig {
 /// [`LSPS5ServiceEvent::SendWebhookNotification`]: super::event::LSPS5ServiceEvent::SendWebhookNotification
 /// [`app_name`]: super::msgs::LSPS5AppName
 /// [`lsps5.webhook_registered`]: super::msgs::WebhookNotificationMethod::LSPS5WebhookRegistered
-pub struct LSPS5ServiceHandler<CM: Deref, NS: Deref, TP: Deref>
+pub struct LSPS5ServiceHandler<CM: Deref, NS: Deref, K: Deref + Clone, TP: Deref>
 where
 	CM::Target: AChannelManager,
 	NS::Target: NodeSigner,
+	K::Target: KVStore,
 	TP::Target: TimeProvider,
 {
 	config: LSPS5ServiceConfig,
@@ -131,19 +137,21 @@ where
 	time_provider: TP,
 	channel_manager: CM,
 	node_signer: NS,
+	kv_store: K,
 	last_pruning: Mutex<Option<LSPSDateTime>>,
 }
 
-impl<CM: Deref, NS: Deref, TP: Deref> LSPS5ServiceHandler<CM, NS, TP>
+impl<CM: Deref, NS: Deref, K: Deref + Clone, TP: Deref> LSPS5ServiceHandler<CM, NS, K, TP>
 where
 	CM::Target: AChannelManager,
 	NS::Target: NodeSigner,
+	K::Target: KVStore,
 	TP::Target: TimeProvider,
 {
 	/// Constructs a `LSPS5ServiceHandler` using the given time provider.
 	pub(crate) fn new_with_time_provider(
 		event_queue: Arc<EventQueue>, pending_messages: Arc<MessageQueue>, channel_manager: CM,
-		node_signer: NS, config: LSPS5ServiceConfig, time_provider: TP,
+		kv_store: K, node_signer: NS, config: LSPS5ServiceConfig, time_provider: TP,
 	) -> Self {
 		assert!(config.max_webhooks_per_client > 0, "`max_webhooks_per_client` must be > 0");
 		Self {
@@ -154,6 +162,7 @@ where
 			time_provider,
 			channel_manager,
 			node_signer,
+			kv_store,
 			last_pruning: Mutex::new(None),
 		}
 	}
@@ -186,6 +195,51 @@ where
 		}
 	}
 
+	async fn persist_peer_state(
+		&self, counterparty_node_id: PublicKey,
+	) -> Result<(), lightning::io::Error> {
+		let fut = {
+			let outer_state_lock = self.per_peer_state.read().unwrap();
+			let encoded = match outer_state_lock.get(&counterparty_node_id) {
+				None => {
+					let err = lightning::io::Error::new(
+						lightning::io::ErrorKind::Other,
+						"Failed to get peer entry",
+					);
+					return Err(err);
+				},
+				Some(entry) => entry.encode(),
+			};
+
+			let key = counterparty_node_id.to_string();
+
+			self.kv_store.write(
+				LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+				LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+				&key,
+				encoded,
+			)
+		};
+
+		fut.await
+	}
+
+	pub(crate) async fn persist(&self) -> Result<(), lightning::io::Error> {
+		// TODO: We should eventually persist in parallel, however, when we do, we probably want to
+		// introduce some batching to upper-bound the number of requests inflight at any given
+		// time.
+		let need_persist: Vec<PublicKey> = {
+			let outer_state_lock = self.per_peer_state.read().unwrap();
+			outer_state_lock.iter().filter_map(|(k, v)| Some(*k)).collect()
+		};
+
+		for counterparty_node_id in need_persist.into_iter() {
+			self.persist_peer_state(counterparty_node_id).await?;
+		}
+
+		Ok(())
+	}
+
 	fn check_prune_stale_webhooks<'a>(
 		&self, outer_state_lock: &mut RwLockWriteGuard<'a, HashMap<PublicKey, PeerState>>,
 	) {
@@ -549,10 +603,12 @@ where
 	}
 }
 
-impl<CM: Deref, NS: Deref, TP: Deref> LSPSProtocolMessageHandler for LSPS5ServiceHandler<CM, NS, TP>
+impl<CM: Deref, NS: Deref, K: Deref + Clone, TP: Deref> LSPSProtocolMessageHandler
+	for LSPS5ServiceHandler<CM, NS, K, TP>
 where
 	CM::Target: AChannelManager,
 	NS::Target: NodeSigner,
+	K::Target: KVStore,
 	TP::Target: TimeProvider,
 {
 	type ProtocolMessage = LSPS5Message;
diff --git a/lightning-liquidity/src/manager.rs b/lightning-liquidity/src/manager.rs
index 9a6ff26..86343d5 100644
--- a/lightning-liquidity/src/manager.rs
+++ b/lightning-liquidity/src/manager.rs
@@ -303,7 +303,7 @@ pub struct LiquidityManager<
 	lsps1_client_handler: Option<LSPS1ClientHandler<ES>>,
 	lsps2_service_handler: Option<LSPS2ServiceHandler<CM, K>>,
 	lsps2_client_handler: Option<LSPS2ClientHandler<ES>>,
-	lsps5_service_handler: Option<LSPS5ServiceHandler<CM, NS, TP>>,
+	lsps5_service_handler: Option<LSPS5ServiceHandler<CM, NS, K, TP>>,
 	lsps5_client_handler: Option<LSPS5ClientHandler<ES>>,
 	service_config: Option<LiquidityServiceConfig>,
 	_client_config: Option<LiquidityClientConfig>,
@@ -423,7 +423,7 @@ where
 		let lsps5_service_handler = service_config.as_ref().and_then(|config| {
 			config.lsps5_service_config.as_ref().map(|config| {
 				if let Some(number) =
-					<LSPS5ServiceHandler<CM, NS, TP> as LSPSProtocolMessageHandler>::PROTOCOL_NUMBER
+					<LSPS5ServiceHandler<CM, NS, K, TP> as LSPSProtocolMessageHandler>::PROTOCOL_NUMBER
 				{
 					supported_protocols.push(number);
 				}
@@ -432,6 +432,7 @@ where
 					Arc::clone(&pending_events),
 					Arc::clone(&pending_messages),
 					channel_manager.clone(),
+					kv_store.clone(),
 					node_signer,
 					config.clone(),
 					time_provider,
@@ -552,7 +553,7 @@ where
 	/// Returns a reference to the LSPS5 server-side handler.
 	///
 	/// The returned handler allows to initiate the LSPS5 service-side flow.
-	pub fn lsps5_service_handler(&self) -> Option<&LSPS5ServiceHandler<CM, NS, TP>> {
+	pub fn lsps5_service_handler(&self) -> Option<&LSPS5ServiceHandler<CM, NS, K, TP>> {
 		self.lsps5_service_handler.as_ref()
 	}
 
@@ -623,6 +624,10 @@ where
 			lsps2_service_handler.persist().await?;
 		}
 
+		if let Some(lsps5_service_handler) = self.lsps5_service_handler.as_ref() {
+			lsps5_service_handler.persist().await?;
+		}
+
 		Ok(())
 	}
 
@@ -1142,7 +1147,9 @@ where
 	/// Returns a reference to the LSPS5 server-side handler.
 	///
 	/// Wraps [`LiquidityManager::lsps5_service_handler`].
-	pub fn lsps5_service_handler(&self) -> Option<&LSPS5ServiceHandler<CM, NS, TP>> {
+	pub fn lsps5_service_handler(
+		&self,
+	) -> Option<&LSPS5ServiceHandler<CM, NS, Arc<KVStoreSyncWrapper<KS>>, TP>> {
 		self.inner.lsps5_service_handler()
 	}
 
diff --git a/lightning-liquidity/src/persist.rs b/lightning-liquidity/src/persist.rs
index 7617142..f90b3ed 100644
--- a/lightning-liquidity/src/persist.rs
+++ b/lightning-liquidity/src/persist.rs
@@ -18,3 +18,8 @@ pub const LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE: &str = "lightning_liq
 ///
 /// [`LSPS2ServiceHandler`]: crate::lsps2::service::LSPS2ServiceHandler
 pub const LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE: &str = "lsps2_service";
+
+/// The secondary namespace under which the [`LSPS5ServiceHandler`] data will be persisted.
+///
+/// [`LSPS5ServiceHandler`]: crate::lsps5::service::LSPS5ServiceHandler
+pub const LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE: &str = "lsps5_service";