kevy_client/
lib.rs

1//! kevy-client — unified KV facade so downstream code can switch between
2//! in-process embedded and TCP-server backends with one URL string.
3//!
4//! ```no_run
5//! use kevy_client::Connection;
6//!
7//! // Same business code regardless of backend:
8//! let mut conn = Connection::connect(std::env::var("MY_KEVY_URL").unwrap().as_str())?;
9//! conn.set(b"hello", b"world")?;
10//! assert_eq!(conn.get(b"hello")?, Some(b"world".to_vec()));
11//! # Ok::<(), kevy_client::KevyError>(())
12//! ```
13//!
14//! URL schemes:
15//! - `mem://`                       — in-process embedded, in-memory only
16//! - `mem://<name>`                 — shared in-process bus keyed by `<name>`
17//! - `file:///abs/path` /
18//!   `file://./rel/path`            — in-process embedded with persistence
19//! - `kevy://host[:port][/db]`      — TCP RESP, kevy-native scheme
20//! - `redis://host[:port][/db]`     — TCP RESP, standard Redis URL (alias)
21//! - `tcp://host[:port]`            — TCP RESP, raw (no SELECT round-trip)
22//!
23//! Auth (`redis://user:pass@…`) and TLS (`rediss://`) are rejected up front
24//! — kevy ships without either. The crate covers the full string/hash/list/
25//! set/zset + one-shot `PUBLISH` surface. The pub/sub *consumer* side is a
26//! separate [`Subscriber`] type — a subscribed connection cannot send
27//! normal commands, so it needs its own socket and lives outside the
28//! `Connection` enum. `mem://<name>` / `file:///path` route through a
29//! process-local registry so the publisher and consumer can find each
30//! other when both opens use the same URL. The trait-vs-enum design
31//! decision is enum for now (closed two-backend universe); see ROADMAP
32//! for the trait extension path.
33//!
34//! Beyond that base the wrap surface tracks the full server op surface:
35//! blocking pops (`blpop`/`brpop`/`bzpopmin`), hash field-TTL
36//! (`hexpire`/`hpexpire`/`hpersist`/`httl`), zset algebra
37//! (`zinterstore`/`zunionstore`/`zintercard` + WEIGHTS/AGGREGATE),
38//! declarative indexes (`idx_*`), the CDC change feed (`feed_*`), and
39//! non-atomic [`Connection::pipeline`] batching. The full
40//! **op-family × wrap-status matrix** lives in this crate's README —
41//! anything listed there as raw-only is still reachable through
42//! [`Connection::pipeline`] / [`Transaction::queue`] argv passthrough.
43
44#![forbid(unsafe_code)]
45#![warn(missing_docs)]
46
47use std::time::Duration;
48
49use kevy_embedded::Store;
50use kevy_resp_client::RespClient;
51
52mod blocking;
53mod cluster;
54mod cluster_coll;
55mod collections;
56mod feed;
57mod hash_ttl;
58mod index;
59mod pipeline;
60mod reply;
61mod scan;
62mod subscribe;
63mod subscribe_io;
64mod transaction;
65mod url;
66mod zalgebra;
67
68pub use blocking::ZPopHit;
69pub use cluster::ClusterClient;
70pub use feed::{FeedBatch, FeedFrame};
71pub use index::{IdxInfo, IdxPage, IdxRow, IdxType};
72pub use pipeline::PipelineBuf;
73pub use subscribe::{PubsubEvent, Subscriber, SubscriberEvents, SubscriberMessages};
74pub use transaction::{Transaction, TransactionReplies};
75
76/// Re-exports so downstream code can name the argument/reply/error types
77/// of the wraps without adding kevy-embedded / kevy-resp deps.
78pub use kevy_embedded::{HExpireCode, HExpireCond, KevyError, KevyResult, StoreError, ZAggregate};
79pub use kevy_resp::Reply;
80
81pub(crate) use reply::{
82    array_to_bulks, num_f64, num_u64, store_err, string, unexpected, vec2, vec3,
83};
84pub(crate) use url::{Target, parse_url, resolve_store};
85
86/// One open connection to a kevy backend, opaque about whether the backend
87/// is in-process or over TCP.
88pub enum Connection {
89    /// In-process [`kevy_embedded::Store`]. Boxed because `Store` is
90    /// sizeable (carries its `Config`, including the replica
91    /// upstream/backoff fields) and dwarfs the `RespClient` variant.
92    Embedded(Box<Store>),
93    /// TCP [`kevy_resp_client::RespClient`].
94    Remote(RespClient),
95}
96
97impl Connection {
98    /// Connect to a backend chosen by URL scheme.
99    ///
100    /// See the crate-level docs for the supported URL forms. Two
101    /// `Connection::connect` calls with the same `mem://<name>` or
102    /// `file:///path` URL share the same backing `Store` — and the same
103    /// pub/sub bus, so `Connection::publish` reaches a
104    /// `Subscriber::connect_channels` opened with the same URL.
105    pub fn connect(url: &str) -> KevyResult<Self> {
106        let parsed = parse_url(url)?;
107        match parsed {
108            Target::Remote(remote_url) => Ok(Self::Remote(RespClient::connect_url(&remote_url)?)),
109            embed => Ok(Self::Embedded(Box::new(resolve_store(&embed)?))),
110        }
111    }
112
113    /// Remote-only feature gate: hand back the RESP client, or explain
114    /// why the embedded backend can't serve `feature` (`Unsupported`,
115    /// pointing at the `Connection::Embedded` escape hatch).
116    pub(crate) fn remote(&mut self, feature: &str) -> KevyResult<&mut RespClient> {
117        match self {
118            Self::Embedded(_) => Err(KevyError::Unsupported(format!(
119                "{feature} is remote-only; on the embedded backend match \
120                 Connection::Embedded and use kevy_embedded::Store's typed API"
121            ))),
122            Self::Remote(c) => Ok(c),
123        }
124    }
125
126    /// `PING`. Returns `()` on `+PONG`, propagates any IO or RESP error.
127    /// The first thing every healthcheck calls.
128    pub fn ping(&mut self) -> KevyResult<()> {
129        match self {
130            Self::Embedded(_) => Ok(()),
131            Self::Remote(c) => match c.request_borrowed(&[b"PING"])? {
132                Reply::Simple(s) if s == b"PONG" => Ok(()),
133                Reply::Error(e) => Err(KevyError::Protocol(string(e))),
134                other => Err(unexpected(other)),
135            },
136        }
137    }
138
139    /// `SET key value`. Unconditional set (no NX/XX). Returns `()` on success.
140    pub fn set(&mut self, key: &[u8], value: &[u8]) -> KevyResult<()> {
141        match self {
142            Self::Embedded(s) => s.set(key, value).map(|_| ()),
143            Self::Remote(c) => match c.request_borrowed(&[b"SET", key, value])? {
144                Reply::Simple(s) if s == b"OK" => Ok(()),
145                Reply::Error(e) => Err(KevyError::Protocol(string(e))),
146                other => Err(unexpected(other)),
147            },
148        }
149    }
150
151    /// `GET key`. `None` if absent or expired.
152    pub fn get(&mut self, key: &[u8]) -> KevyResult<Option<Vec<u8>>> {
153        match self {
154            Self::Embedded(s) => s.get(key),
155            Self::Remote(c) => match c.request_borrowed(&[b"GET", key])? {
156                Reply::Bulk(v) => Ok(Some(v)),
157                Reply::Nil => Ok(None),
158                Reply::Error(e) => Err(KevyError::Protocol(string(e))),
159                other => Err(unexpected(other)),
160            },
161        }
162    }
163
164    /// `DEL key [key ...]`. Returns the count of keys that were actually
165    /// removed (existing + dropped). Missing keys don't contribute.
166    pub fn del(&mut self, keys: &[&[u8]]) -> KevyResult<usize> {
167        match self {
168            Self::Embedded(s) => s.del(keys),
169            Self::Remote(c) => {
170                let mut args: Vec<&[u8]> = Vec::with_capacity(keys.len() + 1);
171                args.push(b"DEL");
172                args.extend_from_slice(keys);
173                match c.request_borrowed(&args)? {
174                    Reply::Int(n) if n >= 0 => Ok(n as usize),
175                    Reply::Error(e) => Err(KevyError::Protocol(string(e))),
176                    other => Err(unexpected(other)),
177                }
178            }
179        }
180    }
181
182    /// `EXISTS key [key ...]`. Count of keys present (a single key can
183    /// contribute >1 if passed multiple times, matching Redis semantics).
184    pub fn exists(&mut self, keys: &[&[u8]]) -> KevyResult<usize> {
185        match self {
186            Self::Embedded(s) => s.exists(keys),
187            Self::Remote(c) => {
188                let mut args: Vec<&[u8]> = Vec::with_capacity(keys.len() + 1);
189                args.push(b"EXISTS");
190                args.extend_from_slice(keys);
191                match c.request_borrowed(&args)? {
192                    Reply::Int(n) if n >= 0 => Ok(n as usize),
193                    Reply::Error(e) => Err(KevyError::Protocol(string(e))),
194                    other => Err(unexpected(other)),
195                }
196            }
197        }
198    }
199
200    /// `INCR key`. Returns the post-increment value. Errors on non-integer
201    /// stored value.
202    pub fn incr(&mut self, key: &[u8]) -> KevyResult<i64> {
203        match self {
204            Self::Embedded(s) => s.incr(key),
205            Self::Remote(c) => match c.request_borrowed(&[b"INCR", key])? {
206                Reply::Int(n) => Ok(n),
207                Reply::Error(e) => Err(KevyError::Protocol(string(e))),
208                other => Err(unexpected(other)),
209            },
210        }
211    }
212
213    /// `INCRBY key delta`. Negative delta is `DECRBY`. Returns post-value.
214    pub fn incr_by(&mut self, key: &[u8], delta: i64) -> KevyResult<i64> {
215        match self {
216            Self::Embedded(s) => s.incr_by(key, delta),
217            Self::Remote(c) => {
218                let delta_s = delta.to_string();
219                match c.request_borrowed(&[b"INCRBY", key, delta_s.as_bytes()])? {
220                    Reply::Int(n) => Ok(n),
221                    Reply::Error(e) => Err(KevyError::Protocol(string(e))),
222                    other => Err(unexpected(other)),
223                }
224            }
225        }
226    }
227
228    /// `PEXPIRE key ttl_ms`. Returns whether the key existed and got a TTL.
229    pub fn expire(&mut self, key: &[u8], ttl: Duration) -> KevyResult<bool> {
230        match self {
231            Self::Embedded(s) => s.expire(key, ttl),
232            Self::Remote(c) => {
233                let ms = ttl.as_millis().min(i64::MAX as u128) as i64;
234                let ms_s = ms.to_string();
235                match c.request_borrowed(&[b"PEXPIRE", key, ms_s.as_bytes()])? {
236                    Reply::Int(1) => Ok(true),
237                    Reply::Int(0) => Ok(false),
238                    Reply::Error(e) => Err(KevyError::Protocol(string(e))),
239                    other => Err(unexpected(other)),
240                }
241            }
242        }
243    }
244
245    /// `PERSIST key`. Returns whether a TTL was actually removed.
246    pub fn persist(&mut self, key: &[u8]) -> KevyResult<bool> {
247        match self {
248            Self::Embedded(s) => s.persist(key),
249            Self::Remote(c) => match c.request_borrowed(&[b"PERSIST", key])? {
250                Reply::Int(1) => Ok(true),
251                Reply::Int(0) => Ok(false),
252                Reply::Error(e) => Err(KevyError::Protocol(string(e))),
253                other => Err(unexpected(other)),
254            },
255        }
256    }
257
258    /// `PTTL key`. Returns ms remaining, -2 if no key, -1 if key has no TTL.
259    pub fn ttl_ms(&mut self, key: &[u8]) -> KevyResult<i64> {
260        match self {
261            Self::Embedded(s) => Ok(s.ttl_ms(key)),
262            Self::Remote(c) => match c.request_borrowed(&[b"PTTL", key])? {
263                Reply::Int(n) => Ok(n),
264                Reply::Error(e) => Err(KevyError::Protocol(string(e))),
265                other => Err(unexpected(other)),
266            },
267        }
268    }
269
270    /// `TYPE key`. Returns the value's type as a Redis-style string (e.g.
271    /// `"string"`, `"hash"`, `"list"`, `"set"`, `"zset"`, or `"none"` if
272    /// the key doesn't exist).
273    pub fn type_of(&mut self, key: &[u8]) -> KevyResult<String> {
274        match self {
275            Self::Embedded(s) => Ok(s.type_of(key).to_string()),
276            Self::Remote(c) => match c.request_borrowed(&[b"TYPE", key])? {
277                Reply::Simple(s) => Ok(string(s)),
278                Reply::Error(e) => Err(KevyError::Protocol(string(e))),
279                other => Err(unexpected(other)),
280            },
281        }
282    }
283
284    /// `DBSIZE`. Total live keys at the time of the call.
285    pub fn dbsize(&mut self) -> KevyResult<usize> {
286        match self {
287            Self::Embedded(s) => Ok(s.dbsize()),
288            Self::Remote(c) => match c.request_borrowed(&[b"DBSIZE"])? {
289                Reply::Int(n) if n >= 0 => Ok(n as usize),
290                Reply::Error(e) => Err(KevyError::Protocol(string(e))),
291                other => Err(unexpected(other)),
292            },
293        }
294    }
295
296    /// `FLUSHALL`. Drops every key. Persistence remains opted-in; embedded
297    /// `with_persist` will rewrite the AOF on its next sync cycle.
298    ///
299    /// Named `flushall` — **not** `flush` — to avoid colliding with
300    /// `Write::flush`'s "sync buffered writes to disk" meaning; this WIPES the
301    /// store rather than persisting it.
302    pub fn flushall(&mut self) -> KevyResult<()> {
303        match self {
304            Self::Embedded(s) => s.flushall(),
305            Self::Remote(c) => match c.request_borrowed(&[b"FLUSHALL"])? {
306                Reply::Simple(s) if s == b"OK" => Ok(()),
307                Reply::Error(e) => Err(KevyError::Protocol(string(e))),
308                other => Err(unexpected(other)),
309            },
310        }
311    }
312
313    /// `SET key value PX ttl_ms`. Convenience for the common
314    /// "cache with expiry" pattern; equivalent to `set` + `expire` but
315    /// atomic.
316    pub fn set_with_ttl(&mut self, key: &[u8], value: &[u8], ttl: Duration) -> KevyResult<()> {
317        match self {
318            Self::Embedded(s) => s.set_with_ttl(key, value, ttl).map(|_| ()),
319            Self::Remote(c) => {
320                let ms = ttl.as_millis().min(i64::MAX as u128) as i64;
321                let ms_s = ms.to_string();
322                match c.request_borrowed(&[b"SET", key, value, b"PX", ms_s.as_bytes()])? {
323                    Reply::Simple(s) if s == b"OK" => Ok(()),
324                    Reply::Error(e) => Err(KevyError::Protocol(string(e))),
325                    other => Err(unexpected(other)),
326                }
327            }
328        }
329    }
330
331    /// `MGET key [key ...]` — one reply per key, `None` for missing /
332    /// wrong-type. Returns in the same order as `keys`.
333    pub fn mget(&mut self, keys: &[&[u8]]) -> KevyResult<Vec<Option<Vec<u8>>>> {
334        match self {
335            Self::Embedded(s) => keys.iter().map(|k| s.get(k)).collect(),
336            Self::Remote(c) => {
337                let mut args: Vec<&[u8]> = Vec::with_capacity(keys.len() + 1);
338                args.push(b"MGET");
339                args.extend_from_slice(keys);
340                match c.request_borrowed(&args)? {
341                    Reply::Array(items) => items
342                        .into_iter()
343                        .map(|r| match r {
344                            Reply::Bulk(v) => Ok(Some(v)),
345                            Reply::Nil => Ok(None),
346                            other => Err(unexpected(other)),
347                        })
348                        .collect(),
349                    Reply::Error(e) => Err(KevyError::Protocol(string(e))),
350                    other => Err(unexpected(other)),
351                }
352            }
353        }
354    }
355
356    /// `MSET key value [key value ...]` — set every pair atomically.
357    pub fn mset(&mut self, pairs: &[(&[u8], &[u8])]) -> KevyResult<()> {
358        match self {
359            Self::Embedded(s) => {
360                for (k, v) in pairs {
361                    s.set(k, v)?;
362                }
363                Ok(())
364            }
365            Self::Remote(c) => {
366                let mut args: Vec<&[u8]> = Vec::with_capacity(pairs.len() * 2 + 1);
367                args.push(b"MSET");
368                for &(k, v) in pairs {
369                    args.push(k);
370                    args.push(v);
371                }
372                match c.request_borrowed(&args)? {
373                    Reply::Simple(s) if s == b"OK" => Ok(()),
374                    Reply::Error(e) => Err(KevyError::Protocol(string(e))),
375                    other => Err(unexpected(other)),
376                }
377            }
378        }
379    }
380
381    /// `PUBLISH channel message`. Returns the count of subscribers
382    /// that received the message.
383    ///
384    /// The embedded backend has a real in-process pub/sub
385    /// bus: when a [`Subscriber`] is open against the same `mem://<name>`
386    /// or `file:///path` URL, this delivers there and returns the actual
387    /// receiver count. Anonymous `mem://` keeps the old "no subscribers,
388    /// returns 0" behaviour (the URL is its own bus, by design).
389    ///
390    /// The pub/sub *consumer* side lives in [`Subscriber`]. On the remote
391    /// backend a subscribed TCP connection cannot send normal commands
392    /// per the RESP spec; the embedded backend has no such restriction
393    /// but `Subscriber` is still a distinct type for API symmetry.
394    pub fn publish(&mut self, channel: &[u8], message: &[u8]) -> KevyResult<usize> {
395        match self {
396            Self::Embedded(s) => Ok(s.publish(channel, message)),
397            Self::Remote(c) => match c.request_borrowed(&[b"PUBLISH", channel, message])? {
398                Reply::Int(n) if n >= 0 => Ok(n as usize),
399                Reply::Error(e) => Err(KevyError::Protocol(string(e))),
400                other => Err(unexpected(other)),
401            },
402        }
403    }
404}
405
406#[cfg(test)]
407#[path = "lib_tests.rs"]
408mod tests;
kevy_client/lib.rs

kevy_client/
lib.rs
