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