diff --git a/Cargo.lock b/Cargo.lock index 180ee8fd8..acd32c777 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2611,6 +2611,15 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "sync-kv" +version = "0.1.0" +dependencies = [ + "serde_json", + "worker", + "worker-macros", +] + [[package]] name = "sync_wrapper" version = "1.0.2" diff --git a/examples/sync-kv/Cargo.toml b/examples/sync-kv/Cargo.toml new file mode 100644 index 000000000..0f808e4c7 --- /dev/null +++ b/examples/sync-kv/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "sync-kv" +version = "0.1.0" +edition = "2021" + +[lib] +crate-type = ["cdylib"] + +[dependencies] +serde_json.workspace = true +worker.workspace = true +worker-macros.workspace = true diff --git a/examples/sync-kv/src/lib.rs b/examples/sync-kv/src/lib.rs new file mode 100644 index 000000000..9cb4c48fa --- /dev/null +++ b/examples/sync-kv/src/lib.rs @@ -0,0 +1,59 @@ +use serde_json::{json, Value}; +use worker::*; + +#[durable_object] +pub struct SyncKvDurableObject { + state: State, +} + +impl DurableObject for SyncKvDurableObject { + fn new(state: State, _env: Env) -> Self { + Self { state } + } + + async fn fetch(&self, _req: Request) -> Result { + let kv = self.state.storage().kv(); + + // CLEAN + kv.delete("a")?; + kv.delete("b")?; + + // CREATE + kv.put("a", json!({ "x": 1 }))?; + kv.put("b", json!({ "x": 2 }))?; + + // READ + let a: Option = kv.get("a")?; + let b: Option = kv.get("b")?; + let missing: Option = kv.get("c")?; + + // UPDATE + kv.put("a", json!({ "x": 42 }))?; + let a_updated: Option = kv.get("a")?; + + // DELETE + let deleted = kv.delete("b")?; + let after_delete: Option = kv.get("b")?; + + // LIST + let mut list = Vec::new(); + for item in kv.list::()? { + let (k, v) = item?; + list.push((k, v)); + } + + Response::from_json(&json!({ + "read": { "a": a, "b": b, "missing": missing }, + "update": a_updated, + "delete": { "deleted": deleted, "after_delete": after_delete }, + "list": list + })) + } +} + +#[event(fetch)] +async fn fetch(req: Request, env: Env, _ctx: Context) -> Result { + let durable_obj = env.durable_object("SYNC_KV")?; + let stub = durable_obj.id_from_name("singleton")?.get_stub()?; + stub.fetch_with_request(req).await +} diff --git a/examples/sync-kv/wrangler.toml b/examples/sync-kv/wrangler.toml new file mode 100644 index 000000000..6ffd23d27 --- /dev/null +++ b/examples/sync-kv/wrangler.toml @@ -0,0 +1,14 @@ +name = "sync-kv-example" +main = "build/index.js" +compatibility_date = "2025-09-23" + +[build] +command = "cargo install \"worker-build@^0.8\" && worker-build --release" + +[[durable_objects.bindings]] +name = "SYNC_KV" +class_name = "SyncKvDurableObject" + +[[migrations]] +tag = "v1" +new_sqlite_classes = ["SyncKvDurableObject"] diff --git a/test/src/lib.rs b/test/src/lib.rs index a5c401e8d..bc084bcfa 100644 --- a/test/src/lib.rs +++ b/test/src/lib.rs @@ -37,6 +37,7 @@ mod signal; mod socket; mod sql_counter; mod sql_iterator; +mod synchronous_storage; mod user; mod ws; diff --git a/test/src/router.rs b/test/src/router.rs index 147c5cd4b..c52fae585 100644 --- a/test/src/router.rs +++ b/test/src/router.rs @@ -2,7 +2,8 @@ use crate::signal; use crate::{ alarm, analytics_engine, assets, auto_response, cache, container, counter, d1, durable, fetch, form, js_snippets, kv, put_raw, queue, r2, rate_limit, request, secret_store, service, socket, - sql_counter, sql_iterator, user, ws, SomeSharedData, GLOBAL_SECOND_START, GLOBAL_STATE, + sql_counter, sql_iterator, synchronous_storage, user, ws, SomeSharedData, GLOBAL_SECOND_START, + GLOBAL_STATE, }; #[cfg(feature = "http")] use std::convert::TryInto; @@ -240,6 +241,14 @@ macro_rules! add_routes ( add_route!($obj, get, format_route!("/rate-limit/key/{}", "key"), rate_limit::handle_rate_limit_with_key); add_route!($obj, get, "/rate-limit/bulk-test", rate_limit::handle_rate_limit_bulk_test); add_route!($obj, get, "/rate-limit/reset", rate_limit::handle_rate_limit_reset); + add_route!($obj, get, "/synchronous-storage/smoke", synchronous_storage::handle_synchronous_storage_smoke); + add_route!($obj, get, "/synchronous-storage/overwrite", synchronous_storage::handle_synchronous_storage_overwrite); + add_route!($obj, get, "/synchronous-storage/not_found", synchronous_storage::handle_synchronous_storage_not_found); + add_route!($obj, get, "/synchronous-storage/list", synchronous_storage::handle_synchronous_storage_list); + add_route!($obj, get, "/synchronous-storage/list_options", synchronous_storage::handle_synchronous_storage_list_options); + add_route!($obj, get, "/synchronous-storage/persist_fill", synchronous_storage::handle_synchronous_storage_persist_fill); + add_route!($obj, get, "/synchronous-storage/persist_check", synchronous_storage::handle_synchronous_storage_persist_check); + add_route!($obj, get, "/synchronous-storage/persist_cleanup", synchronous_storage::handle_synchronous_storage_persist_cleanup); add_route!($obj, get, "/signal/poll", signal::handle_signal_poll); }); diff --git a/test/src/synchronous_storage.rs b/test/src/synchronous_storage.rs new file mode 100644 index 000000000..523fcf29e --- /dev/null +++ b/test/src/synchronous_storage.rs @@ -0,0 +1,270 @@ +use worker::*; + +#[durable_object] +pub struct SynchronousStorage { + state: State, +} + +impl DurableObject for SynchronousStorage { + fn new(state: State, _env: Env) -> Self { + Self { state } + } + + async fn fetch(&self, req: Request) -> Result { + let sync_kv = self.state.storage().kv(); + let path = req.path(); + + match path.as_str() { + "/smoke" => { + let first = serde_json::json!({"x": 1}); + let second = serde_json::json!({"x": 2}); + + sync_kv.put("first", first.clone())?; + sync_kv.put("second", second.clone())?; + + assert_eq!(sync_kv.get("first")?, Some(first.clone())); + assert_eq!(sync_kv.get("second")?, Some(second.clone())); + + let mut original = [ + (String::from("first"), first), + (String::from("second"), second), + ]; + let mut list: Box<[_]> = sync_kv.list()?.map(Result::unwrap).collect(); + + original.sort_unstable_by(|a, b| a.0.cmp(&b.0)); + list.sort_unstable_by(|a, b| a.0.cmp(&b.0)); + + assert_eq!(original.as_slice(), list.as_ref()); + + assert!(sync_kv.delete("first")?); + assert!(sync_kv.delete("second")?); + + Response::ok("smoke ok") + } + "/overwrite" => { + let overwrite = serde_json::json!({"v": 2}); + + sync_kv.put("k", serde_json::json!({"v": 1}))?; + sync_kv.put("k", overwrite.clone())?; + + assert_eq!(sync_kv.get("k")?, Some(overwrite)); + + assert!(sync_kv.delete("k")?); + + Response::ok("overwrite ok") + } + "/not_found" => { + assert_eq!(sync_kv.get::<()>("nope")?, None); + assert!(!sync_kv.delete("nope")?); + + Response::ok("not_found ok") + } + "/list" => { + const KEYS_LEN: usize = 10; + let keys: [_; KEYS_LEN] = std::array::from_fn(|i| format!("k{i}")); + + for (i, key) in keys.iter().enumerate() { + sync_kv.put(key, serde_json::json!({ "i": i }))?; + } + + let count = { + let list: SyncKvIterator = sync_kv.list()?; + list.count() + }; + + assert_eq!(count, KEYS_LEN); + + for key in keys { + assert!(sync_kv.delete(&key)?); + } + + Response::ok("list ok") + } + "/list_options" => { + // Seed: a, b, c, d, e + for k in ["a", "b", "c", "d", "e"] { + sync_kv.put(k, serde_json::json!({"k": k}))?; + } + + // start_after("b") yields c, d, e + let after_b: Vec = sync_kv + .list_with_options::( + &SyncKvListOptionsBuilder::new().start_after("b").build(), + )? + .collect::>>()? + .into_iter() + .map(|(k, _)| k) + .collect(); + assert_eq!(after_b, vec!["c", "d", "e"]); + + // start("b") yields b, c, d, e + let from_b: Vec = sync_kv + .list_with_options::( + &SyncKvListOptionsBuilder::new().start("b").build(), + )? + .collect::>>()? + .into_iter() + .map(|(k, _)| k) + .collect(); + assert_eq!(from_b, vec!["b", "c", "d", "e"]); + + // limit(2) + reverse(true) yields e, d + let last_two: Vec = sync_kv + .list_with_options::( + &SyncKvListOptionsBuilder::new() + .reverse(true) + .limit(2) + .build(), + )? + .collect::>>()? + .into_iter() + .map(|(k, _)| k) + .collect(); + assert_eq!(last_two, vec!["e", "d"]); + + // end("c") yields a, b (exclusive) + let until_c: Vec = sync_kv + .list_with_options::( + &SyncKvListOptionsBuilder::new().end("c").build(), + )? + .collect::>>()? + .into_iter() + .map(|(k, _)| k) + .collect(); + assert_eq!(until_c, vec!["a", "b"]); + + for k in ["a", "b", "c", "d", "e"] { + assert!(sync_kv.delete(k)?); + } + + Response::ok("list_options ok") + } + "/persist_fill" => { + const KEYS_LEN: usize = 10; + let keys: [_; KEYS_LEN] = std::array::from_fn(|i| format!("k{i}")); + + for (i, key) in keys.iter().enumerate() { + sync_kv.put(key, serde_json::json!({ "i": i }))?; + } + + Response::ok("persist_fill ok") + } + "/persist_check" => { + const KEYS_LEN: usize = 10; + let keys: [_; KEYS_LEN] = std::array::from_fn(|i| format!("k{i}")); + + for (i, key) in keys.iter().enumerate() { + let val: Option = sync_kv.get(key)?; + + assert!(val.is_some()); + + assert_eq!(val, Some(serde_json::json!({"i": i}))); + } + + Response::ok("persist_check ok") + } + "/persist_cleanup" => { + let list: SyncKvIterator = sync_kv.list()?; + + let keys_collected: Box<[_]> = + list.filter_map(|e| e.ok().map(|(k, _)| k)).collect(); + + for key in keys_collected { + assert!(sync_kv.delete(&key)?); + } + + Response::ok("persist_cleanup ok") + } + _ => Response::error("unknown test", 404), + } + } +} + +#[worker::send] +pub async fn handle_synchronous_storage_smoke( + _req: Request, + env: Env, + _data: crate::SomeSharedData, +) -> Result { + let namespace = env.durable_object("SYNCHRONOUS_STORAGE")?; + let stub = namespace.unique_id()?.get_stub()?; + stub.fetch_with_str("http://fake-host/smoke").await +} + +#[worker::send] +pub async fn handle_synchronous_storage_overwrite( + _req: Request, + env: Env, + _data: crate::SomeSharedData, +) -> Result { + let namespace = env.durable_object("SYNCHRONOUS_STORAGE")?; + let stub = namespace.unique_id()?.get_stub()?; + stub.fetch_with_str("http://fake-host/overwrite").await +} + +#[worker::send] +pub async fn handle_synchronous_storage_not_found( + _req: Request, + env: Env, + _data: crate::SomeSharedData, +) -> Result { + let namespace = env.durable_object("SYNCHRONOUS_STORAGE")?; + let stub = namespace.unique_id()?.get_stub()?; + stub.fetch_with_str("http://fake-host/not_found").await +} + +#[worker::send] +pub async fn handle_synchronous_storage_list( + _req: Request, + env: Env, + _data: crate::SomeSharedData, +) -> Result { + let namespace = env.durable_object("SYNCHRONOUS_STORAGE")?; + let stub = namespace.unique_id()?.get_stub()?; + stub.fetch_with_str("http://fake-host/list").await +} + +#[worker::send] +pub async fn handle_synchronous_storage_list_options( + _req: Request, + env: Env, + _data: crate::SomeSharedData, +) -> Result { + let namespace = env.durable_object("SYNCHRONOUS_STORAGE")?; + let stub = namespace.unique_id()?.get_stub()?; + stub.fetch_with_str("http://fake-host/list_options").await +} + +#[worker::send] +pub async fn handle_synchronous_storage_persist_fill( + _req: Request, + env: Env, + _data: crate::SomeSharedData, +) -> Result { + let namespace = env.durable_object("SYNCHRONOUS_STORAGE")?; + let stub = namespace.id_from_name("singleton")?.get_stub()?; + stub.fetch_with_str("http://fake-host/persist_fill").await +} + +#[worker::send] +pub async fn handle_synchronous_storage_persist_check( + _req: Request, + env: Env, + _data: crate::SomeSharedData, +) -> Result { + let namespace = env.durable_object("SYNCHRONOUS_STORAGE")?; + let stub = namespace.id_from_name("singleton")?.get_stub()?; + stub.fetch_with_str("http://fake-host/persist_check").await +} + +#[worker::send] +pub async fn handle_synchronous_storage_persist_cleanup( + _req: Request, + env: Env, + _data: crate::SomeSharedData, +) -> Result { + let namespace = env.durable_object("SYNCHRONOUS_STORAGE")?; + let stub = namespace.id_from_name("singleton")?.get_stub()?; + stub.fetch_with_str("http://fake-host/persist_cleanup") + .await +} diff --git a/test/tests/mf.ts b/test/tests/mf.ts index 3c881b187..18cffe0cb 100644 --- a/test/tests/mf.ts +++ b/test/tests/mf.ts @@ -75,6 +75,10 @@ const mf_instance = new Miniflare({ className: "SqlIterator", useSQLite: true, }, + SYNCHRONOUS_STORAGE: { + className: "SynchronousStorage", + useSQLite: true + }, }, kvNamespaces: ["SOME_NAMESPACE", "FILE_SIZES", "TEST"], serviceBindings: { diff --git a/test/tests/synchronous_storage.spec.ts b/test/tests/synchronous_storage.spec.ts new file mode 100644 index 000000000..1ada2832b --- /dev/null +++ b/test/tests/synchronous_storage.spec.ts @@ -0,0 +1,62 @@ +import { describe, test, expect } from "vitest"; +import { mf, mfUrl } from "./mf"; + +describe("synchronous api durable object", () => { + test("smoke", async () => { + const resp = await mf.dispatchFetch(`${mfUrl}synchronous-storage/smoke`); + expect(resp.status).toBe(200); + expect(await resp.text()).toBe("smoke ok"); + }); + + test("overwrite", async () => { + const resp = await mf.dispatchFetch(`${mfUrl}synchronous-storage/overwrite`); + expect(resp.status).toBe(200); + expect(await resp.text()).toBe("overwrite ok"); + }); + + test("not_found", async () => { + const resp = await mf.dispatchFetch(`${mfUrl}synchronous-storage/not_found`); + expect(resp.status).toBe(200); + expect(await resp.text()).toBe("not_found ok"); + }); + + test("list", async () => { + const resp = await mf.dispatchFetch(`${mfUrl}synchronous-storage/list`); + expect(resp.status).toBe(200); + expect(await resp.text()).toBe("list ok"); + }); + + test("list_options", async () => { + const resp = await mf.dispatchFetch( + `${mfUrl}synchronous-storage/list_options` + ); + expect(resp.status).toBe(200); + expect(await resp.text()).toBe("list_options ok"); + }); + + describe.sequential("persist", () => { + test("fill", async () => { + const resp = await mf.dispatchFetch( + `${mfUrl}synchronous-storage/persist_fill` + ); + expect(resp.status).toBe(200); + expect(await resp.text()).toBe("persist_fill ok"); + }); + + test("check", async () => { + const resp = await mf.dispatchFetch( + `${mfUrl}synchronous-storage/persist_check` + ); + expect(resp.status).toBe(200); + expect(await resp.text()).toBe("persist_check ok"); + }); + + test("cleanup", async () => { + const resp = await mf.dispatchFetch( + `${mfUrl}synchronous-storage/persist_cleanup` + ); + expect(resp.status).toBe(200); + expect(await resp.text()).toBe("persist_cleanup ok"); + }); + }); +}); diff --git a/test/wrangler.toml b/test/wrangler.toml index 16b49d87e..9598398d5 100644 --- a/test/wrangler.toml +++ b/test/wrangler.toml @@ -31,6 +31,7 @@ bindings = [ { name = "SQL_ITERATOR", class_name = "SqlIterator" }, { name = "MY_CLASS", class_name = "MyClass" }, { name = "ECHO_CONTAINER", class_name = "EchoContainer" }, + { name = "SYNCHRONOUS_STORAGE", class_name = "SynchronousStorage" }, ] [[analytics_engine_datasets]] @@ -74,7 +75,12 @@ command = "WASM_BINDGEN_BIN=../wasm-bindgen/target/debug/wasm-bindgen ../target/ [[migrations]] tag = "v1" -new_sqlite_classes = ["SqlCounter", "SqlIterator", "EchoContainer"] +new_sqlite_classes = [ + "SqlCounter", + "SqlIterator", + "EchoContainer", + "SynchronousStorage", +] [[secrets_store_secrets]] binding = "SECRETS" diff --git a/worker-sys/src/types/durable_object.rs b/worker-sys/src/types/durable_object.rs index c1d3b2704..c67bb8edc 100644 --- a/worker-sys/src/types/durable_object.rs +++ b/worker-sys/src/types/durable_object.rs @@ -6,6 +6,7 @@ mod namespace; mod sql_storage; mod state; mod storage; +mod sync_kv_storage; mod transaction; pub use container::*; @@ -14,6 +15,7 @@ pub use namespace::*; pub use sql_storage::*; pub use state::*; pub use storage::*; +pub use sync_kv_storage::*; pub use transaction::*; #[wasm_bindgen] diff --git a/worker-sys/src/types/durable_object/storage.rs b/worker-sys/src/types/durable_object/storage.rs index 08dadaa97..77175abad 100644 --- a/worker-sys/src/types/durable_object/storage.rs +++ b/worker-sys/src/types/durable_object/storage.rs @@ -77,4 +77,7 @@ extern "C" { #[wasm_bindgen(method, getter)] pub fn sql(this: &DurableObjectStorage) -> crate::types::SqlStorage; + + #[wasm_bindgen(method, getter)] + pub fn kv(this: &DurableObjectStorage) -> crate::types::SyncKvStorage; } diff --git a/worker-sys/src/types/durable_object/sync_kv_storage.rs b/worker-sys/src/types/durable_object/sync_kv_storage.rs new file mode 100644 index 000000000..99d2cefbc --- /dev/null +++ b/worker-sys/src/types/durable_object/sync_kv_storage.rs @@ -0,0 +1,104 @@ +#[allow(unused_imports)] +use js_sys::*; +use wasm_bindgen::prelude::*; + +#[wasm_bindgen] +extern "C" { + #[wasm_bindgen(extends = Object)] + #[derive(Debug, Clone, PartialEq, Eq)] + pub type SyncKvStorage; + + /// Retrieves the value associated with the given key. + /// + /// Returns `undefined` if the key does not exist. + #[wasm_bindgen(method)] + pub fn get(this: &SyncKvStorage, key: &str) -> JsValue; + /// Retrieves the value associated with the given key. + /// + /// Returns `undefined` if the key does not exist. + #[wasm_bindgen(method, catch, js_name = "get")] + pub fn try_get(this: &SyncKvStorage, key: &str) -> Result; + + /// Stores the value at the given key. + #[wasm_bindgen(method)] + pub fn put(this: &SyncKvStorage, key: &str, value: &JsValue); + /// Stores the value at the given key. + #[wasm_bindgen(method, catch, js_name = "put")] + pub fn try_put(this: &SyncKvStorage, key: &str, value: &JsValue) -> Result<(), JsValue>; + + /// Deletes the key. Returns `true` if the key existed and was removed. + #[wasm_bindgen(method)] + pub fn delete(this: &SyncKvStorage, key: &str) -> bool; + /// Deletes the key. Returns `true` if the key existed and was removed. + #[wasm_bindgen(method, catch, js_name = "delete")] + pub fn try_delete(this: &SyncKvStorage, key: &str) -> Result; + + /// Returns an iterable of `[key, value]` pairs in ascending lexicographic order. + #[wasm_bindgen(method)] + pub fn list(this: &SyncKvStorage) -> JsValue; + /// Returns an iterable of `[key, value]` pairs in ascending lexicographic order. + #[wasm_bindgen(method, catch, js_name = "list")] + pub fn try_list(this: &SyncKvStorage) -> Result; + /// Returns an iterable of `[key, value]` pairs filtered by the given options. + #[wasm_bindgen(method, js_name = "list")] + pub fn list_with_options(this: &SyncKvStorage, options: &SyncKvListOptions) -> JsValue; + /// Returns an iterable of `[key, value]` pairs filtered by the given options. + #[wasm_bindgen(method, catch, js_name = "list")] + pub fn try_list_with_options( + this: &SyncKvStorage, + options: &SyncKvListOptions, + ) -> Result; +} + +#[wasm_bindgen] +extern "C" { + #[wasm_bindgen(extends = Object)] + #[derive(Debug, Clone, PartialEq, Eq)] + pub type SyncKvListOptions; + + /// Key at which the list results should start, inclusive. + #[wasm_bindgen(method, getter)] + pub fn start(this: &SyncKvListOptions) -> Option; + #[wasm_bindgen(method, setter)] + pub fn set_start(this: &SyncKvListOptions, val: &str); + + /// Key at which the list results should start, exclusive. Cannot be used + /// simultaneously with `start`. + #[wasm_bindgen(method, getter, js_name = "startAfter")] + pub fn start_after(this: &SyncKvListOptions) -> Option; + #[wasm_bindgen(method, setter, js_name = "startAfter")] + pub fn set_start_after(this: &SyncKvListOptions, val: &str); + + /// Key at which the list results should end, exclusive. + #[wasm_bindgen(method, getter)] + pub fn end(this: &SyncKvListOptions) -> Option; + #[wasm_bindgen(method, setter)] + pub fn set_end(this: &SyncKvListOptions, val: &str); + + /// Restricts results to only include key-value pairs whose keys begin with the prefix. + #[wasm_bindgen(method, getter)] + pub fn prefix(this: &SyncKvListOptions) -> Option; + #[wasm_bindgen(method, setter)] + pub fn set_prefix(this: &SyncKvListOptions, val: &str); + + /// If true, return results in descending lexicographic order. + #[wasm_bindgen(method, getter)] + pub fn reverse(this: &SyncKvListOptions) -> Option; + #[wasm_bindgen(method, setter)] + pub fn set_reverse(this: &SyncKvListOptions, val: bool); + + /// Maximum number of key-value pairs to return. + #[wasm_bindgen(method, getter)] + pub fn limit(this: &SyncKvListOptions) -> Option; + #[wasm_bindgen(method, setter)] + pub fn set_limit(this: &SyncKvListOptions, val: f64); +} + +impl SyncKvListOptions { + /// Create an empty `SyncKvListOptions` JS object. + #[allow(clippy::new_without_default)] + pub fn new() -> Self { + use wasm_bindgen::JsCast; + JsCast::unchecked_into(js_sys::Object::new()) + } +} diff --git a/worker/src/durable.rs b/worker/src/durable.rs index 446866b49..8218629eb 100644 --- a/worker/src/durable.rs +++ b/worker/src/durable.rs @@ -565,10 +565,17 @@ impl Storage { .map(|_| ()) } - // Add new method to access SQLite APIs + /// Access the SQLite APIs exposed at `ctx.storage.sql`. pub fn sql(&self) -> crate::sql::SqlStorage { crate::sql::SqlStorage::new(self.inner.sql()) } + + /// Access the synchronous key-value APIs exposed at `ctx.storage.kv`. + /// + /// This is Cloudflare's Synchronous KV API for SQLite-backed Durable Objects. + pub fn kv(&self) -> crate::sync_kv::SyncKvStorage { + crate::sync_kv::SyncKvStorage::new(self.inner.kv()) + } } #[derive(Debug)] diff --git a/worker/src/lib.rs b/worker/src/lib.rs index cc90c9f17..86af2f53f 100644 --- a/worker/src/lib.rs +++ b/worker/src/lib.rs @@ -240,6 +240,7 @@ pub mod signal; mod socket; mod sql; mod streams; +mod sync_kv; mod version; mod websocket; @@ -278,3 +279,4 @@ pub type HttpRequest = ::http::Request; pub type HttpResponse = ::http::Response; pub use crate::sql::*; +pub use crate::sync_kv::*; diff --git a/worker/src/sync_kv.rs b/worker/src/sync_kv.rs new file mode 100644 index 000000000..c96288f74 --- /dev/null +++ b/worker/src/sync_kv.rs @@ -0,0 +1,227 @@ +//! Cloudflare Durable Objects' Synchronous KV API exposed via +//! [`Storage::kv`](crate::Storage::kv). +//! +//! This is the `ctx.storage.kv` API available on SQLite-backed Durable Objects. +//! Entries are stored in the Durable Object's hidden `__cf_kv` SQLite table. +//! +//! [`SyncKvListOptions`] is re-exported directly from `worker-sys` as an +//! imported JS handle; build one with [`SyncKvListOptionsBuilder`]. +//! [`SyncKvStorage`] is a thin Rust wrapper that adds `serde`-aware typed +//! `get`/`put`/`list` methods on top of the imported sys type. + +use core::fmt; +use std::marker::PhantomData; + +use serde::{de::DeserializeOwned, Serialize}; +use serde_wasm_bindgen as swb; +use wasm_bindgen::{JsCast as _, JsValue}; +use worker_sys::types::SyncKvStorage as SyncKvStorageSys; + +pub use worker_sys::types::SyncKvListOptions; + +use crate::{Error, Result}; + +/// Cloudflare Durable Objects' Synchronous KV API exposed by [`Storage::kv`](crate::Storage::kv). +/// +/// Wraps the imported [`worker_sys::types::SyncKvStorage`] handle and provides +/// typed `serde`-aware methods. +#[derive(Clone, Debug)] +pub struct SyncKvStorage { + inner: SyncKvStorageSys, +} + +// SAFETY: workers run single-threaded; matches the convention used by other +// Durable Object wrappers in this crate (e.g. `Storage`, `SqlStorage`). +unsafe impl Send for SyncKvStorage {} +unsafe impl Sync for SyncKvStorage {} + +impl SyncKvStorage { + pub(crate) fn new(inner: SyncKvStorageSys) -> Self { + Self { inner } + } + + /// Retrieves the value associated with the given key. + /// + /// Returns `Ok(None)` if the key does not exist. + pub fn get(&self, key: &str) -> Result> + where + T: DeserializeOwned, + { + let val = self.inner.try_get(key).map_err(Error::from)?; + if val.is_undefined() { + Ok(None) + } else { + Ok(Some(swb::from_value(val)?)) + } + } + + /// Stores the value and associates it with the given key. + pub fn put(&self, key: &str, value: T) -> Result<()> + where + T: Serialize, + { + let js = swb::to_value(&value)?; + self.inner.try_put(key, &js).map_err(Error::from) + } + + /// Deletes the key and associated value. + /// + /// Returns `Ok(true)` if the key existed and was removed. + pub fn delete(&self, key: &str) -> Result { + self.inner.try_delete(key).map_err(Error::from) + } + + /// Returns an iterator over all key-value pairs in ascending lexicographic order. + pub fn list(&self) -> Result> + where + T: DeserializeOwned, + { + let iterable = self.inner.try_list().map_err(Error::from)?; + SyncKvIterator::from_iterable(&iterable) + } + + /// Returns an iterator over key-value pairs that match the provided [`SyncKvListOptions`]. + pub fn list_with_options(&self, options: &SyncKvListOptions) -> Result> + where + T: DeserializeOwned, + { + let iterable = self + .inner + .try_list_with_options(options) + .map_err(Error::from)?; + SyncKvIterator::from_iterable(&iterable) + } +} + +/// Fluent builder for [`SyncKvListOptions`]. +/// +/// Each method mutates the underlying JS object in place and returns `self`. +/// [`build`](Self::build) returns the configured imported handle. +#[derive(Debug)] +pub struct SyncKvListOptionsBuilder { + inner: SyncKvListOptions, +} + +impl Default for SyncKvListOptionsBuilder { + fn default() -> Self { + Self { + inner: SyncKvListOptions::new(), + } + } +} + +impl SyncKvListOptionsBuilder { + /// Create an empty builder. + pub fn new() -> Self { + Self::default() + } + + /// Key at which the list results should start, inclusive. + pub fn start(self, val: &str) -> Self { + self.inner.set_start(val); + self + } + + /// Key at which the list results should start, exclusive. + /// Cannot be used simultaneously with [`start`](Self::start). + pub fn start_after(self, val: &str) -> Self { + self.inner.set_start_after(val); + self + } + + /// Key at which the list results should end, exclusive. + pub fn end(self, val: &str) -> Self { + self.inner.set_end(val); + self + } + + /// Restricts results to only include key-value pairs whose keys begin with the prefix. + pub fn prefix(self, val: &str) -> Self { + self.inner.set_prefix(val); + self + } + + /// If true, return results in descending lexicographic order. + pub fn reverse(self, val: bool) -> Self { + self.inner.set_reverse(val); + self + } + + /// Maximum number of key-value pairs to return. + pub fn limit(self, val: u32) -> Self { + self.inner.set_limit(val as f64); + self + } + + /// Consume the builder and return the configured [`SyncKvListOptions`]. + pub fn build(self) -> SyncKvListOptions { + self.inner + } +} + +/// Iterator over typed entries returned by [`SyncKvStorage::list`] and +/// [`SyncKvStorage::list_with_options`]. +/// +/// Each item yields the key together with its `serde`-deserialized value. +pub struct SyncKvIterator { + inner: js_sys::IntoIter, + _phantom: PhantomData, +} + +impl SyncKvIterator { + fn from_iterable(iterable: &JsValue) -> Result { + let inner = js_sys::try_iter(iterable)?.ok_or_else(|| { + Error::JsError("SyncKvStorage.list() did not return an iterable".into()) + })?; + Ok(Self { + inner, + _phantom: PhantomData, + }) + } +} + +impl Iterator for SyncKvIterator +where + T: DeserializeOwned, +{ + type Item = Result<(String, T)>; + + fn next(&mut self) -> Option { + let entry = match self.inner.next()? { + Ok(r) => r, + Err(e) => return Some(Err(Error::from(e))), + }; + + // workerd guarantees `[key, value]` tuples; treat anything else as a JS error. + let arr: js_sys::Array = match entry.dyn_into() { + Ok(a) => a, + Err(_) => { + return Some(Err(Error::JsError( + "Expected SyncKvStorage list entry to be an array".into(), + ))); + } + }; + + let key = match arr.get(0).as_string() { + Some(k) => k, + None => { + return Some(Err(Error::JsError( + "Expected SyncKvStorage list entry key to be a string".into(), + ))); + } + }; + + let val = match swb::from_value(arr.get(1)) { + Ok(v) => v, + Err(e) => return Some(Err(Error::from(e))), + }; + + Some(Ok((key, val))) + } +} + +impl fmt::Debug for SyncKvIterator { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("SyncKvIterator").finish() + } +}