js effect for subscribing to config

fix errors

chore: Fix some things in the manager for clippy

libs/Cargo.lock

@@ -110,9 +110,9 @@ dependencies = [
 
 [[package]]
 name = "async-trait"
-version = "0.1.61"
+version = "0.1.64"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "705339e0e4a9690e2908d2b3d049d85682cf19fbd5782494498fbf7003a6a282"
+checksum = "1cd7fce9ba8c3c042128ce72d8b2ddbf3a05747efb67ea0313c635e10bda47a2"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1304,6 +1304,7 @@ dependencies = [
 name = "helpers"
 version = "0.1.0"
 dependencies = [
+ "async-trait",
  "color-eyre",
  "futures",
  "lazy_async_pool",
@@ -1649,6 +1650,7 @@ dependencies = [
  "helpers",
  "itertools 0.10.5",
  "models",
+ "pin-project",
  "reqwest",
  "serde",
  "serde_json",

libs/helpers/Cargo.toml

@@ -6,6 +6,7 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+async-trait = "0.1.64"
 color-eyre = "0.6.2"
 futures = "0.3.21"
 lazy_async_pool = "0.3.3"

libs/helpers/src/lib.rs

@@ -10,10 +10,12 @@ use tokio::sync::oneshot;
 use tokio::task::{JoinError, JoinHandle, LocalSet};
 
 mod byte_replacement_reader;
+mod os_api;
 mod rpc_client;
 mod rsync;
 mod script_dir;
 pub use byte_replacement_reader::*;
+pub use os_api::*;
 pub use rpc_client::{RpcClient, UnixRpcClient};
 pub use rsync::*;
 pub use script_dir::*;

libs/helpers/src/os_api.rs

@@ -0,0 +1,17 @@
+use models::Error;
+use models::PackageId;
+use serde_json::Value;
+
+pub struct RuntimeDropped;
+
+pub type Callback = Box<dyn Fn(Value) -> Result<(), RuntimeDropped> + Send + Sync + 'static>; // bool indicating if
+
+#[async_trait::async_trait]
+pub trait OsApi: Send + Sync + 'static {
+    async fn get_service_config(
+        &self,
+        id: PackageId,
+        path: &str,
+        callback: Callback,
+    ) -> Result<Value, Error>;
+}

libs/js_engine/Cargo.toml

@@ -43,3 +43,4 @@ serde = { version = "1.0", features = ["derive", "rc"] }
 serde_json = "1.0"
 tokio = { version = "1", features = ["full"] }
 tracing = "0.1"
+pin-project = "1"

libs/js_engine/src/artifacts/loadModule.js

@@ -193,6 +193,18 @@ const diskUsage = async ({
   return { used, total }
 }
 
+const callbackMapping = {}
+const registerCallback = (fn) => {
+  const uuid = generateUuid(); // TODO
+  callbackMapping[uuid] = fn;
+  return uuid
+}
+const runCallback = (uuid, data) => callbackMapping[uuid](data)
+
+const getServiceConfig = async (serviceId, configPath, onChange) => {
+  await Deno.core.opAsync("get_service_config", serviceId, configPath, registerCallback(onChange))
+}
+
 const currentFunction = Deno.core.opSync("current_function");
 const input = Deno.core.opSync("get_input");
 const variable_args = Deno.core.opSync("get_variable_args");
@@ -223,6 +235,7 @@ const effects = {
   runRsync,
   readDir,
   diskUsage,
+  getServiceConfig,
 };
 
 const defaults = {

libs/js_engine/src/lib.rs

@@ -1,7 +1,9 @@
 use std::collections::BTreeMap;
+use std::future::Future;
 use std::path::{Path, PathBuf};
 use std::pin::Pin;
 use std::sync::Arc;
+use std::task::Poll;
 use std::time::SystemTime;
 
 use deno_core::anyhow::{anyhow, bail};
@@ -11,12 +13,12 @@ use deno_core::{
     ModuleSpecifier, ModuleType, OpDecl, RuntimeOptions, Snapshot,
 };
 use embassy_container_init::ProcessGroupId;
-use helpers::{script_dir, spawn_local, Rsync, UnixRpcClient};
+use helpers::{script_dir, spawn_local, OsApi, Rsync, UnixRpcClient};
 use models::{PackageId, ProcedureName, Version, VolumeId};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use tokio::io::AsyncReadExt;
-use tokio::sync::Mutex;
+use tokio::sync::{mpsc, Mutex};
 
 pub trait PathForVolumeId: Send + Sync {
     fn path_for(
@@ -87,6 +89,7 @@ const SNAPSHOT_BYTES: &[u8] = include_bytes!("./artifacts/ARM_JS_SNAPSHOT.bin");
 #[derive(Clone)]
 struct JsContext {
     sandboxed: bool,
+    os: Arc<dyn OsApi>,
     datadir: PathBuf,
     run_function: String,
     version: Version,
@@ -97,6 +100,7 @@ struct JsContext {
     container_process_gid: ProcessGroupId,
     container_rpc_client: Option<Arc<UnixRpcClient>>,
     rsyncs: Arc<Mutex<(usize, BTreeMap<usize, Rsync>)>>,
+    callback_sender: mpsc::UnboundedSender<(String, Value)>,
 }
 #[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "kebab-case")]
@@ -178,6 +182,7 @@ impl ModuleLoader for ModsLoader {
 
 pub struct JsExecutionEnvironment {
     sandboxed: bool,
+    os: Arc<dyn OsApi>,
     base_directory: PathBuf,
     module_loader: ModsLoader,
     package_id: PackageId,
@@ -189,13 +194,14 @@ pub struct JsExecutionEnvironment {
 
 impl JsExecutionEnvironment {
     pub async fn load_from_package(
+        os: Arc<dyn OsApi>,
         data_directory: impl AsRef<std::path::Path>,
         package_id: &PackageId,
         version: &Version,
         volumes: Box<dyn PathForVolumeId>,
         container_process_gid: ProcessGroupId,
         container_rpc_client: Option<Arc<UnixRpcClient>>,
-    ) -> Result<JsExecutionEnvironment, (JsError, String)> {
+    ) -> Result<Self, (JsError, String)> {
         let data_dir = data_directory.as_ref();
         let base_directory = data_dir;
         let js_code = JsCode({
@@ -222,6 +228,7 @@ impl JsExecutionEnvironment {
             buffer
         });
         Ok(JsExecutionEnvironment {
+            os,
             base_directory: base_directory.to_owned(),
             module_loader: ModsLoader { code: js_code },
             package_id: package_id.clone(),
@@ -303,6 +310,7 @@ impl JsExecutionEnvironment {
             fns::rsync::decl(),
             fns::rsync_wait::decl(),
             fns::rsync_progress::decl(),
+            fns::get_service_config::decl(),
         ]
     }
 
@@ -315,7 +323,9 @@ impl JsExecutionEnvironment {
         let base_directory = self.base_directory.clone();
         let answer_state = AnswerState::default();
         let ext_answer_state = answer_state.clone();
+        let (callback_sender, callback_receiver) = mpsc::unbounded_channel();
         let js_ctx = JsContext {
+            os: self.os,
             datadir: base_directory,
             run_function: procedure_name
                 .js_function_name()
@@ -334,6 +344,7 @@ impl JsExecutionEnvironment {
             variable_args,
             container_process_gid: self.container_process_gid,
             container_rpc_client: self.container_rpc_client.clone(),
+            callback_sender,
             rsyncs: Default::default(),
         };
         let ext = Extension::builder()
@@ -352,16 +363,14 @@ impl JsExecutionEnvironment {
             startup_snapshot: Some(Snapshot::Static(SNAPSHOT_BYTES)),
             ..Default::default()
         };
-        let runtime = Arc::new(Mutex::new(JsRuntime::new(runtime_options)));
+        let mut runtime = JsRuntime::new(runtime_options);
 
         let future = async move {
             let mod_id = runtime
-                .lock()
-                .await
                 .load_main_module(&"file:///loadModule.js".parse().unwrap(), None)
                 .await?;
-            let evaluated = runtime.lock().await.mod_evaluate(mod_id);
-            let res = runtime.lock().await.run_event_loop(false).await;
+            let evaluated = runtime.mod_evaluate(mod_id);
+            let res = run_event_loop(&mut runtime, callback_receiver).await;
             res?;
             evaluated.await??;
             Ok::<_, AnyError>(())
@@ -377,6 +386,41 @@ impl JsExecutionEnvironment {
     }
 }
 
+#[pin_project::pin_project]
+struct RuntimeEventLoop<'a> {
+    runtime: &'a mut JsRuntime,
+    callback: mpsc::UnboundedReceiver<(String, Value)>,
+}
+impl<'a> Future for RuntimeEventLoop<'a> {
+    type Output = Result<(), AnyError>;
+    fn poll(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Self::Output> {
+        let this = self.project();
+        if let Poll::Ready(Some((uuid, value))) = this.callback.poll_recv(cx) {
+            match this
+                .runtime
+                .execute_script("callback", &format!("runCallback({uuid}, {value})"))
+            {
+                Ok(_) => (),
+                Err(e) => return Poll::Ready(Err(e)),
+            }
+        }
+        this.runtime.poll_event_loop(cx, false)
+    }
+}
+async fn run_event_loop(
+    runtime: &mut JsRuntime,
+    callback_receiver: mpsc::UnboundedReceiver<(String, Value)>,
+) -> Result<(), AnyError> {
+    RuntimeEventLoop {
+        runtime,
+        callback: callback_receiver,
+    }
+    .await
+}
+
 /// Note: Make sure that we have the assumption that all these methods are callable at any time, and all call restrictions should be in rust
 mod fns {
     use std::cell::RefCell;
@@ -396,9 +440,9 @@ mod fns {
         OutputParams, OutputStrategy, ProcessGroupId, ProcessId, RunCommand, RunCommandParams,
         SendSignal, SendSignalParams, SignalGroup, SignalGroupParams,
     };
-    use helpers::{to_tmp_path, AtomicFile, Rsync, RsyncOptions};
+    use helpers::{to_tmp_path, AtomicFile, Rsync, RsyncOptions, RuntimeDropped};
     use itertools::Itertools;
-    use models::{ErrorKind, VolumeId};
+    use models::{PackageId, VolumeId};
     use serde::{Deserialize, Serialize};
     use serde_json::{json, Value};
     use tokio::io::AsyncWriteExt;
@@ -1373,6 +1417,37 @@ mod fns {
         tokio::fs::set_permissions(new_file, Permissions::from_mode(mode)).await?;
         Ok(())
     }
+
+    #[op]
+    async fn get_service_config(
+        state: Rc<RefCell<OpState>>,
+        service_id: PackageId,
+        path: String,
+        callback: String,
+    ) -> Result<ResultType, AnyError> {
+        let state = state.borrow();
+        let ctx = state.borrow::<JsContext>();
+        let sender = ctx.callback_sender.clone();
+        Ok(
+            match ctx
+                .os
+                .get_service_config(
+                    service_id,
+                    &path,
+                    Box::new(move |value| {
+                        sender
+                            .send((callback.clone(), value))
+                            .map_err(|_| RuntimeDropped)
+                    }),
+                )
+                .await
+            {
+                Ok(a) => ResultType::Result(a),
+                Err(e) => ResultType::ErrorCode(e.kind as i32, e.source.to_string()),
+            },
+        )
+    }
+
     /// We need to make sure that during the file accessing, we don't reach beyond our scope of control
     async fn is_subset(
         parent: impl AsRef<Path>,