Add VSS Store Implementation

Author: G8XSU

Cargo.toml

@@ -78,6 +78,7 @@ tokio = { version = "1", default-features = false, features = [ "rt-multi-thread
 esplora-client = { version = "0.4", default-features = false }
 libc = "0.2"
 uniffi = { version = "0.23.0", features = ["build"], optional = true }
+vss-client = "0.0.1-alpha"
 
 [target.'cfg(windows)'.dependencies]
 winapi = { version = "0.3", features = ["winbase"] }

src/io/vss_store.rs

@@ -0,0 +1,179 @@
+use std::io::Cursor;
+use std::io::Read;
+use std::sync::Arc;
+use std::{error::Error, io};
+
+use super::*;
+use crate::KVStore;
+use lightning::util::persist::KVStorePersister;
+use lightning::util::ser::Writeable;
+use tokio::runtime::Runtime;
+use vss_client::client::VssClient;
+use vss_client::error::VssError;
+use vss_client::types::{
+	DeleteObjectRequest, GetObjectRequest, KeyValue, ListKeyVersionsRequest, PutObjectRequest,
+};
+
+/// Implements [KVStore] and [KVStorePersister] for VSS using [VssClient].
+///
+/// A [`KVStore`] implementation that writes to and reads from VSS using [VssClient].
+///
+/// Learn more about Versioned Storage Service (VSS) [here](https://github.com/lightningdevkit/vss-server/blob/main/README.md).
+pub struct VssKVStore {
+	client: VssClient,
+	store_id: String,
+	runtime: Arc<Runtime>,
+}
+
+impl VssKVStore {
+	pub(crate) fn new(base_url: &str, store_id: String) -> Self {
+		let client = VssClient::new(base_url);
+		let runtime =
+			Arc::new(tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap());
+		Self { client, store_id, runtime }
+	}
+
+	fn build_key(&self, namespace: &str, key: &str) -> io::Result<String> {
+		if key.is_empty() {
+			return Err(io::Error::new(io::ErrorKind::Other, "Empty key is not allowed"));
+		}
+		if namespace.is_empty() {
+			Ok(key.to_string())
+		} else {
+			Ok(format!("{}/{}", namespace, key))
+		}
+	}
+
+	fn split_key(&self, key: &str) -> (String, String) {
+		get_namespace_and_key_from_prefixed(key).unwrap()
+	}
+
+	async fn list_all_keys(&self, namespace: &str) -> Result<Vec<String>, VssError> {
+		let mut page_token = None;
+		let mut keys = vec![];
+
+		while page_token != Some("".to_string()) {
+			let request = ListKeyVersionsRequest {
+				store_id: self.store_id.to_string(),
+				key_prefix: Some(namespace.to_string()),
+				page_token,
+				page_size: None,
+			};
+
+			let response = self.client.list_key_versions(&request).await?;
+
+			for kv in response.key_versions {
+				keys.push(self.split_key(&kv.key).1);
+			}
+			page_token = response.next_page_token;
+		}
+		Ok(keys)
+	}
+}
+
+impl KVStore for VssKVStore {
+	type Reader = Cursor<Vec<u8>>;
+
+	fn read(&self, namespace: &str, key: &str) -> io::Result<Self::Reader> {
+		let request = GetObjectRequest {
+			store_id: self.store_id.to_string(),
+			key: self.build_key(namespace, key)?,
+		};
+
+		let resp =
+			tokio::task::block_in_place(|| self.runtime.block_on(self.client.get_object(&request)))
+				.map_err(|e| {
+					let msg = format!("Failed to read from key {}/{}: {}", namespace, key, e);
+					io::Error::new(io::ErrorKind::Other, msg)
+				})?;
+
+		let value = resp.value.unwrap().value;
+		if value.is_empty() {
+			let msg = format!("Failed to read as key could not be found: {}/{}", namespace, key);
+			Err(io::Error::new(io::ErrorKind::NotFound, msg))
+		} else {
+			Ok(Cursor::new(value))
+		}
+	}
+
+	fn write(&self, namespace: &str, key: &str, buf: &[u8]) -> io::Result<()> {
+		let request = PutObjectRequest {
+			store_id: self.store_id.to_string(),
+			global_version: None,
+			transaction_items: vec![KeyValue {
+				key: self.build_key(namespace, key)?,
+				version: -1,
+				value: buf.to_vec(),
+			}],
+			delete_items: vec![],
+		};
+
+		tokio::task::block_in_place(|| self.runtime.block_on(self.client.put_object(&request)))
+			.map_err(|e| {
+				let msg = format!("Failed to write to key {}/{}: {}", namespace, key, e);
+				io::Error::new(io::ErrorKind::Other, msg)
+			})?;
+
+		Ok(())
+	}
+
+	fn remove(&self, namespace: &str, key: &str) -> io::Result<()> {
+		let request = DeleteObjectRequest {
+			store_id: self.store_id.to_string(),
+			key_value: Some(KeyValue {
+				key: self.build_key(namespace, key)?,
+				version: -1,
+				value: vec![],
+			}),
+		};
+
+		tokio::task::block_in_place(|| self.runtime.block_on(self.client.delete_object(&request)))
+			.map_err(|e| {
+				let msg = format!("Failed to delete key {}/{}: {}", namespace, key, e);
+				io::Error::new(io::ErrorKind::Other, msg)
+			})?;
+		Ok(())
+	}
+
+	fn list(&self, namespace: &str) -> io::Result<Vec<String>> {
+		let keys =
+			tokio::task::block_in_place(|| self.runtime.block_on(self.list_all_keys(namespace)))
+				.map_err(|e| {
+					let msg =
+						format!("Failed to retrieve keys in namespace: {} : {}", namespace, e);
+					io::Error::new(io::ErrorKind::Other, msg)
+				})?;
+
+		Ok(keys)
+	}
+}
+
+impl KVStorePersister for VssKVStore {
+	fn persist<W: Writeable>(&self, prefixed_key: &str, object: &W) -> io::Result<()> {
+		self.write("", &prefixed_key, &object.encode())?;
+		Ok(())
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::test::utils::random_storage_path;
+
+	use proptest::prelude::*;
+	proptest! {
+		#[test]
+		fn read_write_remove_list_persist(data in any::<[u8; 32]>()) {
+				let vss_base_url = std::env::var("TEST_VSS_BASE_URL");
+				if vss_base_url.is_ok()
+				{
+						let rand_store_id = random_storage_path();
+						let vss_store = VssKVStore::new(&vss_base_url.unwrap(), rand_store_id);
+
+						do_read_write_remove_list_persist(&data, &vss_store);
+				}else{
+						eprintln!("** SKIPPING `VssKVStore` test-suite since environment variable `TEST_VSS_BASE_URL` is not set **");
+				}
+		}
+	}
+}