feat(native): provide FSWatcher

packages/native/src/block.rs

@@ -1,216 +0,0 @@
-use super::DynamicValue;
-use jwst::{Block as JwstBlock, Workspace};
-use lib0::any::Any;
-
-#[napi()]
-pub struct Block {
-    pub workspace: Workspace,
-    pub block: JwstBlock,
-}
-
-#[napi()]
-impl Block {
-    #[napi(constructor)]
-    pub fn new(workspace: Workspace, block: JwstBlock) -> Self {
-        Self { workspace, block }
-    }
-
-    #[napi]
-    pub fn get(&self, key: String) -> Option<DynamicValue> {
-        self.workspace
-            .with_trx(|trx| self.block.get(&trx.trx, &key).map(DynamicValue::new))
-    }
-
-    #[napi]
-    pub fn children(&self) -> Vec<String> {
-        self.workspace.with_trx(|trx| self.block.children(&trx.trx))
-    }
-
-    #[napi]
-    pub fn push_children(&self, block: &Block) {
-        self.workspace
-            .with_trx(|mut trx| self.block.push_children(&mut trx.trx, &block.block));
-    }
-
-    #[napi]
-    pub fn insert_children_at(&self, block: &Block, pos: u32) {
-        self.workspace.with_trx(|mut trx| {
-            self.block
-                .insert_children_at(&mut trx.trx, &block.block, pos)
-        });
-    }
-
-    #[napi]
-    pub fn insert_children_before(&self, block: &Block, reference: &str) {
-        self.workspace.with_trx(|mut trx| {
-            self.block
-                .insert_children_before(&mut trx.trx, &block.block, reference)
-        });
-    }
-
-    #[napi]
-    pub fn insert_children_after(&self, block: &Block, reference: &str) {
-        self.workspace.with_trx(|mut trx| {
-            self.block
-                .insert_children_after(&mut trx.trx, &block.block, reference)
-        });
-    }
-
-    #[napi]
-    pub fn remove_children(&self, block: &Block) {
-        self.workspace
-            .with_trx(|mut trx| self.block.remove_children(&mut trx.trx, &block.block));
-    }
-
-    #[napi]
-    pub fn exists_children(&self, block_id: &str) -> i32 {
-        self.workspace
-            .with_trx(|trx| self.block.exists_children(&trx.trx, block_id))
-            .map(|i| i as i32)
-            .unwrap_or(-1)
-    }
-
-    #[napi]
-    pub fn parent(&self) -> String {
-        self.workspace
-            .with_trx(|trx| self.block.parent(&trx.trx).unwrap())
-    }
-
-    #[napi]
-    pub fn updated(&self) -> u64 {
-        self.workspace.with_trx(|trx| self.block.updated(&trx.trx))
-    }
-
-    #[napi]
-    pub fn id(&self) -> String {
-        self.block.block_id()
-    }
-
-    #[napi]
-    pub fn flavor(&self) -> String {
-        self.workspace.with_trx(|trx| self.block.flavor(&trx.trx))
-    }
-
-    #[napi]
-    pub fn version(&self) -> String {
-        self.workspace.with_trx(|trx| {
-            let [major, minor] = self.block.version(&trx.trx);
-            format!("{major}.{minor}")
-        })
-    }
-
-    #[napi]
-    pub fn created(&self) -> u64 {
-        self.workspace.with_trx(|trx| self.block.created(&trx.trx))
-    }
-
-    #[napi]
-    pub fn set_bool(&self, key: String, value: bool) {
-        self.workspace
-            .with_trx(|mut trx| self.block.set(&mut trx.trx, &key, value));
-    }
-
-    #[napi]
-    pub fn set_string(&self, key: String, value: String) {
-        self.workspace
-            .with_trx(|mut trx| self.block.set(&mut trx.trx, &key, value));
-    }
-
-    #[napi]
-    pub fn set_float(&self, key: String, value: f64) {
-        self.workspace
-            .with_trx(|mut trx| self.block.set(&mut trx.trx, &key, value));
-    }
-
-    #[napi]
-    pub fn set_integer(&self, key: String, value: i64) {
-        self.workspace
-            .with_trx(|mut trx| self.block.set(&mut trx.trx, &key, value));
-    }
-
-    #[napi]
-    pub fn set_null(&self, key: String) {
-        self.workspace
-            .with_trx(|mut trx| self.block.set(&mut trx.trx, &key, Any::Null));
-    }
-
-    #[napi]
-    pub fn is_bool(&self, key: String) -> bool {
-        self.workspace.with_trx(|trx| {
-            self.block
-                .get(&trx.trx, &key)
-                .map(|a| matches!(a, Any::Bool(_)))
-                .unwrap_or(false)
-        })
-    }
-
-    #[napi]
-    pub fn is_string(&self, key: String) -> bool {
-        self.workspace.with_trx(|trx| {
-            self.block
-                .get(&trx.trx, &key)
-                .map(|a| matches!(a, Any::String(_)))
-                .unwrap_or(false)
-        })
-    }
-
-    #[napi]
-    pub fn is_float(&self, key: String) -> bool {
-        self.workspace.with_trx(|trx| {
-            self.block
-                .get(&trx.trx, &key)
-                .map(|a| matches!(a, Any::Number(_)))
-                .unwrap_or(false)
-        })
-    }
-
-    #[napi]
-    pub fn is_integer(&self, key: String) -> bool {
-        self.workspace.with_trx(|trx| {
-            self.block
-                .get(&trx.trx, &key)
-                .map(|a| matches!(a, Any::BigInt(_)))
-                .unwrap_or(false)
-        })
-    }
-
-    #[napi]
-    pub fn get_bool(&self, key: String) -> Option<i64> {
-        self.workspace.with_trx(|trx| {
-            self.block.get(&trx.trx, &key).and_then(|a| match a {
-                Any::Bool(i) => Some(i.into()),
-                _ => None,
-            })
-        })
-    }
-
-    #[napi]
-    pub fn get_string(&self, key: String) -> Option<String> {
-        self.workspace.with_trx(|trx| {
-            self.block.get(&trx.trx, &key).and_then(|a| match a {
-                Any::String(i) => Some(i.into()),
-                _ => None,
-            })
-        })
-    }
-
-    #[napi]
-    pub fn get_float(&self, key: String) -> Option<f64> {
-        self.workspace.with_trx(|trx| {
-            self.block.get(&trx.trx, &key).and_then(|a| match a {
-                Any::Number(i) => Some(i),
-                _ => None,
-            })
-        })
-    }
-
-    #[napi]
-    pub fn get_integer(&self, key: String) -> Option<i64> {
-        self.workspace.with_trx(|trx| {
-            self.block.get(&trx.trx, &key).and_then(|a| match a {
-                Any::BigInt(i) => Some(i),
-                _ => None,
-            })
-        })
-    }
-}

packages/native/src/dynamic_value.rs

@@ -1,68 +0,0 @@
-use lib0::any::Any;
-use std::collections::HashMap;
-
-pub type DynamicValueMap = HashMap<String, DynamicValue>;
-
-pub struct DynamicValue {
-    any: Any,
-}
-
-impl DynamicValue {
-    pub fn new(any: Any) -> Self {
-        Self { any }
-    }
-
-    pub fn as_bool(&self) -> Option<bool> {
-        match self.any {
-            Any::Bool(value) => Some(value),
-            _ => None,
-        }
-    }
-
-    pub fn as_number(&self) -> Option<f64> {
-        match self.any {
-            Any::Number(value) => Some(value),
-            _ => None,
-        }
-    }
-
-    pub fn as_int(&self) -> Option<i64> {
-        match self.any {
-            Any::BigInt(value) => Some(value),
-            _ => None,
-        }
-    }
-
-    pub fn as_string(&self) -> Option<String> {
-        match &self.any {
-            Any::String(value) => Some(value.to_string()),
-            _ => None,
-        }
-    }
-
-    pub fn as_buffer(&self) -> Option<Vec<u8>> {
-        match &self.any {
-            Any::Buffer(value) => Some(value.to_vec()),
-            _ => None,
-        }
-    }
-
-    pub fn as_array(&self) -> Option<Vec<DynamicValue>> {
-        match &self.any {
-            Any::Array(value) => Some(value.iter().map(|a| DynamicValue::new(a.clone())).collect()),
-            _ => None,
-        }
-    }
-
-    pub fn as_map(&self) -> Option<HashMap<String, DynamicValue>> {
-        match &self.any {
-            Any::Map(value) => Some(
-                value
-                    .iter()
-                    .map(|(key, value)| (key.clone(), DynamicValue::new(value.clone())))
-                    .collect(),
-            ),
-            _ => None,
-        }
-    }
-}

packages/native/src/fs.rs

@@ -0,0 +1,189 @@
+use std::{collections::HashMap, path::Path, sync::Arc};
+
+use napi::{
+  bindgen_prelude::{FromNapiValue, ToNapiValue},
+  threadsafe_function::{ErrorStrategy, ThreadsafeFunction, ThreadsafeFunctionCallMode},
+};
+use napi_derive::napi;
+use notify::{Event, RecommendedWatcher, RecursiveMode, Watcher};
+use parking_lot::Mutex;
+
+#[napi(object)]
+#[derive(Default)]
+pub struct WatchOptions {
+  pub recursive: Option<bool>,
+}
+
+#[napi(string_enum)]
+/// Watcher kind enumeration
+pub enum WatcherKind {
+  /// inotify backend (linux)
+  Inotify,
+  /// FS-Event backend (mac)
+  Fsevent,
+  /// KQueue backend (bsd,optionally mac)
+  Kqueue,
+  /// Polling based backend (fallback)
+  PollWatcher,
+  /// Windows backend
+  ReadDirectoryChangesWatcher,
+  /// Fake watcher for testing
+  NullWatcher,
+  Unknown,
+}
+
+impl From<notify::WatcherKind> for WatcherKind {
+  fn from(value: notify::WatcherKind) -> Self {
+    match value {
+      notify::WatcherKind::Inotify => WatcherKind::Inotify,
+      notify::WatcherKind::Fsevent => WatcherKind::Fsevent,
+      notify::WatcherKind::Kqueue => WatcherKind::Kqueue,
+      notify::WatcherKind::PollWatcher => WatcherKind::PollWatcher,
+      notify::WatcherKind::ReadDirectoryChangesWatcher => WatcherKind::ReadDirectoryChangesWatcher,
+      notify::WatcherKind::NullWatcher => WatcherKind::NullWatcher,
+      _ => WatcherKind::Unknown,
+    }
+  }
+}
+
+#[napi]
+pub struct Subscription {
+  id: uuid::Uuid,
+  error_uuid: Option<uuid::Uuid>,
+  event_emitter: Arc<Mutex<EventEmitter>>,
+}
+
+#[napi]
+impl Subscription {
+  #[napi]
+  #[allow(clippy::inherent_to_string)]
+  pub fn to_string(&self) -> String {
+    self.id.to_string()
+  }
+
+  #[napi]
+  pub fn unsubscribe(&mut self) {
+    let mut event_emitter = self.event_emitter.lock();
+    event_emitter.listeners.remove(&self.id);
+    if let Some(error_uuid) = &self.error_uuid {
+      event_emitter.error_callbacks.remove(error_uuid);
+    }
+  }
+}
+
+#[napi]
+pub fn watch(p: String, options: Option<WatchOptions>) -> Result<FSWatcher, anyhow::Error> {
+  let event_emitter = Arc::new(Mutex::new(EventEmitter {
+    listeners: Default::default(),
+    error_callbacks: Default::default(),
+  }));
+  let event_emitter_in_handler = event_emitter.clone();
+  let mut watcher: RecommendedWatcher =
+    notify::recommended_watcher(move |res: notify::Result<Event>| {
+      event_emitter_in_handler.lock().on(res);
+    })
+    .map_err(anyhow::Error::from)?;
+
+  let options = options.unwrap_or_default();
+  watcher
+    .watch(
+      Path::new(&p),
+      if options.recursive == Some(false) {
+        RecursiveMode::NonRecursive
+      } else {
+        RecursiveMode::Recursive
+      },
+    )
+    .map_err(anyhow::Error::from)?;
+  Ok(FSWatcher {
+    inner: watcher,
+    event_emitter,
+  })
+}
+
+#[napi]
+pub struct FSWatcher {
+  inner: RecommendedWatcher,
+  event_emitter: Arc<Mutex<EventEmitter>>,
+}
+
+#[napi]
+impl FSWatcher {
+  #[napi(getter)]
+  pub fn kind(&self) -> WatcherKind {
+    RecommendedWatcher::kind().into()
+  }
+
+  #[napi]
+  pub fn to_string(&self) -> napi::Result<String> {
+    Ok(format!("{:?}", self.inner))
+  }
+
+  #[napi]
+  pub fn subscribe(
+    &mut self,
+    callback: ThreadsafeFunction<serde_json::Value, ErrorStrategy::Fatal>,
+    error_callback: Option<ThreadsafeFunction<()>>,
+  ) -> Subscription {
+    let uuid = uuid::Uuid::new_v4();
+    let mut event_emitter = self.event_emitter.lock();
+    event_emitter.listeners.insert(uuid, callback);
+    let mut error_uuid = None;
+    if let Some(error_callback) = error_callback {
+      let uuid = uuid::Uuid::new_v4();
+      event_emitter.error_callbacks.insert(uuid, error_callback);
+      error_uuid = Some(uuid);
+    }
+    drop(event_emitter);
+    Subscription {
+      id: uuid,
+      error_uuid,
+      event_emitter: self.event_emitter.clone(),
+    }
+  }
+
+  #[napi]
+  pub fn close(&mut self) -> napi::Result<()> {
+    // drop the previous watcher
+    self.inner = notify::recommended_watcher(|_| {}).map_err(anyhow::Error::from)?;
+    self.event_emitter.lock().stop();
+    Ok(())
+  }
+}
+
+#[derive(Clone)]
+struct EventEmitter {
+  listeners: HashMap<uuid::Uuid, ThreadsafeFunction<serde_json::Value, ErrorStrategy::Fatal>>,
+  error_callbacks: HashMap<uuid::Uuid, ThreadsafeFunction<()>>,
+}
+
+impl EventEmitter {
+  fn on(&self, event: notify::Result<Event>) {
+    match event {
+      Ok(e) => match serde_json::value::to_value(e) {
+        Err(err) => {
+          let err: napi::Error = anyhow::Error::from(err).into();
+          for on_error in self.error_callbacks.values() {
+            on_error.call(Err(err.clone()), ThreadsafeFunctionCallMode::NonBlocking);
+          }
+        }
+        Ok(v) => {
+          for on_event in self.listeners.values() {
+            on_event.call(v.clone(), ThreadsafeFunctionCallMode::NonBlocking);
+          }
+        }
+      },
+      Err(err) => {
+        let err: napi::Error = anyhow::Error::from(err).into();
+        for on_error in self.error_callbacks.values() {
+          on_error.call(Err(err.clone()), ThreadsafeFunctionCallMode::NonBlocking);
+        }
+      }
+    }
+  }
+
+  fn stop(&mut self) {
+    self.listeners.clear();
+    self.error_callbacks.clear();
+  }
+}

packages/native/src/lib.rs

@@ -1,12 +1 @@
-// mod block;
-mod dynamic_value;
-mod storage;
-mod workspace;
-
-// pub use block::Block;
-pub use dynamic_value::{DynamicValue, DynamicValueMap};
-pub use storage::Storage;
-pub use workspace::Workspace;
-
-#[macro_use]
-extern crate napi_derive;
+pub mod fs;

packages/native/src/storage.rs

@@ -1,125 +0,0 @@
-use crate::Workspace;
-use jwst::{error, info, BlobStorage, DocStorage};
-use jwst_rpc::start_client;
-use jwst_storage::JwstStorage as AutoStorage;
-use std::sync::Arc;
-use tokio::{runtime::Runtime, sync::RwLock};
-use napi::bindgen_prelude::*;
-use napi::{Error, Result, Status};
-
-#[napi]
-#[derive(Clone)]
-pub struct Storage {
-  pub(crate) storage: Option<Arc<RwLock<AutoStorage>>>,
-  pub(crate) error: Option<String>,
-}
-
-#[napi]
-impl Storage {
-  #[napi(constructor)]
-  pub fn new(path: String) -> Self {
-    let rt = Runtime::new().unwrap();
-
-    // FIXME: do not use block_on
-    match rt.block_on(AutoStorage::new(&format!("sqlite:{path}?mode=rwc"))) {
-      Ok(pool) => Self {
-        storage: Some(Arc::new(RwLock::new(pool))),
-        error: None,
-      },
-      Err(e) => Self {
-        storage: None,
-        error: Some(e.to_string()),
-      },
-    }
-  }
-
-  #[napi]
-  pub fn error(&self) -> Option<String> {
-    self.error.clone()
-  }
-
-  #[napi]
-  pub async fn get_blob(&self, workspace_id: Option<String>, id: String) -> Result<Buffer> {
-    if let Some(storage) = &self.storage {
-      let storage_handle = storage.read().await;
-      let blobs = storage_handle.blobs();
-
-      let blob = blobs.get_blob(workspace_id.clone(), id.clone(), None).await.map_err(|e| {
-        Error::new(
-          Status::GenericFailure,
-          format!(
-            "Failed to get blob file {}/{} from storage, error: {}",
-            workspace_id.clone().unwrap_or_default().to_string(),
-            id,
-            e
-          ),
-        )
-      })?;
-
-      Ok(blob.into())
-    } else {
-      return Err(Error::new(
-        Status::GenericFailure,
-        "Storage is not connected",
-      ));
-    }
-  }
-
-  #[napi]
-  pub fn connect(&mut self, workspace_id: String, remote: String) -> Option<Workspace> {
-    match self.sync(workspace_id, remote) {
-      Ok(workspace) => Some(workspace),
-      Err(e) => {
-        error!("Failed to connect to workspace: {}", e);
-        self.error = Some(e.to_string());
-        None
-      }
-    }
-  }
-
-  #[napi]
-  pub fn sync(&self, workspace_id: String, remote: String) -> Result<Workspace> {
-    if let Some(storage) = &self.storage {
-      let rt = Runtime::new().unwrap();
-
-      // FIXME: do not use block_on
-      let mut workspace = rt
-        .block_on(async move {
-          let storage = storage.read().await;
-
-          start_client(&storage, workspace_id, remote).await
-        })
-        .map_err(|e| Error::new(Status::GenericFailure, e.to_string()))?;
-
-      let (sub, workspace) = {
-        let id = workspace.id();
-        let storage = self.storage.clone();
-        let sub = workspace.observe(move |_, e| {
-          let id = id.clone();
-          if let Some(storage) = storage.clone() {
-            let rt = Runtime::new().unwrap();
-            info!("update: {:?}", &e.update);
-            if let Err(e) = rt.block_on(async move {
-              let storage = storage.write().await;
-              storage.docs().write_update(id, &e.update).await
-            }) {
-              error!("Failed to write update to storage: {}", e);
-            }
-          }
-        });
-
-        (sub, workspace)
-      };
-
-      Ok(Workspace {
-        workspace,
-        _sub: sub,
-      })
-    } else {
-      Err(Error::new(
-        Status::GenericFailure,
-        "Storage is not connected",
-      ))
-    }
-  }
-}

packages/native/src/workspace.rs

@@ -1,84 +0,0 @@
-// use super::Block;
-use jwst::Workspace as JwstWorkspace;
-use yrs::UpdateSubscription;
-
-
-#[napi()]
-pub struct Workspace {
-    pub(crate) workspace: JwstWorkspace,
-    pub(crate) _sub: Option<UpdateSubscription>,
-}
-
-#[napi()]
-impl Workspace {
-    #[napi(constructor)]
-    pub fn new(id: String) -> Self {
-        Self {
-            workspace: JwstWorkspace::new(id),
-            _sub: None,
-        }
-    }
-
-    #[napi]
-    pub fn id(&self) -> String {
-        self.workspace.id()
-    }
-
-    #[napi]
-    pub fn client_id(&self) -> i64 {
-        self.workspace.client_id() as i64
-    }
-
-    // #[napi]
-    // pub fn get(&self, block_id: String) -> Option<Block> {
-    //     let workspace = self.workspace.clone();
-    //     self.workspace.with_trx(|mut trx| {
-    //         let block = trx
-    //             .get_blocks()
-    //             .get(&trx.trx, &block_id)
-    //             .map(|b| Block::new(workspace, b));
-    //         drop(trx);
-    //         block
-    //     })
-    // }
-
-    // #[napi]
-    // pub fn create(&self, block_id: String, flavor: String) -> Block {
-    //     let workspace = self.workspace.clone();
-    //     self.workspace.with_trx(|mut trx| {
-    //         let block = Block::new(
-    //             workspace,
-    //             trx.get_blocks().create(&mut trx.trx, block_id, flavor),
-    //         );
-    //         drop(trx);
-    //         block
-    //     })
-    // }
-
-    #[napi]
-    pub fn search(&self, query: String) -> String {
-        self.workspace.search_result(query)
-    }
-
-    // #[napi]
-    // pub fn get_blocks_by_flavour(&self, flavour: &str) -> Vec<Block> {
-    //     self.workspace
-    //         .with_trx(|mut trx| trx.get_blocks().get_blocks_by_flavour(&trx.trx, flavour))
-    //         .iter()
-    //         .map(|block| Block {
-    //             workspace: self.workspace.clone(),
-    //             block: block.clone(),
-    //         })
-    //         .collect()
-    // }
-
-    #[napi]
-    pub fn get_search_index(&self) -> Vec<String> {
-        self.workspace.metadata().search_index
-    }
-
-    #[napi]
-    pub fn set_search_index(&self, fields: Vec<String>) -> bool {
-        self.workspace.set_search_index(fields)
-    }
-}