P3

.config/wasm-pkg/config.toml

@@ -1,3 +1,4 @@
 [namespace_registries]
 wasi = "wasi.dev"
 componentized = "componentized.dev"
+wasmcloud = "wasmcloud.com"

components/as-keyvalue/README.md

@@ -1,3 +1,6 @@
 # as-keyvalue
 
-Adapter to back wasi:keyvalue with the componentized:valkey client.
+Adapter to back wasmcloud:keyvalue with the componentized:valkey client.
+
+> [!NOTE]
+> `wasi:keyvalue` is currently unsupportable with an async based client. It will be restored when either it updates it's own API to be async, or the component model is fully able to make async calls from sync functions.

components/as-keyvalue/src/lib.rs

@@ -1,10 +1,11 @@
 #![no_main]
 
-use componentized::valkey::store::{self as valkey, Connection, HelloOpts};
-use exports::wasi::keyvalue::atomics::{Cas, CasError, Guest as AtomicsGuest, GuestCas};
-use exports::wasi::keyvalue::batch::Guest as BatchGuest;
-use exports::wasi::keyvalue::store::{
-    Bucket, BucketBorrow, Error, Guest as StoreGuest, GuestBucket, KeyResponse,
+use componentized::valkey::store::{self as valkey, Connection, Duration, HelloOpts, HscanOpts};
+use exports::wasmcloud::keyvalue::atomics::Guest as AtomicsGuest;
+use exports::wasmcloud::keyvalue::batch::Guest as BatchGuest;
+use exports::wasmcloud::keyvalue::store::Guest as StoreGuest;
+use exports::wasmcloud::keyvalue::types::{
+    Bucket, BucketBorrow, Error, Guest as TypesGuest, GuestBucket, KeyResponse, SetOptions,
 };
 use wasi::config::store::{self as config};
 
@@ -19,12 +20,14 @@ const KEY_PREFIX_KEY: &str = "key-prefix";
 const KEY_PREFIX_DEFAULT: &str = "";
 
 #[derive(Debug, Clone)]
-struct KeyvalueToValkey;
+struct AsKeyvalue;
 
-impl StoreGuest for KeyvalueToValkey {
+impl TypesGuest for AsKeyvalue {
     type Bucket = KeyvalueToValkeyBucket;
+}
 
-    fn open(identifier: String) -> Result<Bucket, Error> {
+impl StoreGuest for AsKeyvalue {
+    async fn open(identifier: String) -> Result<Bucket, Error> {
         let hostname: String = config::get(HOSTNAME_KEY)?.unwrap_or(HOSTNAME_DEFAULT.to_string());
         let port = config::get(PORT_KEY)?.unwrap_or(PORT_DEFAULT.to_string());
         let port: u16 = port
@@ -43,7 +46,9 @@ impl StoreGuest for KeyvalueToValkey {
             },
             client_name: None,
         };
-        let connection = valkey::connect(&hostname, port, Some(&opts))?;
+        let connection = valkey::connect(hostname, port, Some(opts))
+            .await
+            .map_err(|_| Error::StoreUnavailable)?;
 
         let key_prefix = config::get(KEY_PREFIX_KEY)?.unwrap_or(KEY_PREFIX_DEFAULT.to_string());
         let hash_key = format!("{key_prefix}{identifier}");
@@ -61,74 +66,99 @@ struct KeyvalueToValkeyBucket {
 }
 
 impl GuestBucket for KeyvalueToValkeyBucket {
-    fn get(&self, key: String) -> Result<Option<Vec<u8>>, Error> {
-        match self.connection.hget(&self.hash_key, &key)? {
-            Some(value) => Ok(Some(value.as_bytes().to_vec())),
+    async fn get(&self, key: String) -> Result<Option<Vec<u8>>, Error> {
+        match self.connection.hget(self.hash_key.clone(), key).await? {
+            Some(value) => Ok(Some(value.into_bytes())),
             None => Ok(None),
         }
     }
 
-    fn set(&self, key: String, value: Vec<u8>) -> Result<(), Error> {
+    async fn set(
+        &self,
+        key: String,
+        value: Vec<u8>,
+        options: Option<SetOptions>,
+    ) -> Result<(), Error> {
         let value = String::from_utf8(value).map_err(|e| Error::Other(e.to_string()))?;
-        Ok(self.connection.hset(&self.hash_key, &key, &value)?)
+        let options = options.unwrap_or(SetOptions {
+            ttl_ms: None,
+            if_not_exists: false,
+        });
+
+        if options.if_not_exists {
+            match self
+                .connection
+                .hsetnx(self.hash_key.clone(), key.clone(), value)
+                .await?
+            {
+                true => (),
+                false => Err(Error::PreconditionFailed)?,
+            }
+        } else {
+            self.connection
+                .hset(self.hash_key.clone(), key.clone(), value)
+                .await?
+        }
+        if let Some(ttl_ms) = options.ttl_ms {
+            let ttl: Duration = ttl_ms / 1000;
+            self.connection
+                .hexpire(self.hash_key.clone(), ttl, None, vec![key])
+                .await?;
+        }
+        Ok(())
     }
 
-    fn delete(&self, key: String) -> Result<(), Error> {
-        Ok(self.connection.hdel(&self.hash_key, &key)?)
+    async fn delete(&self, key: String) -> Result<(), Error> {
+        Ok(self.connection.hdel(self.hash_key.clone(), key).await?)
     }
 
-    fn exists(&self, key: String) -> Result<bool, Error> {
-        Ok(self.connection.hexists(&self.hash_key, &key)?)
+    async fn exists(&self, key: String) -> Result<bool, Error> {
+        Ok(self.connection.hexists(self.hash_key.clone(), key).await?)
     }
 
-    fn list_keys(&self, cursor: Option<String>) -> Result<KeyResponse, Error> {
-        if cursor.is_some() {
-            Err(Error::Other("cursor is not supported".to_string()))?;
-        }
+    async fn list_keys(
+        &self,
+        prefix: Option<String>,
+        cursor: Option<String>,
+    ) -> Result<KeyResponse, Error> {
+        let opts = HscanOpts {
+            match_: prefix.map(|prefix| format!("{}*", escape_glob(&prefix))),
+            count: None,
+            no_values: Some(true),
+        };
+        let (cursor, fields) = self
+            .connection
+            .hscan(self.hash_key.clone(), cursor, Some(opts))
+            .await?;
 
         Ok(KeyResponse {
-            cursor: None,
-            keys: self.connection.hkeys(&self.hash_key)?,
+            keys: fields.into_iter().map(|(key, _)| key).collect(),
+            cursor,
         })
     }
 }
 
-impl AtomicsGuest for KeyvalueToValkey {
-    type Cas = KeyvalueToValkeyCas;
-
-    fn increment(bucket: BucketBorrow<'_>, key: String, delta: i64) -> Result<i64, Error> {
+impl AtomicsGuest for AsKeyvalue {
+    async fn increment(bucket: BucketBorrow<'_>, key: String, delta: i64) -> Result<i64, Error> {
         let bucket: &KeyvalueToValkeyBucket = bucket.get();
 
-        Ok(bucket.connection.hincrby(&bucket.hash_key, &key, delta)?)
-    }
-
-    fn swap(_cas: Cas, _value: Vec<u8>) -> Result<(), CasError> {
-        todo!()
-    }
-}
-
-struct KeyvalueToValkeyCas;
-
-impl GuestCas for KeyvalueToValkeyCas {
-    fn new(_bucket: BucketBorrow<'_>, _key: String) -> Result<Cas, Error> {
-        todo!()
-    }
-
-    fn current(&self) -> Result<Option<Vec<u8>>, Error> {
-        todo!()
+        Ok(bucket
+            .connection
+            .hincrby(bucket.hash_key.clone(), key, delta)
+            .await?)
     }
 }
 
-impl BatchGuest for KeyvalueToValkey {
-    fn get_many(
+impl BatchGuest for AsKeyvalue {
+    async fn get_many(
         bucket: BucketBorrow<'_>,
         keys: Vec<String>,
     ) -> Result<Vec<Option<(String, Vec<u8>)>>, Error> {
         let bucket: &KeyvalueToValkeyBucket = bucket.get();
 
         let mut values: Vec<Option<(String, Vec<u8>)>> = vec![];
         for key in keys {
-            let value = match bucket.get(key.clone())? {
+            let value = match bucket.get(key.clone()).await? {
                 Some(value) => Some((key, value)),
                 None => None,
             };
@@ -138,27 +168,43 @@ impl BatchGuest for KeyvalueToValkey {
         Ok(values)
     }
 
-    fn set_many(bucket: BucketBorrow<'_>, key_values: Vec<(String, Vec<u8>)>) -> Result<(), Error> {
+    async fn set_many(
+        bucket: BucketBorrow<'_>,
+        key_values: Vec<(String, Vec<u8>)>,
+    ) -> Result<(), Error> {
         let bucket: &KeyvalueToValkeyBucket = bucket.get();
 
         for (key, value) in key_values {
-            bucket.set(key, value)?;
+            bucket.set(key, value, None).await?;
         }
 
         Ok(())
     }
 
-    fn delete_many(bucket: BucketBorrow<'_>, keys: Vec<String>) -> Result<(), Error> {
+    async fn delete_many(bucket: BucketBorrow<'_>, keys: Vec<String>) -> Result<(), Error> {
         let bucket: &KeyvalueToValkeyBucket = bucket.get();
 
         for key in keys {
-            bucket.delete(key)?;
+            bucket.delete(key).await?;
         }
 
         Ok(())
     }
 }
 
+/// Escapes the glob-style pattern characters Valkey's MATCH interprets, so a prefix matches
+/// literally.
+fn escape_glob(value: &str) -> String {
+    let mut escaped = String::with_capacity(value.len());
+    for c in value.chars() {
+        if matches!(c, '*' | '?' | '[' | ']' | '\\') {
+            escaped.push('\\');
+        }
+        escaped.push(c);
+    }
+    escaped
+}
+
 impl From<config::Error> for Error {
     fn from(e: config::Error) -> Self {
         match e {
@@ -181,4 +227,4 @@ wit_bindgen::generate!({
     generate_all
 });
 
-export!(KeyvalueToValkey);
+export!(AsKeyvalue);

components/client/README.md

@@ -1,5 +1,8 @@
 # client
 
-Client for interacting with a Valkey server, as componentized:valkey and wasi:keyvalue.
+Client for interacting with a Valkey server, as componentized:valkey and wasmcloud:keyvalue.
+
+> [!NOTE]
+> `wasi:keyvalue` is currently unsupportable with an async based client. It will be restored when either it updates it's own API to be async, or the component model is fully able to make async calls from sync functions.
 
 Composes [`ops`](../ops/) with [`as-keyvalue`](../as-keyvalue/), exporting both.

components/dep-cli/Cargo.toml

@@ -4,6 +4,9 @@ version = "0.1.0"
 edition = "2021"
 license = "Apache-2.0"
 
+[lib]
+crate-type = ["cdylib"]
+
 [dependencies]
 clap = { version = "4.6.1", features = ["derive"] }
 wit-bindgen = { workspace = true }