feat: Introduce storage migrations via Importable/Exportable Storage …

…traits, and other trait-based operations.

rust/noosphere-storage/Cargo.toml

@@ -42,11 +42,9 @@ noosphere-core = { version = "0.15.2", path = "../noosphere-core", features = ["
 # examples/bench
 noosphere-common = { version = "0.1.0", path = "../noosphere-common", features = ["helpers"] }
 
-[target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]
-tempfile = { workspace = true }
-
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 sled = "~0.34"
+tempfile = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 rocksdb = { version = "0.21.0", optional = true }
 

rust/noosphere-storage/examples/bench/main.rs

@@ -128,17 +128,15 @@ impl BenchmarkStorage {
         ))]
         let (storage, storage_name) = {
             (
-                noosphere_storage::SledStorage::new(noosphere_storage::SledStorageInit::Path(
-                    storage_path.into(),
-                ))?,
+                noosphere_storage::SledStorage::new(&storage_path)?,
                 "SledDbStorage",
             )
         };
 
         #[cfg(all(not(target_arch = "wasm32"), feature = "rocksdb"))]
         let (storage, storage_name) = {
             (
-                noosphere_storage::RocksDbStorage::new(storage_path.into())?,
+                noosphere_storage::RocksDbStorage::new(&storage_path)?,
                 "RocksDbStorage",
             )
         };

rust/noosphere-storage/src/extra.rs

@@ -0,0 +1,94 @@
+use crate::storage::Storage;
+use anyhow::Result;
+use async_trait::async_trait;
+use noosphere_common::ConditionalSend;
+use std::{fs, path::Path};
+
+/// [Storage] that can be opened via [Path] reference.
+/// [FsBackedStorage] types get a blanket implementation.
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
+pub trait StorageOpen: Storage + Sized {
+    async fn open<P: AsRef<Path> + ConditionalSend>(path: P) -> Result<Self>;
+}
+
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
+impl<T> StorageOpen for T
+where
+    T: FsBackedStorage,
+{
+    async fn open<P: AsRef<Path> + ConditionalSend>(path: P) -> Result<Self> {
+        FsBackedStorage::open(path).await
+    }
+}
+
+/// [Storage] that can be deleted via [Path] reference.
+/// [FsBackedStorage] types get a blanket implementation.
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
+pub trait StorageRemove: Storage + Sized {
+    async fn remove<P: AsRef<Path> + ConditionalSend>(path: P) -> Result<()>;
+}
+
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
+impl<T> StorageRemove for T
+where
+    T: FsBackedStorage,
+{
+    async fn remove<P: AsRef<Path> + ConditionalSend>(path: P) -> Result<()> {
+        <T as FsBackedStorage>::remove(path).await
+    }
+}
+
+/// [Storage] that can be moved/renamed via [Path] reference.
+/// [FsBackedStorage] types get a blanket implementation.
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
+pub trait StorageRename: Storage + Sized {
+    async fn rename<P: AsRef<Path> + ConditionalSend, Q: AsRef<Path> + ConditionalSend>(
+        from: P,
+        to: Q,
+    ) -> Result<()>;
+}
+
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
+impl<T> StorageRename for T
+where
+    T: FsBackedStorage,
+{
+    async fn rename<P: AsRef<Path> + ConditionalSend, Q: AsRef<Path> + ConditionalSend>(
+        from: P,
+        to: Q,
+    ) -> Result<()> {
+        <T as FsBackedStorage>::rename(from, to).await
+    }
+}
+
+/// [Storage] that is based on a file system. Implementing [FsBackedStorage]
+/// provides blanket implementations for other trait-based [Storage] operations.
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
+pub trait FsBackedStorage: Storage + Sized {
+    /// Opens the storage at `path`.
+    async fn open<P: AsRef<Path> + ConditionalSend>(path: P) -> Result<Self>;
+
+    /// Deletes the storage located at `path` directory. Returns `Ok(())` if
+    /// the directory is successfully removed, or if it already does not exist.
+    async fn remove<P: AsRef<Path> + ConditionalSend>(path: P) -> Result<()> {
+        match fs::metadata(path.as_ref()) {
+            Ok(_) => fs::remove_dir_all(path.as_ref()).map_err(|e| e.into()),
+            Err(_) => Ok(()),
+        }
+    }
+
+    /// Moves the storage located at `from` to the `to` location.
+    async fn rename<P: AsRef<Path> + ConditionalSend, Q: AsRef<Path> + ConditionalSend>(
+        from: P,
+        to: Q,
+    ) -> Result<()> {
+        fs::rename(from, to).map_err(|e| e.into())
+    }
+}

rust/noosphere-storage/src/helpers.rs

@@ -2,7 +2,7 @@ use crate::Storage;
 use anyhow::Result;
 
 #[cfg(not(target_arch = "wasm32"))]
-use crate::{SledStorage, SledStorageInit, SledStore};
+use crate::{SledStorage, SledStore};
 
 #[cfg(not(target_arch = "wasm32"))]
 pub async fn make_disposable_store() -> Result<SledStore> {
@@ -13,7 +13,7 @@ pub async fn make_disposable_store() -> Result<SledStore> {
         .into_iter()
         .map(String::from)
         .collect();
-    let provider = SledStorage::new(SledStorageInit::Path(temp_dir.join(temp_name)))?;
+    let provider = SledStorage::new(temp_dir.join(temp_name))?;
     provider.get_block_store("foo").await
 }
 

rust/noosphere-storage/src/implementation/indexed_db.rs

@@ -1,8 +1,9 @@
 use crate::store::Store;
 use crate::{db::SPHERE_DB_STORE_NAMES, storage::Storage};
 use anyhow::{anyhow, Error, Result};
+use async_stream::try_stream;
 use async_trait::async_trait;
-use js_sys::Uint8Array;
+use js_sys::{ArrayBuffer, Uint8Array};
 use rexie::{
     KeyRange, ObjectStore, Rexie, RexieBuilder, Store as IdbStore, Transaction, TransactionMode,
 };
@@ -69,6 +70,11 @@ impl IndexedDbStorage {
         let db = Rc::into_inner(self.db)
             .ok_or_else(|| anyhow!("Could not unwrap inner during database clear."))?;
         db.close();
+        Self::delete(&name)
+    }
+
+    /// Deletes database with key `db_name` from origin storage.
+    pub async fn delete(db_name: &str) -> Result<()> {
         Rexie::delete(&name)
             .await
             .map_err(|error| anyhow!("{:?}", error))
@@ -90,7 +96,21 @@ impl Storage for IndexedDbStorage {
     }
 }
 
-#[derive(Clone)]
+#[async_trait(?Send)]
+impl crate::extra::StorageOpen for IndexedDbStorage {
+    async fn open<P: AsRef<Path> + ConditionalSend>(path: P) -> Result<Self> {
+        IndexedDbStorage::new(&key.as_ref().to_string()).await
+    }
+}
+
+#[async_trait(?Send)]
+impl crate::extra::StorageRemove for IndexedDbStorage {
+    async fn remove<P: AsRef<Path> + ConditionalSend>(path: P) -> Result<Self> {
+        Self::delete(path).await
+    }
+}
+
+#[derive(Clone, Clone)]
 pub struct IndexedDbStore {
     db: Rc<Rexie>,
     store_name: String,
@@ -114,87 +134,104 @@ impl IndexedDbStore {
         Ok(())
     }
 
-    fn bytes_to_typed_array(bytes: &[u8]) -> Result<JsValue> {
-        let array = Uint8Array::new_with_length(bytes.len() as u32);
-        array.copy_from(&bytes);
-        Ok(JsValue::from(array))
-    }
-
-    async fn contains(key: &JsValue, store: &IdbStore) -> Result<bool> {
+    async fn contains(key: &[u8], store: &IdbStore) -> Result<bool> {
+        let key_js = bytes_to_typed_array(key)?;
         let count = store
             .count(Some(
-                &KeyRange::only(key).map_err(|error| anyhow!("{:?}", error))?,
+                &KeyRange::only(&key_js).map_err(|error| anyhow!("{:?}", error))?,
             ))
             .await
             .map_err(|error| anyhow!("{:?}", error))?;
         Ok(count > 0)
     }
 
-    async fn read(key: &JsValue, store: &IdbStore) -> Result<Option<Vec<u8>>> {
+    async fn read(key: &[u8], store: &IdbStore) -> Result<Option<Vec<u8>>> {
+        let key_js = bytes_to_typed_array(key)?;
         Ok(match IndexedDbStore::contains(&key, &store).await? {
-            true => Some(
+            true => Some(typed_array_to_bytes(
                 store
-                    .get(&key)
+                    .get(&key_js)
                     .await
-                    .map_err(|error| anyhow!("{:?}", error))?
-                    .dyn_into::<Uint8Array>()
-                    .map_err(|error| anyhow!("{:?}", error))?
-                    .to_vec(),
-            ),
+                    .map_err(|error| anyhow!("{:?}", error))?,
+            )?),
             false => None,
         })
     }
+
+    async fn put(key: &[u8], value: &[u8], store: &IdbStore) -> Result<()> {
+        let key_js = bytes_to_typed_array(key)?;
+        let value_js = bytes_to_typed_array(value)?;
+        store
+            .put(&value_js, Some(&key_js))
+            .await
+            .map_err(|error| anyhow!("{:?}", error))?;
+        Ok(())
+    }
+
+    async fn delete(key: &[u8], store: &IdbStore) -> Result<()> {
+        let key_js = bytes_to_typed_array(key)?;
+        store
+            .delete(&key_js)
+            .await
+            .map_err(|error| anyhow!("{:?}", error))?;
+        Ok(())
+    }
 }
 
 #[async_trait(?Send)]
 impl Store for IndexedDbStore {
     async fn read(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
         let (store, tx) = self.start_transaction(TransactionMode::ReadOnly)?;
-        let key = IndexedDbStore::bytes_to_typed_array(key)?;
-
-        let maybe_dag = IndexedDbStore::read(&key, &store).await?;
-
+        let maybe_dag = IndexedDbStore::read(key, &store).await?;
         IndexedDbStore::finish_transaction(tx).await?;
-
         Ok(maybe_dag)
     }
 
     async fn write(&mut self, key: &[u8], bytes: &[u8]) -> Result<Option<Vec<u8>>> {
         let (store, tx) = self.start_transaction(TransactionMode::ReadWrite)?;
-
-        let key = IndexedDbStore::bytes_to_typed_array(key)?;
-        let value = IndexedDbStore::bytes_to_typed_array(bytes)?;
-
         let old_bytes = IndexedDbStore::read(&key, &store).await?;
-
-        store
-            .put(&value, Some(&key))
-            .await
-            .map_err(|error| anyhow!("{:?}", error))?;
-
+        IndexedDbStore::put(key, bytes, &store).await?;
         IndexedDbStore::finish_transaction(tx).await?;
-
         Ok(old_bytes)
     }
 
     async fn remove(&mut self, key: &[u8]) -> Result<Option<Vec<u8>>> {
         let (store, tx) = self.start_transaction(TransactionMode::ReadWrite)?;
-
-        let key = IndexedDbStore::bytes_to_typed_array(key)?;
-
-        let old_value = IndexedDbStore::read(&key, &store).await?;
-
-        store
-            .delete(&key)
-            .await
-            .map_err(|error| anyhow!("{:?}", error))?;
-
+        let old_value = IndexedDbStore::read(key, &store).await?;
+        IndexedDbStore::delete(key, &store).await?;
         IndexedDbStore::finish_transaction(tx).await?;
-
         Ok(old_value)
     }
 }
 
+impl crate::IterableStore for IndexedDbStore {
+    fn get_all_entries(&self) -> crate::IterableStoreStream<'_> {
+        Box::pin(try_stream! {
+            let (store, tx) = self.start_transaction(TransactionMode::ReadWrite)?;
+            let limit = 100;
+            let mut offset = 0;
+            loop {
+                let results = store.get_all(None, Some(limit), Some(offset), None).await
+                    .map_err(|error| anyhow!("{:?}", error))?;
+                let count = results.len();
+                if count == 0 {
+                    IndexedDbStore::finish_transaction(tx).await?;
+                    break;
+                }
+
+                offset += count as u32;
+
+                for (key_js, value_js) in results {
+                    yield (
+                        typed_array_to_bytes(JsValue::from(Uint8Array::new(&key_js)))?,
+                        Some(typed_array_to_bytes(value_js)?)
+                    );
+                }
+            }
+        })
+    }
+}
+
 #[cfg(feature = "performance")]
 struct SpaceUsageError(Error);
 
@@ -263,3 +300,16 @@ impl crate::Space for IndexedDbStorage {
         }
     }
 }
+
+fn bytes_to_typed_array(bytes: &[u8]) -> Result<JsValue> {
+    let array = Uint8Array::new_with_length(bytes.len() as u32);
+    array.copy_from(&bytes);
+    Ok(JsValue::from(array))
+}
+
+fn typed_array_to_bytes(js_value: JsValue) -> Result<Vec<u8>> {
+    Ok(js_value
+        .dyn_into::<Uint8Array>()
+        .map_err(|error| anyhow!("{:?}", error))?
+        .to_vec())
+}

rust/noosphere-storage/src/implementation/memory.rs

@@ -1,4 +1,5 @@
 use anyhow::{anyhow, Result};
+use async_stream::try_stream;
 use async_trait::async_trait;
 use cid::Cid;
 use std::{collections::HashMap, sync::Arc};
@@ -131,6 +132,19 @@ impl Store for MemoryStore {
     }
 }
 
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
+impl crate::IterableStore for MemoryStore {
+    fn get_all_entries(&self) -> crate::IterableStoreStream<'_> {
+        Box::pin(try_stream! {
+            let dags = self.entries.lock().await;
+            for key in dags.keys() {
+                yield (key.to_owned(), dags.get(key).cloned());
+            }
+        })
+    }
+}
+
 #[cfg(feature = "performance")]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]