diff --git a/src/builder.rs b/src/builder.rs
index 70dc7ff7a..c5d9b0503 100644
--- a/src/builder.rs
+++ b/src/builder.rs
@@ -46,7 +46,7 @@ use lightning::routing::scoring::{
 use lightning::sign::EntropySource;
 
 use lightning::util::persist::{
-	read_channel_monitors, CHANNEL_MANAGER_PERSISTENCE_KEY,
+	read_channel_monitors, KVStore, CHANNEL_MANAGER_PERSISTENCE_KEY,
 	CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
 };
 use lightning::util::ser::ReadableArgs;
@@ -579,6 +579,148 @@ impl NodeBuilder {
 	}
 }
 
+#[derive(Debug)]
+pub struct KVStoreBuilder {
+	config: Config,
+	entropy_source_config: Option<EntropySourceConfig>,
+	log_writer_config: Option<LogWriterConfig>,
+}
+
+impl KVStoreBuilder {
+	/// Creates a new builder instance with the default configuration.
+	pub fn new() -> Self {
+		let config = Config::default();
+		Self::from_config(config)
+	}
+
+	/// Creates a new builder instance from an [`Config`].
+	pub fn from_config(config: Config) -> Self {
+		let entropy_source_config = None;
+		let log_writer_config = None;
+		Self { config, entropy_source_config, log_writer_config }
+	}
+
+	//TODO: I think these methods should be extracted out to EntropySourceConfigBuilder. Both NodeBuilder
+	// and KVStoreBuilder can directly take EntropySourceConfig as arg.
+	/// Configures the builder instance to source its entropy from a seed file on disk.
+	///
+	/// If the given file does not exist a new random seed file will be generated and
+	/// stored at the given location.
+	pub fn set_entropy_seed_path(&mut self, seed_path: String) -> &mut Self {
+		self.entropy_source_config = Some(EntropySourceConfig::SeedFile(seed_path));
+		self
+	}
+
+	/// Configures the builder instance to source its entropy from the given 64 seed bytes.
+	pub fn set_entropy_seed_bytes(&mut self, seed_bytes: Vec<u8>) -> Result<&mut Self, BuildError> {
+		if seed_bytes.len() != WALLET_KEYS_SEED_LEN {
+			return Err(BuildError::InvalidSeedBytes);
+		}
+		let mut bytes = [0u8; WALLET_KEYS_SEED_LEN];
+		bytes.copy_from_slice(&seed_bytes);
+		self.entropy_source_config = Some(EntropySourceConfig::SeedBytes(bytes));
+		Ok(self)
+	}
+
+	/// Configures the builder instance to source its entropy from a [BIP 39] mnemonic.
+	///
+	/// [BIP 39]: https://github.com/bitcoin/bips/blob/master/bip-0039.mediawiki
+	pub fn set_entropy_bip39_mnemonic(
+		&mut self, mnemonic: Mnemonic, passphrase: Option<String>,
+	) -> &mut Self {
+		self.entropy_source_config =
+			Some(EntropySourceConfig::Bip39Mnemonic { mnemonic, passphrase });
+		self
+	}
+
+	//TODO: Once we have KVStoreBuilder, we can have NodeBuilder directly take Arc<dyn KVStore> as arg.
+	/// Builds a [`VssStore`] backend based implementation of [`KVStore`].
+	pub fn build_vss_store(
+		&self, vss_url: String, store_id: String, lnurl_auth_server_url: String,
+		fixed_headers: HashMap<String, String>,
+	) -> Result<VssStore, BuildError> {
+		use bitcoin::key::Secp256k1;
+
+		let logger = setup_logger(&self.log_writer_config, &self.config)?;
+
+		let seed_bytes = seed_bytes_from_config(
+			&self.config,
+			self.entropy_source_config.as_ref(),
+			Arc::clone(&logger),
+		)?;
+
+		let config = Arc::new(self.config.clone());
+
+		let vss_xprv = derive_vss_xprv(config, &seed_bytes, Arc::clone(&logger))?;
+
+		let lnurl_auth_xprv = vss_xprv
+			.derive_priv(&Secp256k1::new(), &[ChildNumber::Hardened { index: 138 }])
+			.map_err(|e| {
+				log_error!(logger, "Failed to derive VSS secret: {}", e);
+				BuildError::KVStoreSetupFailed
+			})?;
+
+		let lnurl_auth_jwt_provider =
+			LnurlAuthToJwtProvider::new(lnurl_auth_xprv, lnurl_auth_server_url, fixed_headers)
+				.map_err(|e| {
+					log_error!(logger, "Failed to create LnurlAuthToJwtProvider: {}", e);
+					BuildError::KVStoreSetupFailed
+				})?;
+
+		let header_provider = Arc::new(lnurl_auth_jwt_provider);
+
+		self.build_vss_store_with_header_provider(vss_url, store_id, header_provider)
+	}
+
+	/// Builds a [`VssStore`] backend based implementation of [`KVStore`].
+	pub fn build_vss_store_with_header_provider(
+		&self, vss_url: String, store_id: String, header_provider: Arc<dyn VssHeaderProvider>,
+	) -> Result<VssStore, BuildError> {
+		let logger = setup_logger(&self.log_writer_config, &self.config)?;
+
+		let seed_bytes = seed_bytes_from_config(
+			&self.config,
+			self.entropy_source_config.as_ref(),
+			Arc::clone(&logger),
+		)?;
+
+		let config = Arc::new(self.config.clone());
+
+		let vss_xprv = derive_vss_xprv(config.clone(), &seed_bytes, Arc::clone(&logger))?;
+
+		let vss_seed_bytes: [u8; 32] = vss_xprv.private_key.secret_bytes();
+
+		Ok(VssStore::new(vss_url, store_id, vss_seed_bytes, header_provider).map_err(|e| {
+			log_error!(logger, "Failed to setup VssStore: {}", e);
+			BuildError::KVStoreSetupFailed
+		})?)
+	}
+
+	/// Builds a [`FilesystemStore`] backend based implementation of [`KVStore`].
+	pub fn build_fs_store(&self) -> Result<FilesystemStore, BuildError> {
+		let mut storage_dir_path: PathBuf = self.config.storage_dir_path.clone().into();
+		storage_dir_path.push("fs_store");
+
+		fs::create_dir_all(storage_dir_path.clone())
+			.map_err(|_| BuildError::StoragePathAccessFailed)?;
+		Ok(FilesystemStore::new(storage_dir_path))
+	}
+
+	/// Builds a [`SqliteStore`] backend based implementation of [`KVStore`].
+	pub fn build_sqlite_store(&self) -> Result<SqliteStore, BuildError> {
+		let storage_dir_path = self.config.storage_dir_path.clone();
+		fs::create_dir_all(storage_dir_path.clone())
+			.map_err(|_| BuildError::StoragePathAccessFailed)?;
+
+		Ok(SqliteStore::new(
+			storage_dir_path.into(),
+			Some(io::sqlite_store::SQLITE_DB_FILE_NAME.to_string()),
+			Some(io::sqlite_store::KV_TABLE_NAME.to_string()),
+		)
+		.map_err(|_| BuildError::KVStoreSetupFailed)?)
+	}
+}
+
 /// A builder for an [`Node`] instance, allowing to set some configuration and module choices from
 /// the getgo.
 ///
diff --git a/src/io/vss_store.rs b/src/io/vss_store.rs
index 296eaabe3..21dfcd56e 100644
--- a/src/io/vss_store.rs
+++ b/src/io/vss_store.rs
@@ -8,7 +8,7 @@
 use crate::io::utils::check_namespace_key_validity;
 use bitcoin::hashes::{sha256, Hash, HashEngine, Hmac, HmacEngine};
 use lightning::io::{self, Error, ErrorKind};
-use lightning::util::persist::KVStore;
+use lightning::util::persist::{KVStore, MigratableKVStore};
 use prost::Message;
 use rand::RngCore;
 #[cfg(test)]
@@ -84,46 +84,39 @@ impl VssStore {
 		}
 	}
 
-	fn extract_key(&self, unified_key: &str) -> io::Result<String> {
+	fn key_to_key_parts<'a>(
+		&self, unified_key: &'a str,
+	) -> io::Result<(&'a str, &'a str, &'a str)> {
 		let mut parts = unified_key.splitn(3, '#');
-		let (_primary_namespace, _secondary_namespace) = (parts.next(), parts.next());
-		match parts.next() {
-			Some(obfuscated_key) => {
-				let actual_key = self.key_obfuscator.deobfuscate(obfuscated_key)?;
-				Ok(actual_key)
+		let (primary_namespace, secondary_namespace) = (parts.next(), parts.next());
+		match (primary_namespace, secondary_namespace, parts.next()) {
+			(Some(primary_namespace), Some(secondary_namespace), Some(obfuscated_key)) => {
+				Ok((primary_namespace, secondary_namespace, obfuscated_key))
 			},
-			None => Err(Error::new(ErrorKind::InvalidData, "Invalid key format")),
+			_ => Err(Error::new(ErrorKind::InvalidData, "Invalid key format")),
 		}
 	}
 
-	async fn list_all_keys(
-		&self, primary_namespace: &str, secondary_namespace: &str,
-	) -> io::Result<Vec<String>> {
+	async fn list_obfuscated_key_versions(&self, key_prefix: String) -> io::Result<Vec<KeyValue>> {
 		let mut page_token = None;
-		let mut keys = vec![];
-		let key_prefix = format!("{}#{}", primary_namespace, secondary_namespace);
+		let mut key_versions = vec![];
 		while page_token != Some("".to_string()) {
 			let request = ListKeyVersionsRequest {
 				store_id: self.store_id.clone(),
-				key_prefix: Some(key_prefix.clone()),
+				key_prefix: Some(key_prefix.to_string()),
 				page_token,
 				page_size: None,
 			};
 
 			let response = self.client.list_key_versions(&request).await.map_err(|e| {
-				let msg = format!(
-					"Failed to list keys in {}/{}: {}",
-					primary_namespace, secondary_namespace, e
-				);
+				let msg = format!("Failed to list keys with prefix {}: {}", key_prefix, e);
 				Error::new(ErrorKind::Other, msg)
 			})?;
 
-			for kv in response.key_versions {
-				keys.push(self.extract_key(&kv.key)?);
-			}
+			key_versions.extend(response.key_versions);
 			page_token = response.next_page_token;
 		}
-		Ok(keys)
+		Ok(key_versions)
 	}
 }
 
@@ -218,8 +211,9 @@ impl KVStore for VssStore {
 	fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result<Vec<String>> {
 		check_namespace_key_validity(primary_namespace, secondary_namespace, None, "list")?;
 
-		let keys = tokio::task::block_in_place(|| {
-			self.runtime.block_on(self.list_all_keys(primary_namespace, secondary_namespace))
+		let key_prefix = format!("{}#{}", primary_namespace, secondary_namespace);
+		let obfuscated_key_versions = tokio::task::block_in_place(|| {
+			self.runtime.block_on(self.list_obfuscated_key_versions(key_prefix))
 		})
 		.map_err(|e| {
 			let msg = format!(
@@ -229,7 +223,39 @@ impl KVStore for VssStore {
 			Error::new(ErrorKind::Other, msg)
 		})?;
 
-		Ok(keys)
+		let mut deobfuscated_keys = vec![];
+		for kv in obfuscated_key_versions {
+			let obfuscated_key = self.key_to_key_parts(&kv.key)?.2;
+			deobfuscated_keys.push(self.key_obfuscator.deobfuscate(obfuscated_key)?);
+		}
+		Ok(deobfuscated_keys)
+	}
+}
+
+impl MigratableKVStore for VssStore {
+	fn list_all_keys(&self) -> Result<Vec<(String, String, String)>, Error> {
+		// Use empty key_prefix to list all keys.
+		let empty_key_prefix = "".to_string();
+		let obfuscated_key_versions = tokio::task::block_in_place(|| {
+			self.runtime.block_on(self.list_obfuscated_key_versions(empty_key_prefix))
+		})
+		.map_err(|e| {
+			let msg = format!("Failed to list all keys: {}", e);
+			Error::new(ErrorKind::Other, msg)
+		})?;
+
+		let mut deobfuscated_keys = vec![];
+		for kv in obfuscated_key_versions {
+			let (primary_namespace, secondary_namespace, obfuscated_key) =
+				self.key_to_key_parts(&kv.key)?;
+
+			deobfuscated_keys.push((
+				primary_namespace.to_string(),
+				secondary_namespace.to_string(),
+				self.key_obfuscator.deobfuscate(&obfuscated_key)?,
+			));
+		}
+		Ok(deobfuscated_keys)
 	}
 }