kevy_embedded/

ops_p2.rs

//! 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 crate::KevyResult;

use kevy_store::ScoreBound;

use crate::store::ensure_writable;
use crate::store::{Store, commit_write, store_err};

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]) -> KevyResult<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]) -> KevyResult<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]) -> KevyResult<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]) -> KevyResult<Vec<Vec<u8>>> {
        self.wshard(key).store.hkeys(key).map_err(store_err)
    }

    /// `HRANDFIELD key count` — random field names, distinct for a
    /// positive count and with repeats allowed for a negative one.
    ///
    /// ```
    /// let s = kevy_embedded::Store::open(kevy_embedded::Config::default())?;
    /// s.hset(b"h", &[(b"f1", b"v1"), (b"f2", b"v2")])?;
    /// assert_eq!(s.hrandfield(b"h", 5)?.len(), 2);
    /// assert_eq!(s.hrandfield(b"h", -5)?.len(), 5);
    /// # Ok::<(), kevy_embedded::KevyError>(())
    /// ```
    pub fn hrandfield(&self, key: &[u8], count: i64) -> KevyResult<Vec<Vec<u8>>> {
        self.wshard(key).store.hrandfield(key, count).map_err(store_err)
    }

    /// `HRANDFIELD key count WITHVALUES` — [`Self::hrandfield`] with each
    /// field paired with its value.
    ///
    /// ```
    /// let s = kevy_embedded::Store::open(kevy_embedded::Config::default())?;
    /// s.hset(b"h", &[(b"f", b"v")])?;
    /// assert_eq!(s.hrandfield_with_values(b"h", 1)?, vec![(b"f".to_vec(), b"v".to_vec())]);
    /// # Ok::<(), kevy_embedded::KevyError>(())
    /// ```
    pub fn hrandfield_with_values(
        &self,
        key: &[u8],
        count: i64,
    ) -> KevyResult<kevy_store::FieldValuePairs> {
        self.wshard(key).store.hrandfield_with_values(key, count).map_err(store_err)
    }

    /// `HVALS key` — every value in `key`'s hash.
    pub fn hvals(&self, key: &[u8]) -> KevyResult<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]]) -> KevyResult<Vec<Option<Vec<u8>>>> {
        self.wshard(key).store.hmget(key, fields).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) -> KevyResult<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) -> KevyResult<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) -> KevyResult<Vec<(Vec<u8>, f64)>> {
        self.wshard(key).store.zrevrange(key, start, stop).map_err(store_err)
    }

    /// `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,
    ) -> KevyResult<Vec<(Vec<u8>, f64)>> {
        self.wshard(key)
            .store
            .zrange_by_score(key, ScoreBound::inclusive(min), ScoreBound::inclusive(max))
            .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,
    ) -> KevyResult<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 (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,
    ) -> KevyResult<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,
    ) -> KevyResult<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())
    }

    /// `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,
    ) -> KevyResult<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]) -> KevyResult<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) -> KevyResult<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) -> KevyResult<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]) -> KevyResult<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]) -> KevyResult<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]) -> KevyResult<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)
    }
}