kevy_embedded/
ops.rs

1//! Data-type methods on [`Store`] — string, hash, list, set, sorted set,
2//! plus the pub/sub `publish` / `subscribe` / `psubscribe` entry points.
3//!
4//! All of these are thin facades over `kevy_store::Store` (the keyspace)
5//! and `pubsub::PubsubBus` (the in-process bus); they hold the embedded
6//! mutex for the duration of the underlying call, then drop it. AOF
7//! logging + post-write eviction sweep run via `commit_write` from
8//! `store.rs`. Behaviour and ABI are unchanged from the v1.1.0 single-file
9//! layout — this module only exists to keep `store.rs` under the 500-LOC
10//! cap.
11
12use std::io;
13use std::time::Duration;
14
15use kevy_store::StoreError;
16
17use crate::pubsub::Subscription;
18use crate::replica_glue::ensure_writable;
19use crate::store::{Store, commit_write, store_err};
20
21impl Store {
22    // ---- string ops -----------------------------------------------------
23
24    /// `SET key value` (no TTL, no NX/XX). Returns `true` always under the
25    /// embedded API (Redis semantics: SET overwrites; NX/XX vetoes would
26    /// return `false` but we don't expose those here — use [`Store::with`]
27    /// for the full surface).
28    pub fn set(&self, key: &[u8], value: &[u8]) -> io::Result<bool> {
29        ensure_writable(self)?;
30        let mut g = self.wshard(key);
31        let ok = g.store.set(key, value.to_vec(), None, false, false);
32        commit_write(&mut g, &[b"SET", key, value])?;
33        Ok(ok)
34    }
35
36    /// `SET key value PX ms` — overwrites + sets TTL. The AOF records an
37    /// **absolute** `PEXPIREAT` deadline (not the relative `ttl`) so the key
38    /// expires at the same wall-clock instant after a restart — a relative
39    /// `PEXPIRE` would be re-anchored to replay-time, resetting the TTL to a
40    /// fresh full duration on every restart (INC-2026-06-09).
41    pub fn set_with_ttl(&self, key: &[u8], value: &[u8], ttl: Duration) -> io::Result<bool> {
42        ensure_writable(self)?;
43        let mut g = self.wshard(key);
44        let ok = g.store.set(key, value.to_vec(), Some(ttl), false, false);
45        let ms = ttl.as_millis().min(u128::from(u64::MAX)) as u64;
46        let deadline = kevy_store::now_unix_ms().saturating_add(ms);
47        commit_write(&mut g, &[b"SET", key, value])?;
48        commit_write(&mut g, &[b"PEXPIREAT", key, deadline.to_string().as_bytes()])?;
49        Ok(ok)
50    }
51
52    /// `GET key` — `Some(bytes)` on hit, `None` on miss or expired.
53    ///
54    /// With eviction off (`maxmemory == 0`, the default) this takes the **read**
55    /// lock and a non-mutating store lookup, so concurrent readers scale across
56    /// cores — the path a read-heavy embed cache lives on. With eviction on it
57    /// falls back to the exclusive lock + mutating get so each access still
58    /// stamps the LRU clock.
59    pub fn get(&self, key: &[u8]) -> io::Result<Option<Vec<u8>>> {
60        if self.config().maxmemory == 0 {
61            let g = self.rshard(key);
62            return Ok(g.store.get_shared(key).map_err(store_err)?.map(|c| c.into_owned()));
63        }
64        let mut g = self.wshard(key);
65        Ok(g.store.get(key).map_err(store_err)?.map(|c| c.into_owned()))
66    }
67
68    /// `DEL key1 [key2 ...]`. Returns the count of keys actually removed.
69    /// Keys fan out to their owning shards.
70    pub fn del(&self, keys: &[&[u8]]) -> io::Result<usize> {
71        ensure_writable(self)?;
72        let mut total = 0;
73        for k in keys {
74            let owned = vec![k.to_vec()];
75            let mut g = self.wshard(k);
76            let n = g.store.del(&owned);
77            if n > 0 {
78                total += n;
79                commit_write(&mut g, &[b"DEL", k])?;
80            }
81        }
82        Ok(total)
83    }
84
85    /// `EXISTS key1 [key2 ...]`. Count of existing keys (duplicates counted
86    /// multiple times, matching Redis).
87    pub fn exists(&self, keys: &[&[u8]]) -> io::Result<usize> {
88        let mut total = 0;
89        for k in keys {
90            total += self.wshard(k).store.exists(&[k.to_vec()]);
91        }
92        Ok(total)
93    }
94
95    /// `INCR key`. Returns the post-increment value.
96    pub fn incr(&self, key: &[u8]) -> io::Result<i64> {
97        self.incr_by(key, 1)
98    }
99
100    /// `INCRBY key delta`. Negative `delta` does DECR-style work.
101    pub fn incr_by(&self, key: &[u8], delta: i64) -> io::Result<i64> {
102        ensure_writable(self)?;
103        let mut g = self.wshard(key);
104        let n = g.store.incr_by(key, delta).map_err(store_err)?;
105        commit_write(&mut g, &[b"INCRBY", key, delta.to_string().as_bytes()])?;
106        Ok(n)
107    }
108
109    /// `EXPIRE key seconds`. Returns `true` if a key was touched. The AOF
110    /// records an absolute `PEXPIREAT` deadline (see [`Self::set_with_ttl`])
111    /// so the TTL survives a restart unchanged.
112    pub fn expire(&self, key: &[u8], ttl: Duration) -> io::Result<bool> {
113        ensure_writable(self)?;
114        let mut g = self.wshard(key);
115        let touched = g.store.expire(key, ttl);
116        if touched {
117            let ms = ttl.as_millis().min(u128::from(u64::MAX)) as u64;
118            let deadline = kevy_store::now_unix_ms().saturating_add(ms);
119            commit_write(&mut g, &[b"PEXPIREAT", key, deadline.to_string().as_bytes()])?;
120        }
121        Ok(touched)
122    }
123
124    /// `PERSIST key`. Returns `true` if a TTL was actually cleared.
125    pub fn persist(&self, key: &[u8]) -> io::Result<bool> {
126        ensure_writable(self)?;
127        let mut g = self.wshard(key);
128        let touched = g.store.persist(key);
129        if touched {
130            commit_write(&mut g, &[b"PERSIST", key])?;
131        }
132        Ok(touched)
133    }
134
135    /// Remaining TTL in ms (or Redis-style `-1`/`-2` for no-TTL/no-key).
136    pub fn ttl_ms(&self, key: &[u8]) -> i64 {
137        self.wshard(key).store.pttl(key)
138    }
139
140    /// `TYPE key` — `"string"`, `"hash"`, `"list"`, `"set"`, `"zset"`, or `"none"`.
141    pub fn type_of(&self, key: &[u8]) -> &'static str {
142        self.wshard(key).store.type_of(key)
143    }
144
145    /// `DBSIZE` — total live keys across all shards.
146    pub fn dbsize(&self) -> usize {
147        self.sum_shards(|i| i.store.dbsize())
148    }
149
150    /// `FLUSHALL` — empty every shard (each logs `FLUSHALL` so a replay reaches
151    /// the same empty state).
152    ///
153    /// Named `flushall` — **not** `flush` — to avoid colliding with
154    /// `Write::flush`'s "sync buffered writes to disk" meaning. This call
155    /// WIPES the store; durability needs no explicit call (each write appends
156    /// to the AOF, the shard's `BufWriter` lands per [`AppendFsync`] cadence
157    /// and on drop).
158    ///
159    /// [`AppendFsync`]: crate::AppendFsync
160    pub fn flushall(&self) -> io::Result<()> {
161        ensure_writable(self)?;
162        self.try_for_each_shard(|inner| {
163            inner.store.flushall();
164            commit_write(inner, &[b"FLUSHALL"])
165        })
166    }
167
168    /// Deprecated alias for [`Self::flushall`]. The old name read like
169    /// `Write::flush` (sync-to-disk) but actually WIPES the store — a
170    /// data-loss footgun.
171    #[deprecated(
172        since = "1.2.0",
173        note = "renamed to `flushall`: `flush` collides with Write::flush (sync-to-disk); this WIPES the store"
174    )]
175    pub fn flush(&self) -> io::Result<()> {
176        self.flushall()
177    }
178
179    /// `MEMORY USAGE` for one key — `Some(bytes)` or `None` if absent.
180    pub fn key_bytes(&self, key: &[u8]) -> Option<u64> {
181        self.wshard(key).store.estimate_key_bytes(key)
182    }
183
184    /// Live `used_memory` estimate (summed across shards).
185    pub fn used_memory(&self) -> u64 {
186        self.sum_shards_u64(|i| i.store.used_memory())
187    }
188
189    /// `INFO`-style counter: total keys evicted by `maxmemory` (all shards).
190    pub fn evictions_total(&self) -> u64 {
191        self.sum_shards_u64(|i| i.store.evictions_total())
192    }
193
194    /// `INFO`-style counter: total keys expired (lazy + active reaper, all shards).
195    pub fn expired_keys_total(&self) -> u64 {
196        self.sum_shards_u64(|i| i.store.expired_keys_total())
197    }
198
199    // ---- hash ops -------------------------------------------------------
200
201    /// `HSET key field value [field value ...]`. Returns count newly added.
202    pub fn hset(&self, key: &[u8], pairs: &[(&[u8], &[u8])]) -> io::Result<usize> {
203        ensure_writable(self)?;
204        let mut g = self.wshard(key);
205        let owned: Vec<(Vec<u8>, Vec<u8>)> =
206            pairs.iter().map(|(f, v)| (f.to_vec(), v.to_vec())).collect();
207        let added = g.store.hset(key, &owned).map_err(store_err)?;
208        let mut parts: Vec<&[u8]> = Vec::with_capacity(2 + pairs.len() * 2);
209        parts.push(b"HSET");
210        parts.push(key);
211        for (f, v) in pairs {
212            parts.push(f);
213            parts.push(v);
214        }
215        commit_write(&mut g, &parts)?;
216        Ok(added)
217    }
218
219    /// `HGET key field`. `None` if absent.
220    pub fn hget(&self, key: &[u8], field: &[u8]) -> io::Result<Option<Vec<u8>>> {
221        let mut g = self.wshard(key);
222        Ok(g.store
223            .hget(key, field)
224            .map_err(store_err)?
225            .map(<[u8]>::to_vec))
226    }
227
228    /// `HDEL key field [field ...]`. Returns count actually removed.
229    pub fn hdel(&self, key: &[u8], fields: &[&[u8]]) -> io::Result<usize> {
230        ensure_writable(self)?;
231        let mut g = self.wshard(key);
232        let owned: Vec<Vec<u8>> = fields.iter().map(|f| f.to_vec()).collect();
233        let removed = g.store.hdel(key, &owned).map_err(store_err)?;
234        if removed > 0 {
235            let mut parts: Vec<&[u8]> = Vec::with_capacity(2 + fields.len());
236            parts.push(b"HDEL");
237            parts.push(key);
238            for f in fields {
239                parts.push(f);
240            }
241            commit_write(&mut g, &parts)?;
242        }
243        Ok(removed)
244    }
245
246    // ---- list ops -------------------------------------------------------
247
248    /// `LPUSH key value [value ...]`. Returns the new list length.
249    pub fn lpush(&self, key: &[u8], values: &[&[u8]]) -> io::Result<usize> {
250        push_helper(self, key, values, b"LPUSH", kevy_store::Store::lpush)
251    }
252
253    /// `RPUSH key value [value ...]`. Returns the new list length.
254    pub fn rpush(&self, key: &[u8], values: &[&[u8]]) -> io::Result<usize> {
255        push_helper(self, key, values, b"RPUSH", kevy_store::Store::rpush)
256    }
257
258    /// `LPOP key count`. Returns popped values from the head.
259    pub fn lpop(&self, key: &[u8], count: usize) -> io::Result<Vec<Vec<u8>>> {
260        pop_helper(self, key, count, false)
261    }
262
263    /// `RPOP key count`. Symmetric to `LPOP` from the tail.
264    pub fn rpop(&self, key: &[u8], count: usize) -> io::Result<Vec<Vec<u8>>> {
265        pop_helper(self, key, count, true)
266    }
267
268    /// `LLEN key`. Length of the list at `key`; 0 if absent.
269    pub fn llen(&self, key: &[u8]) -> io::Result<usize> {
270        self.wshard(key).store.llen(key).map_err(store_err)
271    }
272
273    // ---- set ops --------------------------------------------------------
274
275    /// `SADD key member [member ...]`. Returns count newly added.
276    pub fn sadd(&self, key: &[u8], members: &[&[u8]]) -> io::Result<usize> {
277        push_helper(self, key, members, b"SADD", kevy_store::Store::sadd)
278    }
279
280    /// `SREM key member [member ...]`. Returns count actually removed.
281    pub fn srem(&self, key: &[u8], members: &[&[u8]]) -> io::Result<usize> {
282        ensure_writable(self)?;
283        let mut g = self.wshard(key);
284        let owned: Vec<Vec<u8>> = members.iter().map(|m| m.to_vec()).collect();
285        let removed = g.store.srem(key, &owned).map_err(store_err)?;
286        if removed > 0 {
287            let mut parts: Vec<&[u8]> = Vec::with_capacity(2 + members.len());
288            parts.push(b"SREM");
289            parts.push(key);
290            for m in members {
291                parts.push(m);
292            }
293            commit_write(&mut g, &parts)?;
294        }
295        Ok(removed)
296    }
297
298    /// `SMEMBERS key`. Order implementation-defined; empty if absent.
299    pub fn smembers(&self, key: &[u8]) -> io::Result<Vec<Vec<u8>>> {
300        self.wshard(key).store.smembers(key).map_err(store_err)
301    }
302
303    /// `SCARD key`. Member count; 0 if absent.
304    pub fn scard(&self, key: &[u8]) -> io::Result<usize> {
305        self.wshard(key).store.scard(key).map_err(store_err)
306    }
307
308    // ---- zset ops -------------------------------------------------------
309
310    /// `ZADD key score member [score member ...]`. Returns count newly added.
311    pub fn zadd(&self, key: &[u8], pairs: &[(f64, &[u8])]) -> io::Result<usize> {
312        ensure_writable(self)?;
313        let mut g = self.wshard(key);
314        let owned: Vec<(f64, Vec<u8>)> =
315            pairs.iter().map(|(s, m)| (*s, m.to_vec())).collect();
316        let added = g.store.zadd(key, &owned).map_err(store_err)?;
317        let mut score_strs: Vec<Vec<u8>> = Vec::with_capacity(pairs.len());
318        for (s, _) in pairs {
319            score_strs.push(format!("{s}").into_bytes());
320        }
321        let mut parts: Vec<&[u8]> = Vec::with_capacity(2 + pairs.len() * 2);
322        parts.push(b"ZADD");
323        parts.push(key);
324        for (i, (_, m)) in pairs.iter().enumerate() {
325            parts.push(&score_strs[i]);
326            parts.push(m);
327        }
328        commit_write(&mut g, &parts)?;
329        Ok(added)
330    }
331
332    /// `ZREM key member [member ...]`. Returns count actually removed.
333    pub fn zrem(&self, key: &[u8], members: &[&[u8]]) -> io::Result<usize> {
334        ensure_writable(self)?;
335        let mut g = self.wshard(key);
336        let owned: Vec<Vec<u8>> = members.iter().map(|m| m.to_vec()).collect();
337        let removed = g.store.zrem(key, &owned).map_err(store_err)?;
338        if removed > 0 {
339            let mut parts: Vec<&[u8]> = Vec::with_capacity(2 + members.len());
340            parts.push(b"ZREM");
341            parts.push(key);
342            for m in members {
343                parts.push(m);
344            }
345            commit_write(&mut g, &parts)?;
346        }
347        Ok(removed)
348    }
349
350    /// `ZSCORE key member`. `Some(score)` if present.
351    pub fn zscore(&self, key: &[u8], member: &[u8]) -> io::Result<Option<f64>> {
352        self.wshard(key).store.zscore(key, member).map_err(store_err)
353    }
354
355    /// `ZCARD key`. Member count; 0 if absent.
356    pub fn zcard(&self, key: &[u8]) -> io::Result<usize> {
357        self.wshard(key).store.zcard(key).map_err(store_err)
358    }
359
360    // ---- pub/sub --------------------------------------------------------
361
362    /// `PUBLISH channel payload`. Delivers `payload` to every subscriber on
363    /// `channel` (direct + pattern matches) inside this process. Returns
364    /// the count of receivers the message reached.
365    pub fn publish(&self, channel: &[u8], payload: &[u8]) -> usize {
366        // Clone matching senders under the lock, then release before
367        // send() so a slow receiver can't stall unrelated traffic.
368        let plans = {
369            // Pub/sub is process-wide; the bus lives on shard 0.
370            let g = self.lock();
371            g.bus.collect_delivery(channel, payload)
372        };
373        let mut count = 0;
374        for (frame, sender) in plans {
375            if sender.send(frame).is_ok() {
376                count += 1;
377            }
378        }
379        count
380    }
381
382    /// Open a [`Subscription`] subscribed to `channels`. Drop the handle
383    /// to unsubscribe from everything atomically. Pass `&[]` to start
384    /// with no subscriptions and add some later via
385    /// [`Subscription::subscribe`] / [`Subscription::psubscribe`].
386    pub fn subscribe(&self, channels: &[&[u8]]) -> Subscription {
387        let mut sub = Subscription::new(self.inner_handle(), self.guard_handle());
388        if !channels.is_empty() {
389            sub.subscribe(channels);
390        }
391        sub
392    }
393
394    /// Convenience: open a [`Subscription`] starting on pattern subscriptions.
395    pub fn psubscribe(&self, patterns: &[&[u8]]) -> Subscription {
396        let mut sub = Subscription::new(self.inner_handle(), self.guard_handle());
397        if !patterns.is_empty() {
398            sub.psubscribe(patterns);
399        }
400        sub
401    }
402}
403
404// ─────────────────────────────────────────────────────────────────────────
405// Shared list/set push + list pop helpers. `&Store` so we can lock + AOF-log.
406// ─────────────────────────────────────────────────────────────────────────
407
408fn push_helper<F>(
409    s: &Store,
410    key: &[u8],
411    values: &[&[u8]],
412    verb: &'static [u8],
413    op: F,
414) -> io::Result<usize>
415where
416    F: FnOnce(&mut kevy_store::Store, &[u8], &[Vec<u8>]) -> Result<usize, StoreError>,
417{
418    ensure_writable(s)?;
419    let mut g = s.wshard(key);
420    let owned: Vec<Vec<u8>> = values.iter().map(|v| v.to_vec()).collect();
421    let n = op(&mut g.store, key, &owned).map_err(store_err)?;
422    let mut parts: Vec<&[u8]> = Vec::with_capacity(2 + values.len());
423    parts.push(verb);
424    parts.push(key);
425    for v in values {
426        parts.push(v);
427    }
428    commit_write(&mut g, &parts)?;
429    Ok(n)
430}
431
432fn pop_helper(s: &Store, key: &[u8], count: usize, from_tail: bool) -> io::Result<Vec<Vec<u8>>> {
433    ensure_writable(s)?;
434    let mut g = s.wshard(key);
435    let popped = if from_tail {
436        g.store.rpop(key, count).map_err(store_err)?
437    } else {
438        g.store.lpop(key, count).map_err(store_err)?
439    };
440    if !popped.is_empty() {
441        let verb: &[u8] = if from_tail { b"RPOP" } else { b"LPOP" };
442        let count_str = popped.len().to_string();
443        let parts: [&[u8]; 3] = [verb, key, count_str.as_bytes()];
444        commit_write(&mut g, &parts)?;
445    }
446    Ok(popped)
447}
kevy_embedded/ops.rs

kevy_embedded/
ops.rs
