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
//! Set membership and pop, sorted-set ranking and pop, list `lset` /
//! `ltrim`, and key rename: `sismember`, `spop`, `srandmember`,
//! `zrank`, `zcount`, `zpopmin`, `zremrangebyrank`,
//! `zremrangebyscore`, `zrev_range_by_score`, `lset`, `ltrim`,
//! `rename`, `renamenx`.
//!
//! Every method wraps a corresponding `kevy_store::Store` method.
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 {
// ---- set extras --------------------------------------------------
/// `SISMEMBER key member` — `true` when `member` is in the set.
pub fn sismember(&self, key: &[u8], member: &[u8]) -> io::Result<bool> {
self.wshard(key).store.sismember(key, member).map_err(store_err)
}
/// `SPOP key count` — atomically remove + return up to `count`
/// random members.
///
/// AOF form: logged as `SREM key <popped…>` (the members actually
/// removed), not `SPOP key count` — a random pick replayed against
/// a store whose internal layout differs (replica applying frames
/// onto snapshot-loaded state) would remove *different* members.
/// Redis propagates SPOP the same way.
pub fn spop(&self, key: &[u8], count: usize) -> io::Result<Vec<Vec<u8>>> {
ensure_writable(self)?;
let mut g = self.wshard(key);
let popped = g.store.spop(key, count).map_err(store_err)?;
if !popped.is_empty() {
let mut argv: Vec<&[u8]> = Vec::with_capacity(2 + popped.len());
argv.push(b"SREM");
argv.push(key);
argv.extend(popped.iter().map(Vec::as_slice));
commit_write(&mut g, &argv)?;
}
Ok(popped)
}
/// `SRANDMEMBER key count` — return up to `count` random members
/// without removing them.
pub fn srandmember(&self, key: &[u8], count: usize) -> io::Result<Vec<Vec<u8>>> {
self.wshard(key).store.srandmember(key, count).map_err(store_err)
}
// ---- sorted set extras ------------------------------------------
/// `ZRANK key member` — rank (0-based, ascending) of `member`;
/// `None` if not present.
pub fn zrank(&self, key: &[u8], member: &[u8]) -> io::Result<Option<usize>> {
self.wshard(key).store.zrank(key, member).map_err(store_err)
}
/// `ZCOUNT key min max` — count members whose score falls in
/// `[min, max]` (inclusive). Pass `±INFINITY` for open bounds.
pub fn zcount(&self, key: &[u8], min: f64, max: f64) -> io::Result<usize> {
self.wshard(key)
.store
.zcount(
key,
ScoreBound { value: min, exclusive: false },
ScoreBound { value: max, exclusive: false },
)
.map_err(store_err)
}
/// `ZPOPMIN key count` — atomically remove + return up to `count`
/// members with the lowest scores. Pairs are `(member, score)`.
pub fn zpopmin(&self, key: &[u8], count: usize) -> io::Result<Vec<(Vec<u8>, f64)>> {
ensure_writable(self)?;
let mut g = self.wshard(key);
let popped = g.store.zpopmin(key, count).map_err(store_err)?;
if !popped.is_empty() {
let count_str = format!("{count}");
commit_write(&mut g, &[b"ZPOPMIN", key, count_str.as_bytes()])?;
}
Ok(popped)
}
/// `ZREMRANGEBYRANK key start stop` — remove members in
/// `[start, stop]` rank range (inclusive, Redis-style negative
/// indexing). Returns count removed.
pub fn zremrangebyrank(
&self,
key: &[u8],
start: i64,
stop: i64,
) -> io::Result<usize> {
ensure_writable(self)?;
let mut g = self.wshard(key);
let removed = g.store.zrem_range_by_rank(key, start, stop).map_err(store_err)?;
if removed > 0 {
let s = format!("{start}");
let e = format!("{stop}");
commit_write(&mut g, &[b"ZREMRANGEBYRANK", key, s.as_bytes(), e.as_bytes()])?;
}
Ok(removed)
}
/// `ZREMRANGEBYSCORE key min max` — remove members with scores
/// in `[min, max]` (inclusive). Returns count removed.
pub fn zremrangebyscore(
&self,
key: &[u8],
min: f64,
max: f64,
) -> io::Result<usize> {
ensure_writable(self)?;
let mut g = self.wshard(key);
let removed = g
.store
.zrem_range_by_score(
key,
ScoreBound { value: min, exclusive: false },
ScoreBound { value: max, exclusive: false },
)
.map_err(store_err)?;
if removed > 0 {
let s = format!("{min}");
let e = format!("{max}");
commit_write(&mut g, &[b"ZREMRANGEBYSCORE", key, s.as_bytes(), e.as_bytes()])?;
}
Ok(removed)
}
/// `ZREVRANGEBYSCORE key max min` — members with scores in
/// `[min, max]` in DESCENDING score order. Inclusive bounds.
pub fn zrev_range_by_score(
&self,
key: &[u8],
max: f64,
min: f64,
) -> io::Result<Vec<(Vec<u8>, f64)>> {
self.wshard(key)
.store
.zrev_range_by_score(
key,
ScoreBound { value: min, exclusive: false },
ScoreBound { value: max, exclusive: false },
)
.map_err(store_err)
}
// ---- list extras -------------------------------------------------
/// `LSET key idx value` — set the element at `idx` (negative
/// indexes count from tail). Errors `NoSuchKey` / `OutOfRange`
/// matching Redis.
pub fn lset(&self, key: &[u8], idx: i64, value: &[u8]) -> io::Result<()> {
ensure_writable(self)?;
let mut g = self.wshard(key);
g.store.lset(key, idx, value).map_err(store_err)?;
let i = format!("{idx}");
commit_write(&mut g, &[b"LSET", key, i.as_bytes(), value])?;
Ok(())
}
/// `LTRIM key start stop` — trim list to `[start, stop]`
/// inclusive (Redis-style negative indexing).
pub fn ltrim(&self, key: &[u8], start: i64, stop: i64) -> io::Result<()> {
ensure_writable(self)?;
let mut g = self.wshard(key);
g.store.ltrim(key, start, stop).map_err(store_err)?;
let s = format!("{start}");
let e = format!("{stop}");
commit_write(&mut g, &[b"LTRIM", key, s.as_bytes(), e.as_bytes()])?;
Ok(())
}
// ---- keyspace extras --------------------------------------------
/// `RENAME src dst` — atomic rename. Returns `true` when the
/// rename happened. Errors when `src` doesn't exist (Redis would
/// reply `-ERR no such key`, here `Err(NoSuchKey)`).
pub fn rename(&self, src: &[u8], dst: &[u8]) -> io::Result<bool> {
ensure_writable(self)?;
// Cross-shard rename is non-trivial; the single-shard
// embedded default lands src+dst on the same lock.
let mut g = self.wshard(src);
let outcome = g.store.rename(src, dst, false);
match outcome {
kevy_store::RenameOutcome::Renamed => {
commit_write(&mut g, &[b"RENAME", src, dst])?;
Ok(true)
}
kevy_store::RenameOutcome::NoSuchSrc => {
Err(io::Error::other("no such key"))
}
kevy_store::RenameOutcome::DstExists => Ok(false),
}
}
/// `RENAMENX src dst` — rename only when `dst` doesn't exist.
/// Returns `true` when the rename happened.
pub fn renamenx(&self, src: &[u8], dst: &[u8]) -> io::Result<bool> {
ensure_writable(self)?;
let mut g = self.wshard(src);
let outcome = g.store.rename(src, dst, true);
match outcome {
kevy_store::RenameOutcome::Renamed => {
commit_write(&mut g, &[b"RENAMENX", src, dst])?;
Ok(true)
}
kevy_store::RenameOutcome::DstExists => Ok(false),
kevy_store::RenameOutcome::NoSuchSrc => {
Err(io::Error::other("no such key"))
}
}
}
}