Read persisted LSPS5 service state in `LiquidityManager::new`

diff --git a/lightning-liquidity/src/lsps5/service.rs b/lightning-liquidity/src/lsps5/service.rs
index a4a636c..a2dee15 100644
--- a/lightning-liquidity/src/lsps5/service.rs
+++ b/lightning-liquidity/src/lsps5/service.rs
@@ -150,13 +150,15 @@ where
 {
 	/// Constructs a `LSPS5ServiceHandler` using the given time provider.
 	pub(crate) fn new_with_time_provider(
-		event_queue: Arc<EventQueue<K>>, pending_messages: Arc<MessageQueue>, channel_manager: CM,
-		kv_store: K, node_signer: NS, config: LSPS5ServiceConfig, time_provider: TP,
+		peer_states: HashMap<PublicKey, PeerState>, event_queue: Arc<EventQueue<K>>,
+		pending_messages: Arc<MessageQueue>, channel_manager: CM, 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");
+		let per_peer_state = RwLock::new(peer_states);
 		Self {
 			config,
-			per_peer_state: RwLock::new(new_hash_map()),
+			per_peer_state,
 			event_queue,
 			pending_messages,
 			time_provider,
@@ -648,7 +650,7 @@ where
 }
 
 #[derive(Debug, Default)]
-struct PeerState {
+pub(crate) struct PeerState {
 	webhooks: Vec<(LSPS5AppName, Webhook)>,
 }
 
diff --git a/lightning-liquidity/src/manager.rs b/lightning-liquidity/src/manager.rs
index 76403f1..6a899c6 100644
--- a/lightning-liquidity/src/manager.rs
+++ b/lightning-liquidity/src/manager.rs
@@ -24,7 +24,7 @@ use crate::lsps5::client::{LSPS5ClientConfig, LSPS5ClientHandler};
 use crate::lsps5::msgs::LSPS5Message;
 use crate::lsps5::service::{LSPS5ServiceConfig, LSPS5ServiceHandler};
 use crate::message_queue::MessageQueue;
-use crate::persist::read_lsps2_service_peer_states;
+use crate::persist::{read_lsps2_service_peer_states, read_lsps5_service_peer_states};
 
 use crate::lsps1::client::{LSPS1ClientConfig, LSPS1ClientHandler};
 use crate::lsps1::msgs::LSPS1Message;
@@ -434,25 +434,31 @@ where
 			})
 		});
 
-		let lsps5_service_handler = service_config.as_ref().and_then(|config| {
-			config.lsps5_service_config.as_ref().map(|config| {
+		let lsps5_service_handler = if let Some(service_config) = service_config.as_ref() {
+			if let Some(lsps5_service_config) = service_config.lsps5_service_config.as_ref() {
 				if let Some(number) =
 					<LSPS5ServiceHandler<CM, NS, K, TP> as LSPSProtocolMessageHandler>::PROTOCOL_NUMBER
 				{
 					supported_protocols.push(number);
 				}
 
-				LSPS5ServiceHandler::new_with_time_provider(
+				let peer_states = read_lsps5_service_peer_states(kv_store.clone()).await?;
+				Some(LSPS5ServiceHandler::new_with_time_provider(
+					peer_states,
 					Arc::clone(&pending_events),
 					Arc::clone(&pending_messages),
 					channel_manager.clone(),
 					kv_store.clone(),
 					node_signer,
-					config.clone(),
+					lsps5_service_config.clone(),
 					time_provider,
-				)
-			})
-		});
+				))
+			} else {
+				None
+			}
+		} else {
+			None
+		};
 
 		let lsps1_client_handler = client_config.as_ref().and_then(|config| {
 			config.lsps1_client_config.as_ref().map(|config| {
diff --git a/lightning-liquidity/src/persist.rs b/lightning-liquidity/src/persist.rs
index 1f7e60d..5b8a63e 100644
--- a/lightning-liquidity/src/persist.rs
+++ b/lightning-liquidity/src/persist.rs
@@ -10,6 +10,7 @@
 //! Types and utils for persistence.
 
 use crate::lsps2::service::PeerState as LSPS2ServicePeerState;
+use crate::lsps5::service::PeerState as LSPS5ServicePeerState;
 use crate::prelude::{new_hash_map, HashMap};
 use crate::sync::Mutex;
 
@@ -90,3 +91,47 @@ where
 	}
 	Ok(res)
 }
+
+pub(crate) async fn read_lsps5_service_peer_states<K: Deref>(
+	kv_store: K,
+) -> Result<HashMap<PublicKey, LSPS5ServicePeerState>, lightning::io::Error>
+where
+	K::Target: KVStore,
+{
+	let mut res = new_hash_map();
+
+	for stored_key in kv_store
+		.list(
+			LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+			LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+		)
+		.await?
+	{
+		let mut reader = Cursor::new(
+			kv_store
+				.read(
+					LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+					LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+					&stored_key,
+				)
+				.await?,
+		);
+
+		let peer_state = LSPS5ServicePeerState::read(&mut reader).map_err(|_| {
+			lightning::io::Error::new(
+				lightning::io::ErrorKind::InvalidData,
+				"Failed to deserialize LSPS5 peer state",
+			)
+		})?;
+
+		let key = PublicKey::from_str(&stored_key).map_err(|_| {
+			lightning::io::Error::new(
+				lightning::io::ErrorKind::InvalidData,
+				"Failed to deserialize stored key entry",
+			)
+		})?;
+
+		res.insert(key, peer_state);
+	}
+	Ok(res)
+}