Add support for sqlite backed durable objects.

Fixes #645

Author: connyay

worker-sandbox/src/lib.rs

@@ -26,6 +26,7 @@ mod test;
 mod user;
 mod utils;
 mod ws;
+mod sql_counter;
 
 #[derive(Deserialize, Serialize)]
 struct MyData {

worker-sandbox/src/router.rs

@@ -1,6 +1,6 @@
 use crate::{
     alarm, analytics_engine, assets, cache, d1, fetch, form, kv, queue, r2, request, service,
-    socket, user, ws, SomeSharedData, GLOBAL_STATE,
+    socket, user, ws, sql_counter, SomeSharedData, GLOBAL_STATE,
 };
 #[cfg(feature = "http")]
 use std::convert::TryInto;
@@ -224,6 +224,7 @@ pub fn make_router(data: SomeSharedData, env: Env) -> axum::Router {
             "/analytics-engine",
             get(handler!(analytics_engine::handle_analytics_event)),
         )
+        .route("/sql-counter/:name", get(handler!(sql_counter::handle_sql_counter)))
         .fallback(get(handler!(catchall)))
         .layer(Extension(env))
         .layer(Extension(data))
@@ -364,6 +365,7 @@ pub fn make_router<'a>(data: SomeSharedData) -> Router<'a, SomeSharedData> {
         .delete_async("/r2/delete", handler!(r2::delete))
         .get_async("/socket/failed", handler!(socket::handle_socket_failed))
         .get_async("/socket/read", handler!(socket::handle_socket_read))
+        .get_async("/sql-counter/:name", handler!(sql_counter::handle_sql_counter))
         .or_else_any_method_async("/*catchall", handler!(catchall))
 }
 

worker-sandbox/src/sql_counter.rs

@@ -0,0 +1,59 @@
+use worker::*;
+use wasm_bindgen::JsValue;
+
+/// A simple SQLite-backed counter stored in Durable Object storage.
+///
+/// Each Durable Object instance owns its own private SQLite database.  We keep a
+/// single table `counter` with one row that stores the current value.
+#[durable_object]
+pub struct SqlCounter {
+    sql: SqlStorage,
+}
+
+#[durable_object]
+impl DurableObject for SqlCounter {
+    fn new(state: State, _env: Env) -> Self {
+        let sql = state.storage().sql();
+        // Create table if it does not exist.  Note: `exec` is synchronous.
+        sql.exec("CREATE TABLE IF NOT EXISTS counter(value INTEGER);", Vec::new())
+            .expect("create table");
+        Self { sql }
+    }
+
+    async fn fetch(&mut self, _req: Request) -> Result<Response> {
+        // Read current value (if any)
+        #[derive(serde::Deserialize)]
+        struct Row {
+            value: i64,
+        }
+
+        let rows: Vec<Row> = self
+            .sql
+            .exec("SELECT value FROM counter LIMIT 1;", Vec::new())?
+            .to_array()?;
+        let current = rows.get(0).map(|r| r.value).unwrap_or(0);
+        let next = current + 1;
+
+        // Upsert new value – simplest way: delete and insert again.
+        self.sql.exec("DELETE FROM counter;", Vec::new())?;
+        let val_js = JsValue::from_f64(next as f64);
+        self.sql
+            .exec("INSERT INTO counter(value) VALUES (?);", vec![val_js])?;
+
+        Response::ok(format!("SQL counter is now {}", next))
+    }
+}
+
+#[worker::send]
+/// Route handler that proxies a request to our SqlCounter Durable Object with id derived from the
+/// path `/sql-counter/{name}` (so every name gets its own instance).
+pub async fn handle_sql_counter(req: Request, env: Env, _data: super::SomeSharedData) -> Result<Response> {
+    let uri = req.url()?;
+    let mut segments = uri.path_segments().unwrap();
+    // skip "sql-counter"
+    let _ = segments.next();
+    let name = segments.next().unwrap_or("default");
+    let namespace = env.durable_object("SQL_COUNTER")?;
+    let stub = namespace.id_from_name(name)?.get_stub()?;
+    stub.fetch_with_str("https://fake-host/").await
+}

worker-sandbox/tests/mf.ts

@@ -59,6 +59,7 @@ export const mf = new Miniflare({
       durableObjects: {
         COUNTER: "Counter",
         PUT_RAW_TEST_OBJECT: "PutRawTestObject",
+        SQL_COUNTER: "SqlCounter",
       },
       kvNamespaces: ["SOME_NAMESPACE", "FILE_SIZES", "TEST"],
       serviceBindings: {

worker-sandbox/tests/sqlite.spec.ts

@@ -0,0 +1,22 @@
+import { describe, test, expect } from "vitest";
+import { mf } from "./mf";
+
+// Unsupported by miniflare
+describe.skip("sqlite durable object", () => {
+  test("counter increments per object", async () => {
+    // First access for object "alice"
+    let resp = await mf.dispatchFetch("http://fake.host/sql-counter/alice");
+    expect(resp.status).toBe(200);
+    expect(await resp.text()).toBe("SQL counter is now 1");
+
+    // Second access for same object should increment
+    resp = await mf.dispatchFetch("http://fake.host/sql-counter/alice");
+    expect(resp.status).toBe(200);
+    expect(await resp.text()).toBe("SQL counter is now 2");
+
+    // Different object name should have its own counter starting at 1
+    resp = await mf.dispatchFetch("http://fake.host/sql-counter/bob");
+    expect(resp.status).toBe(200);
+    expect(await resp.text()).toBe("SQL counter is now 1");
+  });
+});

worker-sandbox/wrangler.toml

@@ -29,6 +29,7 @@ bindings = [
     { name = "COUNTER", class_name = "Counter" },
     { name = "ALARM", class_name = "AlarmObject" },
     { name = "PUT_RAW_TEST_OBJECT", class_name = "PutRawTestObject" },
+    { name = "SQL_COUNTER", class_name = "SqlCounter" },
 ]
 
 [[analytics_engine_datasets]]
@@ -82,3 +83,7 @@ type = "CompiledWasm"
 
 [package.metadata.wasm-pack.profile.dev.wasm-bindgen]
 dwarf-debug-info = true
+
+[[migrations]]
+tag = "v1"
+new_sqlite_classes = ["SqlCounter"]

worker-sys/src/types/durable_object.rs

@@ -5,12 +5,14 @@ mod namespace;
 mod state;
 mod storage;
 mod transaction;
+mod sql_storage;
 
 pub use id::*;
 pub use namespace::*;
 pub use state::*;
 pub use storage::*;
 pub use transaction::*;
+pub use sql_storage::*;
 
 #[wasm_bindgen]
 extern "C" {

worker-sys/src/types/durable_object/sql_storage.rs

@@ -0,0 +1,61 @@
+use js_sys::{Array, Iterator as JsIterator, Object};
+use wasm_bindgen::prelude::*;
+
+#[wasm_bindgen]
+extern "C" {
+    #[wasm_bindgen(extends = js_sys::Object)]
+    #[derive(Clone)]
+    pub type SqlStorage;
+
+    /// Returns the on-disk size of the SQLite database, in bytes.
+    #[wasm_bindgen(method, getter, js_name = databaseSize)]
+    pub fn database_size(this: &SqlStorage) -> f64;
+
+    /// Execute a SQL statement (optionally with bindings) and return a cursor over the results.
+    ///
+    /// The JavaScript definition of `exec` is variadic, taking a SQL string followed by an
+    /// arbitrary number of binding parameters.  In Rust we accept the bindings packed into an
+    /// `Array` – callers can construct the array (or use the helper provided in the higher-level
+    /// `worker` crate).
+    #[wasm_bindgen(structural, method, catch, variadic, js_class = SqlStorage, js_name = exec)]
+    pub fn exec(
+        this: &SqlStorage,
+        query: &str,
+        bindings: Array,
+    ) -> Result<SqlStorageCursor, JsValue>;
+}
+
+#[wasm_bindgen]
+extern "C" {
+    #[wasm_bindgen(extends = js_sys::Object)]
+    #[derive(Clone)]
+    pub type SqlStorageCursor;
+
+    /// JavaScript `Iterator.next()` implementation.  Returns an object with `{ done, value }`.
+    #[wasm_bindgen(method, js_name = next)]
+    pub fn next(this: &SqlStorageCursor) -> Object;
+
+    /// Convert the remaining rows into an array of objects.
+    #[wasm_bindgen(method, js_name = toArray)]
+    pub fn to_array(this: &SqlStorageCursor) -> Array;
+
+    /// Returns the single row if exactly one row exists, otherwise throws.
+    #[wasm_bindgen(method)]
+    pub fn one(this: &SqlStorageCursor) -> JsValue;
+
+    /// Returns an iterator where each row is an array rather than an object.
+    #[wasm_bindgen(method)]
+    pub fn raw(this: &SqlStorageCursor) -> JsIterator;
+
+    /// Column names in the order they appear in `raw()` row arrays.
+    #[wasm_bindgen(method, getter, js_name = columnNames)]
+    pub fn column_names(this: &SqlStorageCursor) -> Array;
+
+    /// Rows read so far.
+    #[wasm_bindgen(method, getter, js_name = rowsRead)]
+    pub fn rows_read(this: &SqlStorageCursor) -> f64;
+
+    /// Rows written so far.
+    #[wasm_bindgen(method, getter, js_name = rowsWritten)]
+    pub fn rows_written(this: &SqlStorageCursor) -> f64;
+}

worker-sys/src/types/durable_object/storage.rs

@@ -74,4 +74,7 @@ extern "C" {
         this: &DurableObjectStorage,
         options: js_sys::Object,
     ) -> Result<js_sys::Promise, JsValue>;
+
+    #[wasm_bindgen(method, getter)]
+    pub fn sql(this: &DurableObjectStorage) -> crate::types::SqlStorage;
 }

worker/src/durable.rs

@@ -501,6 +501,11 @@ impl Storage {
             .map_err(Error::from)
             .map(|_| ())
     }
+
+    // Add new method to access SQLite APIs
+    pub fn sql(&self) -> crate::sql::SqlStorage {
+        crate::sql::SqlStorage::new(self.inner.sql())
+    }
 }
 
 pub struct Transaction {

worker/src/lib.rs

@@ -234,6 +234,7 @@ mod socket;
 mod streams;
 mod version;
 mod websocket;
+mod sql;
 
 pub type Result<T> = StdResult<T, error::Error>;
 
@@ -252,3 +253,5 @@ pub type HttpRequest = ::http::Request<http::body::Body>;
 #[cfg(feature = "http")]
 /// **Requires** `http` feature. Type alias for `http::Response<worker::Body>`.
 pub type HttpResponse = ::http::Response<http::body::Body>;
+
+pub use crate::sql::*;

worker/src/sql.rs

@@ -0,0 +1,102 @@
+use js_sys::{Array};
+use wasm_bindgen::JsValue;
+use worker_sys::types::{SqlStorage as SqlStorageSys, SqlStorageCursor as SqlStorageCursorSys};
+
+use serde::de::DeserializeOwned;
+use serde_wasm_bindgen as swb;
+
+use crate::Error;
+use crate::Result;
+
+/// Wrapper around the Workers `SqlStorage` interface exposed at `ctx.storage.sql`.
+///
+/// The API is intentionally minimal for now – additional helper utilities can be built on top
+/// as needed.
+#[derive(Clone)]
+pub struct SqlStorage {
+    inner: SqlStorageSys,
+}
+
+unsafe impl Send for SqlStorage {}
+unsafe impl Sync for SqlStorage {}
+
+impl SqlStorage {
+    pub(crate) fn new(inner: SqlStorageSys) -> Self {
+        Self { inner }
+    }
+
+    /// Size of the underlying SQLite database in bytes.
+    pub fn database_size(&self) -> usize {
+        self.inner.database_size() as usize
+    }
+
+    /// Execute a SQL query and return a cursor over the results.
+    ///
+    /// `bindings` correspond to positional `?` placeholders in the query.
+    pub fn exec(&self, query: &str, bindings: Vec<JsValue>) -> Result<SqlCursor> {
+        let array = Array::new();
+        for v in bindings {
+            array.push(&v);
+        }
+        let cursor = self.inner.exec(query, array).map_err(Error::from)?;
+        Ok(SqlCursor { inner: cursor })
+    }
+}
+
+impl AsRef<JsValue> for SqlStorage {
+    fn as_ref(&self) -> &JsValue {
+        &self.inner
+    }
+}
+
+/// A cursor returned from `SqlStorage::exec`.
+#[derive(Clone)]
+pub struct SqlCursor {
+    inner: SqlStorageCursorSys,
+}
+
+unsafe impl Send for SqlCursor {}
+unsafe impl Sync for SqlCursor {}
+
+impl SqlCursor {
+    /// Consume the remaining rows of the cursor into a `Vec` of deserialised objects.
+    pub fn to_array<T>(&self) -> Result<Vec<T>>
+    where
+        T: DeserializeOwned,
+    {
+        let arr = self.inner.to_array();
+        let mut out = Vec::with_capacity(arr.length() as usize);
+        for val in arr.iter() {
+            out.push(swb::from_value(val)?);
+        }
+        Ok(out)
+    }
+
+    /// Return the first (and only) row of the query result.
+    pub fn one<T>(&self) -> Result<T>
+    where
+        T: DeserializeOwned,
+    {
+        let val = self.inner.one();
+        Ok(swb::from_value(val)?)
+    }
+
+    /// Column names returned by the query.
+    pub fn column_names(&self) -> Vec<String> {
+        self.inner
+            .column_names()
+            .iter()
+            .map(|v| v.as_string().unwrap_or_default())
+            .collect()
+    }
+
+    /// Number of rows read so far by the cursor.
+    pub fn rows_read(&self) -> usize {
+        self.inner.rows_read() as usize
+    }
+
+    /// Number of rows written by the query so far.
+    pub fn rows_written(&self) -> usize {
+        self.inner.rows_written() as usize
+    }
+}