kevy_embedded/ops_p3.rs
1//! Multi-key string operations, keyspace scan, atomic `getex`, set
2//! algebra (`sinter` / `sunion` / `sdiff`), and absolute-time TTL
3//! variants (`expireat` / `pexpire`).
4//!
5//! The set algebra is implemented at the embedded layer (compose
6//! `smembers` per key + Rust set operations) instead of touching
7//! `kevy_store::Store` — over N small sets that is faster than
8//! serialising N RESP arrays.
9
10use crate::KevyResult;
11use std::collections::BTreeSet;
12use std::time::Duration;
13
14use crate::store::ensure_writable;
15use crate::store::{Store, commit_write, store_err};
16
17impl Store {
18 // ---- multi-key string ops ---------------------------------------
19
20 /// `MSET key value [key value ...]` — set every pair atomically
21 /// per-key. Each pair is logged independently to its shard's
22 /// AOF (no cross-shard atomic guarantee — a crash mid-call may
23 /// leave a prefix applied; matches Redis Cluster semantics).
24 pub fn mset(&self, pairs: &[(&[u8], &[u8])]) -> KevyResult<()> {
25 ensure_writable(self)?;
26 for (k, v) in pairs {
27 let mut g = self.wshard(k);
28 g.store.set(k, v.to_vec(), None, false, false);
29 commit_write(&mut g, &[b"SET", k, v])?;
30 }
31 Ok(())
32 }
33
34 /// `MGET key [key ...]` — return `Some(value)` per requested key
35 /// that's present, `None` per absent / wrong-type.
36 pub fn mget(&self, keys: &[&[u8]]) -> KevyResult<Vec<Option<Vec<u8>>>> {
37 let mut out = Vec::with_capacity(keys.len());
38 for k in keys {
39 out.push(
40 self.wshard(k)
41 .store
42 .get(k)
43 .map_err(store_err)?
44 .as_deref()
45 .map(<[u8]>::to_vec),
46 );
47 }
48 Ok(out)
49 }
50
51 // ---- keyspace introspection -------------------------------------
52
53 /// `KEYS pattern` — glob-match every key in the keyspace
54 /// (across all shards). `pattern = None` matches everything.
55 /// `limit = None` is unbounded; otherwise bounds the TOTAL
56 /// returned across shards. Glob syntax matches Redis (`*` /
57 /// `?` / `[abc]` / escape).
58 pub fn keys(&self, pattern: Option<&[u8]>, limit: Option<usize>) -> Vec<Vec<u8>> {
59 self.collect_keys(pattern, limit)
60 }
61
62 // ---- atomic get + TTL -------------------------------------------
63
64 /// `GETEX key TTL` — get the value and update the TTL atomically
65 /// (single lock cycle on the owning shard). Returns the value;
66 /// `None` when absent. AOF-logged as `PEXPIRE`.
67 pub fn getex(&self, key: &[u8], ttl: Duration) -> KevyResult<Option<Vec<u8>>> {
68 ensure_writable(self)?;
69 let mut g = self.wshard(key);
70 let val = g.store.get(key).map_err(store_err)?.as_deref().map(<[u8]>::to_vec);
71 if val.is_some() {
72 g.store.expire(key, ttl);
73 let ttl_ms = ttl.as_millis().min(i64::MAX as u128) as i64;
74 let ttl_str = format!("{ttl_ms}");
75 commit_write(&mut g, &[b"PEXPIRE", key, ttl_str.as_bytes()])?;
76 }
77 Ok(val)
78 }
79
80 // ---- set algebra (compose-side, not Store-side) ------------------
81
82 /// `SINTER key [key ...]` — set intersection. Reads each key's
83 /// members, computes the intersection in BTreeSet order
84 /// (sorted, no duplicates).
85 pub fn sinter(&self, keys: &[&[u8]]) -> KevyResult<Vec<Vec<u8>>> {
86 if keys.is_empty() {
87 return Ok(Vec::new());
88 }
89 let first: BTreeSet<Vec<u8>> = self
90 .smembers(keys[0])?
91 .into_iter()
92 .collect();
93 let mut acc = first;
94 for k in &keys[1..] {
95 if acc.is_empty() {
96 break;
97 }
98 let next: BTreeSet<Vec<u8>> = self.smembers(k)?.into_iter().collect();
99 acc.retain(|m| next.contains(m));
100 }
101 Ok(acc.into_iter().collect())
102 }
103
104 /// `SUNION key [key ...]` — set union over N sets.
105 pub fn sunion(&self, keys: &[&[u8]]) -> KevyResult<Vec<Vec<u8>>> {
106 let mut acc: BTreeSet<Vec<u8>> = BTreeSet::new();
107 for k in keys {
108 for m in self.smembers(k)? {
109 acc.insert(m);
110 }
111 }
112 Ok(acc.into_iter().collect())
113 }
114
115 /// `SDIFF key [key ...]` — `keys[0]` minus the union of every
116 /// subsequent set.
117 pub fn sdiff(&self, keys: &[&[u8]]) -> KevyResult<Vec<Vec<u8>>> {
118 if keys.is_empty() {
119 return Ok(Vec::new());
120 }
121 let mut acc: BTreeSet<Vec<u8>> = self
122 .smembers(keys[0])?
123 .into_iter()
124 .collect();
125 for k in &keys[1..] {
126 let next: BTreeSet<Vec<u8>> = self.smembers(k)?.into_iter().collect();
127 acc.retain(|m| !next.contains(m));
128 }
129 Ok(acc.into_iter().collect())
130 }
131
132 // ---- absolute-time TTL variants ----------------------------------
133
134 /// `EXPIREAT key unix_secs` — schedule expiry for the given
135 /// absolute UNIX wall-clock time. Returns `true` when the key
136 /// existed and the deadline was set; `false` when absent.
137 pub fn expireat(&self, key: &[u8], unix_secs: u64) -> KevyResult<bool> {
138 ensure_writable(self)?;
139 let mut g = self.wshard(key);
140 let unix_ms = unix_secs.saturating_mul(1000);
141 let ok = g.store.expire_at_unix_ms(key, unix_ms);
142 if ok {
143 let ts_str = format!("{unix_ms}");
144 commit_write(&mut g, &[b"PEXPIREAT", key, ts_str.as_bytes()])?;
145 }
146 Ok(ok)
147 }
148
149 /// `PEXPIREAT key unix_ms` — same as `expireat` but in
150 /// milliseconds.
151 pub fn pexpireat(&self, key: &[u8], unix_ms: u64) -> KevyResult<bool> {
152 ensure_writable(self)?;
153 let mut g = self.wshard(key);
154 let ok = g.store.expire_at_unix_ms(key, unix_ms);
155 if ok {
156 let ts_str = format!("{unix_ms}");
157 commit_write(&mut g, &[b"PEXPIREAT", key, ts_str.as_bytes()])?;
158 }
159 Ok(ok)
160 }
161
162 /// `PEXPIRE key ms` — relative TTL in milliseconds. (`expire`
163 /// takes `Duration`; this is the integer-ms variant matching
164 /// the Redis wire command.)
165 pub fn pexpire(&self, key: &[u8], ms: u64) -> KevyResult<bool> {
166 self.expire(key, Duration::from_millis(ms))
167 }
168
169 // ---- hash float increment ----------------------------------------
170
171 /// `HINCRBYFLOAT key field delta` — atomic float increment of a
172 /// hash field. Returns the post-increment value. Errors on
173 /// `NotFloat` when the field is present but not parseable.
174 pub fn hincrbyfloat(
175 &self,
176 key: &[u8],
177 field: &[u8],
178 delta: f64,
179 ) -> KevyResult<f64> {
180 ensure_writable(self)?;
181 let mut g = self.wshard(key);
182 let new_val = g
183 .store
184 .hincrbyfloat(key, field, delta)
185 .map_err(store_err)?;
186 let delta_str = format!("{delta}");
187 commit_write(&mut g, &[b"HINCRBYFLOAT", key, field, delta_str.as_bytes()])?;
188 Ok(new_val)
189 }
190
191 // ---- list positional insert --------------------------------------
192
193 /// `LINSERT key BEFORE|AFTER pivot value` — insert `value` before
194 /// or after the first occurrence of `pivot` in the list. Returns:
195 /// - `Ok(new_len)` on success (`>= 1`);
196 /// - `Ok(0)` when `key` does not exist;
197 /// - `Ok(-1)` when `pivot` was not found in the list.
198 ///
199 /// `before = true` matches Redis `LINSERT … BEFORE`, `false`
200 /// matches `LINSERT … AFTER`.
201 pub fn linsert(
202 &self,
203 key: &[u8],
204 before: bool,
205 pivot: &[u8],
206 value: &[u8],
207 ) -> KevyResult<i64> {
208 ensure_writable(self)?;
209 let mut g = self.wshard(key);
210 let new_len = g
211 .store
212 .linsert(key, before, pivot, value)
213 .map_err(store_err)?;
214 if new_len > 0 {
215 let dir = if before { b"BEFORE".as_slice() } else { b"AFTER".as_slice() };
216 commit_write(&mut g, &[b"LINSERT", key, dir, pivot, value])?;
217 }
218 Ok(new_len)
219 }
220
221 // ---- observability ----------------------------------------------
222
223 /// `Store::ping_us()` — return the round-trip duration of a
224 /// shard-0 read-lock acquire + release in **nanoseconds**, for
225 /// perfgate observability. Always returns immediately; the
226 /// duration reflects current shard-0 contention (= shorter when
227 /// idle, longer when many readers/writers compete).
228 pub fn ping_ns(&self) -> u128 {
229 let t = std::time::Instant::now();
230 let _g = self.lock();
231 t.elapsed().as_nanos()
232 }
233}