kevy_embedded/
ops_p2.rs

1//! Hash field reads, sorted-set range queries, list slice access, and
2//! the atomic single-call string helpers `getset` / `getdel`.
3//!
4//! Every method is a thin facade over the corresponding
5//! `kevy_store::Store` method, with `commit_write` AOF logging on the
6//! write paths.
7
8use std::io;
9
10use kevy_store::ScoreBound;
11
12#[cfg(not(target_arch = "wasm32"))]
13use crate::replica_glue::ensure_writable;
14use crate::store::{Store, commit_write, store_err};
15
16#[cfg(target_arch = "wasm32")]
17fn ensure_writable(_s: &Store) -> io::Result<()> { Ok(()) }
18
19impl Store {
20    // ---- hash mass-getters --------------------------------------------
21
22    /// `HGETALL key` — every `(field, value)` pair in `key`'s hash, in
23    /// arbitrary order. Empty when `key` is absent. Errors on wrong type.
24    pub fn hgetall(&self, key: &[u8]) -> io::Result<Vec<(Vec<u8>, Vec<u8>)>> {
25        let flat = self.wshard(key).store.hgetall(key).map_err(store_err)?;
26        // kevy-store returns [f0, v0, f1, v1, ...] — pair them up.
27        let mut out = Vec::with_capacity(flat.len() / 2);
28        let mut it = flat.into_iter();
29        while let (Some(f), Some(v)) = (it.next(), it.next()) {
30            out.push((f, v));
31        }
32        Ok(out)
33    }
34
35    /// `HEXISTS key field` — `true` when `field` is present.
36    pub fn hexists(&self, key: &[u8], field: &[u8]) -> io::Result<bool> {
37        self.wshard(key).store.hexists(key, field).map_err(store_err)
38    }
39
40    /// `HLEN key` — number of fields; 0 when absent.
41    pub fn hlen(&self, key: &[u8]) -> io::Result<usize> {
42        self.wshard(key).store.hlen(key).map_err(store_err)
43    }
44
45    /// `HKEYS key` — every field name in `key`'s hash.
46    pub fn hkeys(&self, key: &[u8]) -> io::Result<Vec<Vec<u8>>> {
47        self.wshard(key).store.hkeys(key).map_err(store_err)
48    }
49
50    /// `HVALS key` — every value in `key`'s hash.
51    pub fn hvals(&self, key: &[u8]) -> io::Result<Vec<Vec<u8>>> {
52        self.wshard(key).store.hvals(key).map_err(store_err)
53    }
54
55    /// `HMGET key field [field ...]` — read multiple fields in one
56    /// call. `None` per requested field that is absent.
57    pub fn hmget(&self, key: &[u8], fields: &[&[u8]]) -> io::Result<Vec<Option<Vec<u8>>>> {
58        let owned: Vec<Vec<u8>> = fields.iter().map(|f| f.to_vec()).collect();
59        self.wshard(key).store.hmget(key, &owned).map_err(store_err)
60    }
61
62    /// `HINCRBY key field delta` — atomic integer increment of a hash
63    /// field. Returns the post-increment value.
64    pub fn hincrby(&self, key: &[u8], field: &[u8], delta: i64) -> io::Result<i64> {
65        ensure_writable(self)?;
66        let mut g = self.wshard(key);
67        let new_val = g.store.hincrby(key, field, delta).map_err(store_err)?;
68        let delta_str = format!("{delta}");
69        commit_write(&mut g, &[b"HINCRBY", key, field, delta_str.as_bytes()])?;
70        Ok(new_val)
71    }
72
73    // ---- zset mass-readers + atomic incr -----------------------------
74
75    /// `ZRANGE key start stop WITHSCORES` — members in ascending score
76    /// order between rank `start..=stop` (Redis-style inclusive
77    /// indexing; negatives count from the tail). Returns `(member,
78    /// score)` pairs.
79    pub fn zrange(&self, key: &[u8], start: i64, stop: i64) -> io::Result<Vec<(Vec<u8>, f64)>> {
80        self.wshard(key).store.zrange(key, start, stop).map_err(store_err)
81    }
82
83    /// `ZREVRANGE key start stop WITHSCORES` — `zrange` with the order
84    /// reversed (highest score first). The `start..=stop` indexing is
85    /// against the reversed list, matching Redis semantics.
86    pub fn zrevrange(
87        &self,
88        key: &[u8],
89        start: i64,
90        stop: i64,
91    ) -> io::Result<Vec<(Vec<u8>, f64)>> {
92        let mut all = self
93            .wshard(key)
94            .store
95            .zrange(key, 0, -1)
96            .map_err(store_err)?;
97        all.reverse();
98        let n = all.len() as i64;
99        if n == 0 {
100            return Ok(Vec::new());
101        }
102        let clamp = |x: i64| -> usize {
103            let v = if x < 0 { (n + x).max(0) } else { x.min(n - 1) };
104            v as usize
105        };
106        let s = clamp(start);
107        let e = clamp(stop);
108        if s > e {
109            return Ok(Vec::new());
110        }
111        Ok(all.into_iter().skip(s).take(e - s + 1).collect())
112    }
113
114    /// `ZRANGEBYSCORE` — score-range read. `min` / `max` are
115    /// inclusive; pass `f64::NEG_INFINITY` / `f64::INFINITY` for open
116    /// bounds. Returns `(member, score)` pairs in ascending score
117    /// order. Exclusive bounds are `ZRANGEBYSCORE (` syntax in Redis;
118    /// expose via the dedicated [`Self::zrange_by_score_excl`].
119    pub fn zrange_by_score(
120        &self,
121        key: &[u8],
122        min: f64,
123        max: f64,
124    ) -> io::Result<Vec<(Vec<u8>, f64)>> {
125        self.wshard(key)
126            .store
127            .zrange_by_score(
128                key,
129                ScoreBound { value: min, exclusive: false },
130                ScoreBound { value: max, exclusive: false },
131            )
132            .map_err(store_err)
133    }
134
135    /// Same as [`Self::zrange_by_score`] but with explicit
136    /// inclusive/exclusive control on each bound (`(min` / `(max` in
137    /// Redis syntax).
138    pub fn zrange_by_score_excl(
139        &self,
140        key: &[u8],
141        min: ScoreBound,
142        max: ScoreBound,
143    ) -> io::Result<Vec<(Vec<u8>, f64)>> {
144        self.wshard(key)
145            .store
146            .zrange_by_score(key, min, max)
147            .map_err(store_err)
148    }
149
150    /// `ZRANGEBYSCORE key min max LIMIT offset count` — score-range
151    /// read with pagination (v2.2; closes the embedded LIMIT gap —
152    /// the server parser always had it).
153    pub fn zrange_by_score_limit(
154        &self,
155        key: &[u8],
156        min: f64,
157        max: f64,
158        offset: usize,
159        count: usize,
160    ) -> io::Result<Vec<(Vec<u8>, f64)>> {
161        let all = self.zrange_by_score(key, min, max)?;
162        Ok(all.into_iter().skip(offset).take(count).collect())
163    }
164
165    /// `ZREVRANGEBYSCORE key max min LIMIT offset count` — descending
166    /// score-range read with pagination.
167    pub fn zrevrange_by_score_limit(
168        &self,
169        key: &[u8],
170        max: f64,
171        min: f64,
172        offset: usize,
173        count: usize,
174    ) -> io::Result<Vec<(Vec<u8>, f64)>> {
175        let mut all = self.zrange_by_score(key, min, max)?;
176        all.reverse();
177        Ok(all.into_iter().skip(offset).take(count).collect())
178    }
179
180    /// v2.4 `zpopmin_below` — pop up to `count` lowest members with
181    /// score strictly `< below` (delayed-job "pop what's due").
182    /// AOF logs the effect (`ZREM` of the popped members).
183    pub fn zpopmin_below(
184        &self,
185        key: &[u8],
186        below: f64,
187        count: usize,
188    ) -> io::Result<Vec<(Vec<u8>, f64)>> {
189        ensure_writable(self)?;
190        let mut g = self.wshard(key);
191        let items = g.store.zpopmin_below(key, below, count).map_err(store_err)?;
192        if !items.is_empty() {
193            let mut argv: Vec<&[u8]> = Vec::with_capacity(2 + items.len());
194            argv.push(b"ZREM");
195            argv.push(key);
196            argv.extend(items.iter().map(|(m, _)| m.as_slice()));
197            commit_write(&mut g, &argv)?;
198        }
199        Ok(items)
200    }
201
202    /// `ZINCRBY key delta member` — atomic float increment of a member's
203    /// score. Returns the post-increment score.
204    pub fn zincrby(&self, key: &[u8], delta: f64, member: &[u8]) -> io::Result<f64> {
205        ensure_writable(self)?;
206        let mut g = self.wshard(key);
207        let new_score = g.store.zincrby(key, delta, member).map_err(store_err)?;
208        let delta_str = format!("{delta}");
209        commit_write(&mut g, &[b"ZINCRBY", key, delta_str.as_bytes(), member])?;
210        Ok(new_score)
211    }
212
213    // ---- list slice + index ops --------------------------------------
214
215    /// `LRANGE key start stop` — list slice. Negative indices count
216    /// from the tail. Empty when absent.
217    pub fn lrange(&self, key: &[u8], start: i64, stop: i64) -> io::Result<Vec<Vec<u8>>> {
218        self.wshard(key).store.lrange(key, start, stop).map_err(store_err)
219    }
220
221    /// `LINDEX key idx` — element at index `idx`; `None` out of range.
222    pub fn lindex(&self, key: &[u8], idx: i64) -> io::Result<Option<Vec<u8>>> {
223        self.wshard(key).store.lindex(key, idx).map_err(store_err)
224    }
225
226    /// `LREM key count value` — remove up to `|count|` occurrences of
227    /// `value`. `count > 0` from head, `count < 0` from tail,
228    /// `count == 0` all. Returns the count actually removed.
229    pub fn lrem(&self, key: &[u8], count: i64, value: &[u8]) -> io::Result<usize> {
230        ensure_writable(self)?;
231        let mut g = self.wshard(key);
232        let removed = g.store.lrem(key, count, value).map_err(store_err)?;
233        if removed > 0 {
234            let count_str = format!("{count}");
235            commit_write(&mut g, &[b"LREM", key, count_str.as_bytes(), value])?;
236        }
237        Ok(removed)
238    }
239
240    // ---- string single-call atomic patterns --------------------------
241
242    /// `GETSET key new` — set `key` to `new`, return the previous
243    /// value (or `None` when `key` was absent).
244    pub fn getset(&self, key: &[u8], new: &[u8]) -> io::Result<Option<Vec<u8>>> {
245        ensure_writable(self)?;
246        let mut g = self.wshard(key);
247        let prev = g.store.getset(key, new.to_vec()).map_err(store_err)?;
248        commit_write(&mut g, &[b"SET", key, new])?;
249        Ok(prev)
250    }
251
252    /// `GETDEL key` — delete `key`, return the previous value
253    /// (`None` when `key` was absent).
254    pub fn getdel(&self, key: &[u8]) -> io::Result<Option<Vec<u8>>> {
255        ensure_writable(self)?;
256        let mut g = self.wshard(key);
257        let prev = g.store.getdel(key).map_err(store_err)?;
258        if prev.is_some() {
259            commit_write(&mut g, &[b"DEL", key])?;
260        }
261        Ok(prev)
262    }
263}
kevy_embedded/ops_p2.rs

kevy_embedded/
ops_p2.rs
