make blockstore put a command action

this way it can be called from a component via event messages

Author: DougAnderson444

Cargo.lock

@@ -3,32 +3,20 @@
 //! as Wurbo).
 //! That wat the WIT interface can simply import the same API and use it to communicate with the
 
-pub mod blockstore_idb;
-
-use futures::SinkExt;
 use futures::{channel::mpsc, StreamExt};
 use gloo_utils::format::JsValueSerdeExt;
-pub use peerpiper_core::events::PeerPiperCommand;
+pub use peerpiper_core::events::{PeerPiperCommand, SystemCommand};
+use peerpiper_core::Commander;
 use std::sync::Mutex;
 use std::sync::OnceLock;
 use wasm_bindgen::prelude::*;
 use wasm_bindgen_futures::spawn_local;
-use wnfs::common::CODEC_RAW;
-use wnfs::common::{BlockStore, Storable};
-use wnfs_unixfs_file::{
-    builder::{Config, FileBuilder},
-    chunker::{Chunker, ChunkerConfig},
-};
+
+use crate::blockstore::BrowserBlockStore;
 
 const MAX_CHANNELS: usize = 16;
 
-/// This wraps command_sender in a Mutex so we can call it from multiple wasm_bindgen functions
-/// without worrying about thread safety.
-/// Make the OnceLock inner value mutable so we can call get_mut() on it.
-/// This makes our wasm API clean enough to interface directly to WIT world.
-static COMMAND_SENDER: OnceLock<Mutex<mpsc::Sender<PeerPiperCommand>>> = OnceLock::new();
-/// BrowserBlockStore
-static BSTORE: OnceLock<Mutex<blockstore_idb::BrowserBlockStore>> = OnceLock::new();
+static COMMANDER: OnceLock<Mutex<Commander<BrowserBlockStore>>> = OnceLock::new();
 
 cfg_if::cfg_if! {
     if #[cfg(feature = "logging")] {
@@ -60,11 +48,13 @@ cfg_if::cfg_if! {
 pub async fn start() -> Result<(), JsValue> {
     init_log();
 
-    // Set up a blockstore in the browser
-    let blockstore = blockstore_idb::BrowserBlockStore::new("peerpiper");
-    blockstore.open().await?;
-
-    BSTORE.get_or_init(|| Mutex::new(blockstore));
+    let blockstore = BrowserBlockStore::new("peerpiper");
+    blockstore
+        .open()
+        .await
+        .map_err(|err| JsValue::from_str(&format!("Error opening blockstore: {:?}", err)))?;
+    let commander = Commander::new(blockstore);
+    COMMANDER.get_or_init(|| Mutex::new(commander));
 
     Ok(())
 }
@@ -76,8 +66,13 @@ pub async fn connect(libp2p_endpoint: &str, on_event: &js_sys::Function) -> Resu
     // command_sender will be used by other wasm_bindgen functions to send commands to the network
     // so we will need to wrap it in a Mutex or something to make it thread safe.
     let (command_sender, command_receiver) = mpsc::channel(8);
-    // move command_sender into COMMAND_SENDER
-    COMMAND_SENDER.get_or_init(|| Mutex::new(command_sender));
+    // move command_sender into COMMANDER
+    COMMANDER
+        .get()
+        .ok_or_else(|| JsError::new("Commander not initialized. Did `start()` complete?"))?
+        .lock()
+        .map_err(|err| JsError::new(&format!("Failed to lock commander: {}", err)))?
+        .with_network(command_sender);
 
     let endpoint = libp2p_endpoint.to_string().clone();
 
@@ -99,49 +94,11 @@ pub async fn connect(libp2p_endpoint: &str, on_event: &js_sys::Function) -> Resu
     Ok(())
 }
 
-/// Uses COMMAND_SENDER (if initialized) to send a command to the network.
-/// Else, returns an error.
-pub async fn send_command(command: PeerPiperCommand) -> Result<(), JsError> {
-    tracing::trace!("Sending command");
-    let command_sender = COMMAND_SENDER.get().ok_or_else(|| {
-        JsError::new(
-            "Command sender not initialized. Did you call `connect()` first to establish a connection?",
-        )
-    })?;
-
-    command_sender
-        .lock()
-        .map_err(|err| JsError::new(&format!("Failed to lock command sender: {}", err)))?
-        .send(command)
-        .await
-        .map_err(|err| JsError::new(&format!("Failed to send command: {}", err)))?;
-    Ok(())
-}
-
-/// Publish to this topic String these bytes
-#[wasm_bindgen]
-pub async fn publish(topic: String, data: Vec<u8>) -> Result<(), JsError> {
-    send_command(PeerPiperCommand::Publish { topic, data }).await
-}
-
-/// Subscribe to this topic String
-#[wasm_bindgen]
-pub async fn subscribe(topic: String) -> Result<(), JsError> {
-    send_command(PeerPiperCommand::Subscribe { topic }).await
-}
-
-/// Unsubscribe from this topic String
-/// This will stop receiving messages from this topic.
-#[wasm_bindgen]
-pub async fn unsubscribe(topic: String) -> Result<(), JsError> {
-    send_command(PeerPiperCommand::Unsubscribe { topic }).await
-}
-
-/// Takes any json string and tries to deserialize it into a PeerPiperCommand,
-/// then sends it to the network.
+/// Takes any json string from a Guest Component, and tries to deserialize it into a PeerPiperCommand,
+/// then sends it to the COMMANDER who routes it to either the network or the system depending on the command.
 /// If it fails, returns an error.
 #[wasm_bindgen]
-pub async fn command(json: &str) -> Result<(), JsError> {
+pub async fn command(json: &str) -> Result<JsValue, JsError> {
     let example_publish = PeerPiperCommand::Publish {
         topic: "example".to_string(),
         data: vec![1, 2, 3],
@@ -153,36 +110,21 @@ pub async fn command(json: &str) -> Result<(), JsError> {
             serde_json::to_string(&example_publish).unwrap()
         ))
     })?;
-    send_command(command).await
-}
-
-/// Allows the user to save a file to the system (IndexedDB. TODO: Memory too?)
-#[wasm_bindgen]
-pub async fn save(data: Vec<u8>) -> Result<(), JsError> {
-    tracing::info!("Saving to blockstore bytes {:?}", data.len());
 
-    let blockstore = BSTORE
+    let maybe_result = COMMANDER
         .get()
-        .ok_or_else(|| {
-            JsError::new(
-            "Blockstore not initialized. Did you call `start()` first to establish a connection?",
-        )
-        })?
+        .ok_or_else(|| JsError::new("Commander not initialized. Did `start()` complete?"))?
         .lock()
-        .unwrap();
-
-    // The chunker needs to be here because it is specific to IndexedDB having a max size of 256 *
-    // 1024 bytes. In another system (like a desktop disk) it could be chunked differently.
-    let root_cid = FileBuilder::new()
-        .content_bytes(data.clone())
-        .fixed_chunker(256 * 1024)
-        .build()
-        .map_err(|err| JsError::new(&format!("Failed to build file: {}", err)))?
-        .store(&blockstore.clone())
+        .map_err(|err| JsError::new(&format!("Failed to lock commander: {}", err)))?
+        .order(command)
         .await
-        .map_err(|err| JsError::new(&format!("Failed to store file: {}", err)))?;
-
-    tracing::info!("Saved file to blockstore with CID: {:?}", root_cid);
+        .map_err(|err| JsError::new(&format!("Failed to send command: {}", err)))?;
 
-    Ok(())
+    // convert the ReturnValues enum to a JsValue (Cid as String, Vec<u8> as Uint8Array, or null)
+    let js_val = match maybe_result {
+        peerpiper_core::ReturnValues::Data(data) => JsValue::from_serde(&data)?,
+        peerpiper_core::ReturnValues::ID(cid) => JsValue::from_str(&cid.to_string()),
+        peerpiper_core::ReturnValues::None => JsValue::null(),
+    };
+    Ok(js_val)
 }

crates/peerpiper-browser/src/bindgen.rs

@@ -139,38 +139,3 @@ impl BrowserBlockStore {
             .ok_or(JsValue::from_str("Error converting to bool"))?)
     }
 }
-
-impl WNFSBlockStore for BrowserBlockStore {
-    async fn get_block(&self, cid: &Cid) -> Result<Bytes, BlockStoreError> {
-        let key = CID::parse(&cid.to_string());
-        let js_uint8array = self
-            .get_idb(&key)
-            .await
-            .map_err(|_| BlockStoreError::CIDNotFound(*cid))?;
-        let bytes: Bytes = js_uint8array.to_vec().into();
-        Ok(bytes)
-    }
-
-    async fn put_block_keyed(
-        &self,
-        cid: Cid,
-        bytes: impl Into<Bytes> + CondSend,
-    ) -> Result<(), BlockStoreError> {
-        let key = CID::parse(&cid.to_string());
-
-        let bytes: Bytes = bytes.into();
-
-        let val = js_sys::Uint8Array::from(bytes.as_ref());
-        let _cid = self.put_idb(&key, val).await;
-
-        Ok(())
-    }
-
-    async fn has_block(&self, cid: &Cid) -> Result<bool, BlockStoreError> {
-        let key = CID::parse(&cid.to_string());
-        Ok(self
-            .has_in_idb(&key)
-            .await
-            .map_err(|_| BlockStoreError::CIDNotFound(*cid))?)
-    }
-}

crates/peerpiper-browser/src/bindgen/blockstore-idb.js renamed to crates/peerpiper-browser/src/blockstore/blockstore-idb.js

@@ -0,0 +1,175 @@
+/// WebNative File System Blockstore module
+mod wnfs_blockstore;
+
+use crate::error;
+use js_sys::Uint8Array;
+use parking_lot::Mutex;
+use peerpiper_core::SystemCommandHandler;
+use send_wrapper::SendWrapper;
+use tokio::io::AsyncReadExt as _;
+use wasm_bindgen::prelude::*;
+use wnfs::common::utils::Arc;
+
+use wnfs::common::libipld::Cid;
+use wnfs_unixfs_file::unixfs::UnixFsFile;
+
+#[wasm_bindgen(module = "/src/blockstore/blockstore-idb.js")]
+extern "C" {
+
+    /// CID Class
+    pub type CID;
+
+    /// CID Constructor
+    #[wasm_bindgen(constructor)]
+    pub fn new(version: u64, codec: u64, multihash: &Uint8Array, bytes: &Uint8Array) -> CID;
+
+    /// static parse(string)
+    /// Parse a CID from a string
+    #[wasm_bindgen(static_method_of = CID)]
+    pub fn parse(string: &str) -> CID;
+
+    /// CID method toString()
+    #[wasm_bindgen(method, js_name = toString)]
+    pub fn to_string(this: &CID) -> String;
+
+    #[derive(Debug)]
+    pub type IDBBlockstore;
+
+    #[wasm_bindgen(constructor)]
+    pub fn new(location: &str) -> IDBBlockstore;
+
+    /// Open method, open(): Promise<void>
+    #[wasm_bindgen(method)]
+    pub fn open(this: &IDBBlockstore) -> js_sys::Promise;
+
+    /// close(): Promise<void>
+    #[wasm_bindgen(method)]
+    pub fn close(this: &IDBBlockstore) -> js_sys::Promise;
+
+    /// delete(key): Promise<void>
+    #[wasm_bindgen(method)]
+    pub fn delete(this: &IDBBlockstore, key: &str) -> js_sys::Promise;
+
+    /// deleteMany(source, options?): AwaitIterable<CID<unknown, number, number, Version>>
+    #[wasm_bindgen(method)]
+    pub fn delete_many(this: &IDBBlockstore, source: &JsValue) -> js_sys::Promise;
+
+    /// destroy(): Promise<void>
+    #[wasm_bindgen(method)]
+    pub fn destroy(this: &IDBBlockstore) -> js_sys::Promise;
+
+    /// get(key): Promise<Uint8Array>
+    #[wasm_bindgen(method)]
+    pub fn get(this: &IDBBlockstore, key: &CID) -> js_sys::Promise;
+
+    /// getAll(options?): AwaitIterable<Pair>
+    #[wasm_bindgen(method)]
+    pub fn get_all(this: &IDBBlockstore) -> js_sys::Promise;
+
+    /// getMany(source, options?): AwaitIterable<Pair>
+    /// Retrieve values for the passed keys
+    ///
+    /// Parameters
+    /// source: AwaitIterable<CID<unknown, number, number, Version>>
+    /// Optional options: AbortOptions
+    /// Returns AwaitIterable<Pair>
+    #[wasm_bindgen(method)]
+    pub fn get_many(this: &IDBBlockstore, source: &JsValue) -> js_sys::Promise;
+
+    /// has(key): Promise<boolean>
+    #[wasm_bindgen(method)]
+    pub fn has(this: &IDBBlockstore, key: &CID) -> js_sys::Promise;
+
+    /// put(key, val: Uint8Array): Promise<CID<unknown, number, number, Version>>
+    #[wasm_bindgen(method)]
+    pub fn put(this: &IDBBlockstore, key: &CID, val: Uint8Array) -> js_sys::Promise;
+
+    /// putMany(source, options?): AwaitIterable<CID<unknown, number, number, Version>>
+    #[wasm_bindgen(method)]
+    pub fn put_many(this: &IDBBlockstore, source: &JsValue) -> js_sys::Promise;
+
+}
+
+impl SystemCommandHandler for BrowserBlockStore {
+    type Error = crate::error::Error;
+    async fn put(&self, data: Vec<u8>) -> Result<Cid, Self::Error> {
+        let root_cid = wnfs_blockstore::put_chunks(self.clone(), data)
+            .await
+            .map_err(|err| error::Error::Anyhow(err))?;
+        Ok(root_cid)
+    }
+
+    async fn get(&self, cid: String) -> Result<Vec<u8>, Self::Error> {
+        let cid = Cid::try_from(cid).map_err(|err| {
+            error::Error::Anyhow(anyhow::Error::msg(format!("Failed to parse CID: {}", err)))
+        })?;
+        let file = UnixFsFile::load(&cid, &self).await.map_err(|err| {
+            error::Error::Anyhow(anyhow::Error::msg(format!("Failed to load file: {}", err)))
+        })?;
+
+        let mut buffer = Vec::new();
+        let mut reader = file.into_content_reader(&self, None).map_err(|err| {
+            error::Error::Anyhow(anyhow::Error::msg(format!(
+                "Failed to create content reader: {}",
+                err
+            )))
+        })?;
+        reader.read_to_end(&mut buffer).await.map_err(|err| {
+            error::Error::Anyhow(anyhow::Error::msg(format!("Failed to read file: {}", err)))
+        })?;
+
+        Ok(buffer)
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct BrowserBlockStore {
+    pub(crate) idb: SendWrapper<Arc<Mutex<IDBBlockstore>>>,
+}
+
+impl BrowserBlockStore {
+    /// Creates a new in-memory block store.
+    pub fn new(namespace: &str) -> Self {
+        let blockstore = IDBBlockstore::new(namespace);
+        Self {
+            idb: SendWrapper::new(Arc::new(Mutex::new(blockstore))),
+        }
+    }
+
+    /// Opens the blockstore.
+    pub async fn open(&self) -> Result<(), JsValue> {
+        let promise = self.idb.lock().open();
+        let _ = wasm_bindgen_futures::JsFuture::from(promise)
+            .await
+            .map_err(|_| JsValue::from_str("Error opening blockstore"));
+        Ok(())
+    }
+
+    pub async fn get_idb(&self, cid: &CID) -> Result<Uint8Array, JsValue> {
+        let promise = self.idb.lock().get(cid);
+        let js_val = wasm_bindgen_futures::JsFuture::from(promise)
+            .await
+            .map_err(|_| JsValue::from_str("Error getting block"))?;
+        Ok(js_val.into())
+    }
+
+    /// Puts bytes into the blockstore.
+    pub async fn put_idb(&self, cid: &CID, bytes: Uint8Array) -> Result<CID, JsValue> {
+        let promise = self.idb.lock().put(cid, bytes);
+        let js_val = wasm_bindgen_futures::JsFuture::from(promise)
+            .await
+            .map_err(|_| JsValue::from_str("Error putting block"))?;
+        Ok(js_val.into())
+    }
+
+    /// Checks if the blockstore has a block with the given CID.
+    pub async fn has_in_idb(&self, cid: &CID) -> Result<bool, JsValue> {
+        let promise = self.idb.lock().has(cid);
+        let js_val = wasm_bindgen_futures::JsFuture::from(promise)
+            .await
+            .map_err(|_| JsValue::from_str("Error checking for block"))?;
+        Ok(js_val
+            .as_bool()
+            .ok_or(JsValue::from_str("Error converting to bool"))?)
+    }
+}