Attempt 2: Move fastn_ds::wasm::Store to fastn_wasm::Store

Cargo.toml

@@ -71,7 +71,6 @@ dirs = "5"
 deadpool-postgres = "0.12"
 deadpool = "0.10"
 tokio-postgres = { version = "0.7", features = ["with-serde_json-1", "with-uuid-1"] }
-async-lock = "3"
 async-recursion = "1"
 async-trait = "0.1"
 camino = "1"
@@ -144,9 +143,6 @@ quick-js = "0.4"
 git = "https://github.com/fastn-stack/fastn-observer"
 rev = "5f64c7b"
 
-[workspace.dependencies.libsqlite3-sys]
-# this should be synced with whatever rusqlite uses
-version = "0.28.0"
 
 [workspace.dependencies.rusqlite]
 version = "0.31"

fastn-core/Cargo.toml

@@ -59,6 +59,7 @@ zip.workspace = true
 fastn-expr.workspace = true
 fastn-resolved.workspace = true
 fastn-runtime.workspace = true
+fastn-wasm.workspace = true
 
 [dev-dependencies]
 fbt-lib.workspace = true

fastn-core/src/config/config_temp.rs

@@ -84,7 +84,7 @@ impl ConfigTemp {
                 .to_package(package_root, package_name, ds, package, session_id)
                 .await?;
             ConfigTemp::check_dependencies_provided(package, &mut current_package)?;
-            fastn_ds::insert_or_update(&all_packages, package_name.clone(), current_package);
+            fastn_wasm::insert_or_update(&all_packages, package_name.clone(), current_package);
         }
 
         Ok(all_packages)

fastn-core/src/config/mod.rs

@@ -1085,7 +1085,7 @@ impl Config {
             config.package.endpoints
         };
 
-        fastn_ds::insert_or_update(
+        fastn_wasm::insert_or_update(
             &config.all_packages,
             package.name.to_string(),
             package.to_owned(),
@@ -1139,7 +1139,7 @@ impl Config {
 
     #[cfg(not(feature = "use-config-json"))]
     pub(crate) fn add_package(&self, package: &fastn_core::Package) {
-        fastn_ds::insert_or_update(
+        fastn_wasm::insert_or_update(
             &self.all_packages,
             package.name.to_string(),
             package.to_owned(),
@@ -1213,7 +1213,7 @@ async fn get_all_packages(
     session_id: &Option<String>,
 ) -> fastn_core::Result<scc::HashMap<String, fastn_core::Package>> {
     let all_packages = scc::HashMap::new();
-    fastn_ds::insert_or_update(&all_packages, package.name.to_string(), package.to_owned());
+    fastn_wasm::insert_or_update(&all_packages, package.name.to_string(), package.to_owned());
     let config_temp = config_temp::ConfigTemp::read(ds, session_id).await?;
     let other = config_temp
         .get_all_packages(ds, package, package_root, session_id)
@@ -1236,6 +1236,6 @@ async fn get_all_packages(
     _session_id: &Option<String>,
 ) -> fastn_core::Result<scc::HashMap<String, fastn_core::Package>> {
     let all_packages = scc::HashMap::new();
-    fastn_ds::insert_or_update(&all_packages, package.name.to_string(), package.to_owned());
+    fastn_wasm::insert_or_update(&all_packages, package.name.to_string(), package.to_owned());
     Ok(all_packages)
 }

fastn-core/src/library2022/mod.rs

@@ -236,7 +236,7 @@ impl Library2022 {
                 line_number: 0,
             })?;
 
-        fastn_ds::insert_or_update(&self.config.all_packages, package.name.clone(), package);
+        fastn_wasm::insert_or_update(&self.config.all_packages, package.name.clone(), package);
 
         Ok(())
     }

fastn-ds/Cargo.toml

@@ -15,16 +15,11 @@ tokio.workspace = true
 thiserror.workspace = true
 actix-web.workspace = true
 deadpool-postgres.workspace = true
-deadpool.workspace = true
 actix-http.workspace = true
-async-lock.workspace = true
 http.workspace = true
 bytes.workspace = true
 camino.workspace = true
-libsqlite3-sys.workspace = true
-futures-util.workspace = true
 async-trait.workspace = true
-tokio-postgres.workspace = true
 ignore.workspace = true
 dirs.workspace = true
 tracing.workspace = true
@@ -33,8 +28,6 @@ reqwest.workspace = true
 once_cell.workspace = true
 url.workspace = true
 scc.workspace = true
-serde.workspace = true
 serde_json.workspace = true
 wasmtime.workspace = true
 ft-sys-shared = { workspace = true, features = ["rusqlite"] }
-

fastn-ds/src/lib.rs

@@ -2,17 +2,13 @@
 #![deny(unused_crate_dependencies)]
 
 extern crate self as fastn_ds;
-
-mod create_pool;
 pub mod http;
 pub mod reqwest_util;
 mod user_data;
 mod utils;
 pub mod wasm;
 pub use user_data::UserDataError;
 
-pub use create_pool::create_pool;
-
 #[derive(Debug, Clone)]
 pub struct DocumentStore {
     pub wasm_modules: scc::HashMap<String, wasmtime::Module>,
@@ -201,9 +197,9 @@ impl DocumentStore {
             return Ok(p.get().clone());
         }
 
-        let pool = fastn_ds::create_pool(db_path.as_str()).await?;
+        let pool = fastn_wasm::create_pool(db_path.as_str()).await?;
 
-        fastn_ds::insert_or_update(&self.pg_pools, db_path.to_string(), pool.clone());
+        fastn_wasm::insert_or_update(&self.pg_pools, db_path.to_string(), pool.clone());
 
         Ok(pool)
     }
@@ -244,7 +240,11 @@ impl DocumentStore {
 
                 // we are only storing compiled module if we are not in debug mode
                 if !self.env_bool("FASTN_DEBUG", false).await? {
-                    fastn_ds::insert_or_update(&self.wasm_modules, path.to_string(), module.clone())
+                    fastn_wasm::insert_or_update(
+                        &self.wasm_modules,
+                        path.to_string(),
+                        module.clone(),
+                    )
                 }
 
                 Ok(module)

fastn-ds/src/wasm.rs

@@ -0,0 +1,89 @@
+pub struct Store;
+
+impl fastn_wasm::StoreExt for Store {
+    fn get_db_url(&self, store_db_url: &str, db_url: &str) -> String {
+        if db_url == "default" {
+            store_db_url
+        } else {
+            db_url
+        }
+        .to_string()
+    }
+
+    fn connection_open(
+        &self,
+        store_db_url: &str,
+        db_url: &str,
+    ) -> wasmtime::Result<Box<dyn fastn_wasm::ConnectionExt>> {
+        let conn = Connection(rusqlite::Connection::open(
+            self.get_db_url(store_db_url, db_url),
+        )?);
+
+        Ok(Box::new(conn))
+    }
+}
+
+pub struct Connection(rusqlite::Connection);
+
+impl fastn_wasm::ConnectionExt for Connection {
+    fn prepare(&self, sql: &str) -> Result<rusqlite::Statement, fastn_wasm::ExecuteError> {
+        self.0
+            .prepare(sql)
+            .map_err(fastn_wasm::ExecuteError::Rusqlite)
+    }
+
+    fn execute(
+        &self,
+        query: &str,
+        binds: Vec<fastn_wasm::Value>,
+    ) -> Result<usize, fastn_wasm::ExecuteError> {
+        self.0
+            .execute(query, rusqlite::params_from_iter(binds))
+            .map_err(fastn_wasm::ExecuteError::Rusqlite)
+    }
+
+    fn execute_batch(&self, query: &str) -> Result<(), fastn_wasm::ExecuteError> {
+        self.0
+            .execute_batch(query)
+            .map_err(fastn_wasm::ExecuteError::Rusqlite)
+    }
+}
+
+#[tracing::instrument(skip_all)]
+pub async fn process_http_request(
+    req: ft_sys_shared::Request,
+    module: wasmtime::Module,
+    wasm_pg_pools: actix_web::web::Data<scc::HashMap<String, deadpool_postgres::Pool>>,
+    db_url: String,
+) -> wasmtime::Result<ft_sys_shared::Request> {
+    let path = req.uri.clone();
+    let hostn_store = fastn_wasm::Store::new(req, wasm_pg_pools, db_url, Store);
+    let mut linker = wasmtime::Linker::new(module.engine());
+    hostn_store.register_functions(&mut linker);
+    let wasm_store = wasmtime::Store::new(module.engine(), hostn_store);
+    let (wasm_store, r) = fastn_utils::handle(wasm_store, module, linker, path).await?;
+
+    if let Some(r) = r {
+        return Ok(r);
+    }
+
+    Ok(wasm_store
+        .into_data()
+        .response
+        .ok_or(fastn_utils::WasmError::EndpointDidNotReturnResponse)?)
+}
+
+pub fn to_response(req: ft_sys_shared::Request) -> actix_web::HttpResponse {
+    println!("{req:?}");
+    let mut builder = actix_web::HttpResponse::build(req.method.parse().unwrap());
+    let mut resp = builder.status(req.method.parse().unwrap()).body(req.body);
+
+    for (k, v) in req.headers {
+        resp.headers_mut().insert(
+            k.parse().unwrap(),
+            actix_http::header::HeaderValue::from_bytes(v.as_slice()).unwrap(),
+        );
+    }
+
+    resp
+}