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::open(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::<(), std::io::Error>(())
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. v1.1.0 added the full string/hash/list/set/
25//! zset + one-shot `PUBLISH` surface. v1.2.0 added the pub/sub *consumer*
26//! side as a separate [`Subscriber`] type — a subscribed connection cannot
27//! send normal commands, so it needs its own socket and lives outside the
28//! `Connection` enum. v1.3.0 routes `mem://<name>` / `file:///path` through
29//! a 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//! v1.14.0 closes the wrap-parity gap against the kevy 3.17 server op
35//! surface: 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::io;
48use std::time::Duration;
49
50use kevy_embedded::Store;
51use kevy_resp_client::RespClient;
52
53mod blocking;
54mod cluster;
55mod cluster_coll;
56mod collections;
57mod feed;
58mod hash_ttl;
59mod index;
60mod pipeline;
61mod reply;
62mod scan;
63mod subscribe;
64mod subscribe_io;
65mod transaction;
66mod url;
67mod zalgebra;
68
69pub use blocking::ZPopHit;
70pub use cluster::ClusterClient;
71pub use feed::{FeedBatch, FeedFrame};
72pub use index::{IdxInfo, IdxPage, IdxRow, IdxType};
73pub use pipeline::PipelineBuf;
74pub use subscribe::{PubsubEvent, Subscriber, SubscriberEvents, SubscriberMessages};
75pub use transaction::{Transaction, TransactionReplies};
76
77/// Re-exports so downstream code can name the argument/reply types of
78/// the v1.14.0 wraps without adding kevy-embedded / kevy-resp deps.
79pub use kevy_embedded::{HExpireCode, HExpireCond, ZAggregate};
80pub use kevy_resp::Reply;
81
82pub(crate) use reply::{array_to_bulks, num_f64, num_u64, store_err, string, unexpected, vec2, vec3};
83pub(crate) use url::{Target, parse_url, resolve_store};
84
85/// One open connection to a kevy backend, opaque about whether the backend
86/// is in-process or over TCP.
87pub enum Connection {
88    /// In-process [`kevy_embedded::Store`]. Boxed because `Store` is
89    /// sizeable (carries its `Config`, including v1.20 replica
90    /// upstream/backoff fields) and dwarfs the `RespClient` variant.
91    Embedded(Box<Store>),
92    /// TCP [`kevy_resp_client::RespClient`].
93    Remote(RespClient),
94}
95
96impl Connection {
97    /// Open a backend chosen by URL scheme.
98    ///
99    /// See the crate-level docs for the supported URL forms. From v1.3.0,
100    /// two `Connection::open` calls with the same `mem://<name>` or
101    /// `file:///path` URL share the same backing `Store` — and the same
102    /// pub/sub bus, so `Connection::publish` reaches a `Subscriber::open`
103    /// opened with the same URL.
104    pub fn open(url: &str) -> io::Result<Self> {
105        let parsed = parse_url(url)?;
106        match parsed {
107            Target::Remote(remote_url) => Ok(Self::Remote(RespClient::from_url(&remote_url)?)),
108            embed => Ok(Self::Embedded(Box::new(resolve_store(&embed)?))),
109        }
110    }
111
112    /// Remote-only feature gate: hand back the RESP client, or explain
113    /// why the embedded backend can't serve `feature` (`Unsupported`,
114    /// pointing at the `Connection::Embedded` escape hatch).
115    pub(crate) fn remote(&mut self, feature: &str) -> io::Result<&mut RespClient> {
116        match self {
117            Self::Embedded(_) => Err(io::Error::new(
118                io::ErrorKind::Unsupported,
119                format!(
120                    "{feature} is remote-only; on the embedded backend match \
121                     Connection::Embedded and use kevy_embedded::Store's typed API"
122                ),
123            )),
124            Self::Remote(c) => Ok(c),
125        }
126    }
127
128    /// `PING`. Returns `()` on `+PONG`, propagates any IO or RESP error.
129    /// The first thing every healthcheck calls.
130    pub fn ping(&mut self) -> io::Result<()> {
131        match self {
132            Self::Embedded(_) => Ok(()),
133            Self::Remote(c) => match c.request_borrowed(&[b"PING"])? {
134                Reply::Simple(s) if s == b"PONG" => Ok(()),
135                Reply::Error(e) => Err(io::Error::other(string(e))),
136                other => Err(unexpected(other)),
137            },
138        }
139    }
140
141    /// `SET key value`. Unconditional set (no NX/XX). Returns `()` on success.
142    pub fn set(&mut self, key: &[u8], value: &[u8]) -> io::Result<()> {
143        match self {
144            Self::Embedded(s) => s.set(key, value).map(|_| ()),
145            Self::Remote(c) => match c.request_borrowed(&[b"SET", key, value])? {
146                Reply::Simple(s) if s == b"OK" => Ok(()),
147                Reply::Error(e) => Err(io::Error::other(string(e))),
148                other => Err(unexpected(other)),
149            },
150        }
151    }
152
153    /// `GET key`. `None` if absent or expired.
154    pub fn get(&mut self, key: &[u8]) -> io::Result<Option<Vec<u8>>> {
155        match self {
156            Self::Embedded(s) => s.get(key),
157            Self::Remote(c) => match c.request_borrowed(&[b"GET", key])? {
158                Reply::Bulk(v) => Ok(Some(v)),
159                Reply::Nil => Ok(None),
160                Reply::Error(e) => Err(io::Error::other(string(e))),
161                other => Err(unexpected(other)),
162            },
163        }
164    }
165
166    /// `DEL key [key ...]`. Returns the count of keys that were actually
167    /// removed (existing + dropped). Missing keys don't contribute.
168    pub fn del(&mut self, keys: &[&[u8]]) -> io::Result<usize> {
169        match self {
170            Self::Embedded(s) => s.del(keys),
171            Self::Remote(c) => {
172                let mut args = Vec::with_capacity(keys.len() + 1);
173                args.push(b"DEL".to_vec());
174                args.extend(keys.iter().map(|k| k.to_vec()));
175                match c.request(&args)? {
176                    Reply::Int(n) if n >= 0 => Ok(n as usize),
177                    Reply::Error(e) => Err(io::Error::other(string(e))),
178                    other => Err(unexpected(other)),
179                }
180            }
181        }
182    }
183
184    /// `EXISTS key [key ...]`. Count of keys present (a single key can
185    /// contribute >1 if passed multiple times, matching Redis semantics).
186    pub fn exists(&mut self, keys: &[&[u8]]) -> io::Result<usize> {
187        match self {
188            Self::Embedded(s) => s.exists(keys),
189            Self::Remote(c) => {
190                let mut args = Vec::with_capacity(keys.len() + 1);
191                args.push(b"EXISTS".to_vec());
192                args.extend(keys.iter().map(|k| k.to_vec()));
193                match c.request(&args)? {
194                    Reply::Int(n) if n >= 0 => Ok(n as usize),
195                    Reply::Error(e) => Err(io::Error::other(string(e))),
196                    other => Err(unexpected(other)),
197                }
198            }
199        }
200    }
201
202    /// `INCR key`. Returns the post-increment value. Errors on non-integer
203    /// stored value.
204    pub fn incr(&mut self, key: &[u8]) -> io::Result<i64> {
205        match self {
206            Self::Embedded(s) => s.incr(key),
207            Self::Remote(c) => match c.request_borrowed(&[b"INCR", key])? {
208                Reply::Int(n) => Ok(n),
209                Reply::Error(e) => Err(io::Error::other(string(e))),
210                other => Err(unexpected(other)),
211            },
212        }
213    }
214
215    /// `INCRBY key delta`. Negative delta is `DECRBY`. Returns post-value.
216    pub fn incr_by(&mut self, key: &[u8], delta: i64) -> io::Result<i64> {
217        match self {
218            Self::Embedded(s) => s.incr_by(key, delta),
219            Self::Remote(c) => {
220                let args = vec![
221                    b"INCRBY".to_vec(),
222                    key.to_vec(),
223                    delta.to_string().into_bytes(),
224                ];
225                match c.request(&args)? {
226                    Reply::Int(n) => Ok(n),
227                    Reply::Error(e) => Err(io::Error::other(string(e))),
228                    other => Err(unexpected(other)),
229                }
230            }
231        }
232    }
233
234    /// `PEXPIRE key ttl_ms`. Returns whether the key existed and got a TTL.
235    pub fn expire(&mut self, key: &[u8], ttl: Duration) -> io::Result<bool> {
236        match self {
237            Self::Embedded(s) => s.expire(key, ttl),
238            Self::Remote(c) => {
239                let ms = ttl.as_millis().min(i64::MAX as u128) as i64;
240                let args = vec![b"PEXPIRE".to_vec(), key.to_vec(), ms.to_string().into_bytes()];
241                match c.request(&args)? {
242                    Reply::Int(1) => Ok(true),
243                    Reply::Int(0) => Ok(false),
244                    Reply::Error(e) => Err(io::Error::other(string(e))),
245                    other => Err(unexpected(other)),
246                }
247            }
248        }
249    }
250
251    /// `PERSIST key`. Returns whether a TTL was actually removed.
252    pub fn persist(&mut self, key: &[u8]) -> io::Result<bool> {
253        match self {
254            Self::Embedded(s) => s.persist(key),
255            Self::Remote(c) => match c.request_borrowed(&[b"PERSIST", key])? {
256                Reply::Int(1) => Ok(true),
257                Reply::Int(0) => Ok(false),
258                Reply::Error(e) => Err(io::Error::other(string(e))),
259                other => Err(unexpected(other)),
260            },
261        }
262    }
263
264    /// `PTTL key`. Returns ms remaining, -2 if no key, -1 if key has no TTL.
265    pub fn ttl_ms(&mut self, key: &[u8]) -> io::Result<i64> {
266        match self {
267            Self::Embedded(s) => Ok(s.ttl_ms(key)),
268            Self::Remote(c) => match c.request_borrowed(&[b"PTTL", key])? {
269                Reply::Int(n) => Ok(n),
270                Reply::Error(e) => Err(io::Error::other(string(e))),
271                other => Err(unexpected(other)),
272            },
273        }
274    }
275
276    /// `TYPE key`. Returns the value's type as a Redis-style string (e.g.
277    /// `"string"`, `"hash"`, `"list"`, `"set"`, `"zset"`, or `"none"` if
278    /// the key doesn't exist).
279    pub fn type_of(&mut self, key: &[u8]) -> io::Result<String> {
280        match self {
281            Self::Embedded(s) => Ok(s.type_of(key).to_string()),
282            Self::Remote(c) => match c.request_borrowed(&[b"TYPE", key])? {
283                Reply::Simple(s) => Ok(string(s)),
284                Reply::Error(e) => Err(io::Error::other(string(e))),
285                other => Err(unexpected(other)),
286            },
287        }
288    }
289
290    /// `DBSIZE`. Total live keys at the time of the call.
291    pub fn dbsize(&mut self) -> io::Result<usize> {
292        match self {
293            Self::Embedded(s) => Ok(s.dbsize()),
294            Self::Remote(c) => match c.request_borrowed(&[b"DBSIZE"])? {
295                Reply::Int(n) if n >= 0 => Ok(n as usize),
296                Reply::Error(e) => Err(io::Error::other(string(e))),
297                other => Err(unexpected(other)),
298            },
299        }
300    }
301
302    /// `FLUSHALL`. Drops every key. Persistence remains opted-in; embedded
303    /// `with_persist` will rewrite the AOF on its next sync cycle.
304    ///
305    /// Named `flushall` — **not** `flush` — to avoid colliding with
306    /// `Write::flush`'s "sync buffered writes to disk" meaning; this WIPES the
307    /// store rather than persisting it.
308    pub fn flushall(&mut self) -> io::Result<()> {
309        match self {
310            Self::Embedded(s) => s.flushall(),
311            Self::Remote(c) => match c.request_borrowed(&[b"FLUSHALL"])? {
312                Reply::Simple(s) if s == b"OK" => Ok(()),
313                Reply::Error(e) => Err(io::Error::other(string(e))),
314                other => Err(unexpected(other)),
315            },
316        }
317    }
318
319    /// Deprecated alias for [`Self::flushall`]. The old name read like
320    /// `Write::flush` (sync-to-disk) but actually WIPES the store.
321    #[deprecated(
322        since = "1.8.0",
323        note = "renamed to `flushall`: `flush` collides with Write::flush (sync-to-disk); this WIPES the store"
324    )]
325    pub fn flush(&mut self) -> io::Result<()> {
326        self.flushall()
327    }
328
329    /// `SET key value PX ttl_ms`. Convenience for the common
330    /// "cache with expiry" pattern; equivalent to `set` + `expire` but
331    /// atomic.
332    pub fn set_with_ttl(&mut self, key: &[u8], value: &[u8], ttl: Duration) -> io::Result<()> {
333        match self {
334            Self::Embedded(s) => s.set_with_ttl(key, value, ttl).map(|_| ()),
335            Self::Remote(c) => {
336                let ms = ttl.as_millis().min(i64::MAX as u128) as i64;
337                let args = vec![
338                    b"SET".to_vec(),
339                    key.to_vec(),
340                    value.to_vec(),
341                    b"PX".to_vec(),
342                    ms.to_string().into_bytes(),
343                ];
344                match c.request(&args)? {
345                    Reply::Simple(s) if s == b"OK" => Ok(()),
346                    Reply::Error(e) => Err(io::Error::other(string(e))),
347                    other => Err(unexpected(other)),
348                }
349            }
350        }
351    }
352
353    /// `MGET key [key ...]` — one reply per key, `None` for missing /
354    /// wrong-type. Returns in the same order as `keys`.
355    pub fn mget(&mut self, keys: &[&[u8]]) -> io::Result<Vec<Option<Vec<u8>>>> {
356        match self {
357            Self::Embedded(s) => keys.iter().map(|k| s.get(k)).collect(),
358            Self::Remote(c) => {
359                let mut args = Vec::with_capacity(keys.len() + 1);
360                args.push(b"MGET".to_vec());
361                args.extend(keys.iter().map(|k| k.to_vec()));
362                match c.request(&args)? {
363                    Reply::Array(items) => items
364                        .into_iter()
365                        .map(|r| match r {
366                            Reply::Bulk(v) => Ok(Some(v)),
367                            Reply::Nil => Ok(None),
368                            other => Err(unexpected(other)),
369                        })
370                        .collect(),
371                    Reply::Error(e) => Err(io::Error::other(string(e))),
372                    other => Err(unexpected(other)),
373                }
374            }
375        }
376    }
377
378    /// `MSET key value [key value ...]` — set every pair atomically.
379    pub fn mset(&mut self, pairs: &[(&[u8], &[u8])]) -> io::Result<()> {
380        match self {
381            Self::Embedded(s) => {
382                for (k, v) in pairs {
383                    s.set(k, v)?;
384                }
385                Ok(())
386            }
387            Self::Remote(c) => {
388                let mut args = Vec::with_capacity(pairs.len() * 2 + 1);
389                args.push(b"MSET".to_vec());
390                for (k, v) in pairs {
391                    args.push(k.to_vec());
392                    args.push(v.to_vec());
393                }
394                match c.request(&args)? {
395                    Reply::Simple(s) if s == b"OK" => Ok(()),
396                    Reply::Error(e) => Err(io::Error::other(string(e))),
397                    other => Err(unexpected(other)),
398                }
399            }
400        }
401    }
402
403    /// `PUBLISH channel message`. Returns the count of subscribers
404    /// that received the message.
405    ///
406    /// As of v1.3.0, the embedded backend has a real in-process pub/sub
407    /// bus: when a [`Subscriber`] is open against the same `mem://<name>`
408    /// or `file:///path` URL, this delivers there and returns the actual
409    /// receiver count. Anonymous `mem://` keeps the old "no subscribers,
410    /// returns 0" behaviour (the URL is its own bus, by design).
411    ///
412    /// The pub/sub *consumer* side lives in [`Subscriber`]. On the remote
413    /// backend a subscribed TCP connection cannot send normal commands
414    /// per the RESP spec; the embedded backend has no such restriction
415    /// but `Subscriber` is still a distinct type for API symmetry.
416    pub fn publish(&mut self, channel: &[u8], message: &[u8]) -> io::Result<usize> {
417        match self {
418            Self::Embedded(s) => Ok(s.publish(channel, message)),
419            Self::Remote(c) => match c.request_borrowed(&[b"PUBLISH", channel, message])? {
420                Reply::Int(n) if n >= 0 => Ok(n as usize),
421                Reply::Error(e) => Err(io::Error::other(string(e))),
422                other => Err(unexpected(other)),
423            },
424        }
425    }
426}
427
428
429#[cfg(test)]
430#[path = "lib_tests.rs"]
431mod tests;
kevy_client/lib.rs

kevy_client/
lib.rs
