kevy-client 1.7.1

//! Connection methods for the four collection-typed Redis data types:
//! hash, list, set, sorted set. Plus the small `LPUSH`/`SADD`-style
//! request builders shared between them.
//!
//! Lives in its own module so `lib.rs` stays focused on the `Connection`
//! enum + open + the generic + string ops. Behaviour and API are
//! unchanged from the single-file layout in v1.2.0 / v1.3.0.

use std::io;

use kevy_resp::Reply;
use kevy_resp_client::RespClient;

use crate::{Connection, array_to_bulks, store_err, string, unexpected, vec2, vec3};

impl Connection {
    // ===== Hash =====

    /// `HSET key field value [field value ...]`. Returns the number of
    /// fields that were newly added (not overwrites).
    pub fn hset(&mut self, key: &[u8], pairs: &[(&[u8], &[u8])]) -> io::Result<usize> {
        match self {
            Self::Embedded(s) => s.hset(key, pairs),
            Self::Remote(c) => {
                let mut args = Vec::with_capacity(2 + pairs.len() * 2);
                args.push(b"HSET".to_vec());
                args.push(key.to_vec());
                for (f, v) in pairs {
                    args.push(f.to_vec());
                    args.push(v.to_vec());
                }
                match c.request(&args)? {
                    Reply::Int(n) if n >= 0 => Ok(n as usize),
                    Reply::Error(e) => Err(io::Error::other(string(e))),
                    other => Err(unexpected(other)),
                }
            }
        }
    }

    /// `HGET key field`. `None` when the key or field is absent.
    pub fn hget(&mut self, key: &[u8], field: &[u8]) -> io::Result<Option<Vec<u8>>> {
        match self {
            Self::Embedded(s) => s.hget(key, field),
            Self::Remote(c) => match c.request(&vec3(b"HGET", key, field))? {
                Reply::Bulk(v) => Ok(Some(v)),
                Reply::Nil => Ok(None),
                Reply::Error(e) => Err(io::Error::other(string(e))),
                other => Err(unexpected(other)),
            },
        }
    }

    /// `HDEL key field [field ...]`. Returns the number of fields actually
    /// removed.
    pub fn hdel(&mut self, key: &[u8], fields: &[&[u8]]) -> io::Result<usize> {
        match self {
            Self::Embedded(s) => s.hdel(key, fields),
            Self::Remote(c) => {
                let mut args = Vec::with_capacity(fields.len() + 2);
                args.push(b"HDEL".to_vec());
                args.push(key.to_vec());
                args.extend(fields.iter().map(|f| f.to_vec()));
                match c.request(&args)? {
                    Reply::Int(n) if n >= 0 => Ok(n as usize),
                    Reply::Error(e) => Err(io::Error::other(string(e))),
                    other => Err(unexpected(other)),
                }
            }
        }
    }

    /// `HLEN key`. Number of fields in the hash (0 if absent).
    pub fn hlen(&mut self, key: &[u8]) -> io::Result<usize> {
        match self {
            Self::Embedded(s) => s.with(|inner| inner.hlen(key)).map_err(store_err),
            Self::Remote(c) => match c.request(&vec2(b"HLEN", key))? {
                Reply::Int(n) if n >= 0 => Ok(n as usize),
                Reply::Error(e) => Err(io::Error::other(string(e))),
                other => Err(unexpected(other)),
            },
        }
    }

    /// `HGETALL key`. Returns a flat `[f0, v0, f1, v1, ...]` matching the
    /// Redis wire shape; empty when the key is absent.
    pub fn hgetall(&mut self, key: &[u8]) -> io::Result<Vec<Vec<u8>>> {
        match self {
            Self::Embedded(s) => s.with(|inner| inner.hgetall(key)).map_err(store_err),
            Self::Remote(c) => match c.request(&vec2(b"HGETALL", key))? {
                Reply::Array(items) => array_to_bulks(items),
                Reply::Error(e) => Err(io::Error::other(string(e))),
                other => Err(unexpected(other)),
            },
        }
    }

    /// `HKEYS key`. Returns the hash's field names (empty if absent).
    pub fn hkeys(&mut self, key: &[u8]) -> io::Result<Vec<Vec<u8>>> {
        match self {
            Self::Embedded(s) => s.with(|inner| inner.hkeys(key)).map_err(store_err),
            Self::Remote(c) => match c.request(&vec2(b"HKEYS", key))? {
                Reply::Array(items) => array_to_bulks(items),
                Reply::Error(e) => Err(io::Error::other(string(e))),
                other => Err(unexpected(other)),
            },
        }
    }

    /// `HVALS key`. Returns the hash's values (empty if absent).
    pub fn hvals(&mut self, key: &[u8]) -> io::Result<Vec<Vec<u8>>> {
        match self {
            Self::Embedded(s) => s.with(|inner| inner.hvals(key)).map_err(store_err),
            Self::Remote(c) => match c.request(&vec2(b"HVALS", key))? {
                Reply::Array(items) => array_to_bulks(items),
                Reply::Error(e) => Err(io::Error::other(string(e))),
                other => Err(unexpected(other)),
            },
        }
    }

    // ===== List =====

    /// `LPUSH key value [value ...]`. Returns the new list length.
    pub fn lpush(&mut self, key: &[u8], values: &[&[u8]]) -> io::Result<usize> {
        match self {
            Self::Embedded(s) => s.lpush(key, values),
            Self::Remote(c) => list_push(c, b"LPUSH", key, values),
        }
    }

    /// `RPUSH key value [value ...]`. Returns the new list length.
    pub fn rpush(&mut self, key: &[u8], values: &[&[u8]]) -> io::Result<usize> {
        match self {
            Self::Embedded(s) => s.rpush(key, values),
            Self::Remote(c) => list_push(c, b"RPUSH", key, values),
        }
    }

    /// `LPOP key count`. Returns up to `count` values from the head; empty
    /// when the key is absent or already drained.
    pub fn lpop(&mut self, key: &[u8], count: usize) -> io::Result<Vec<Vec<u8>>> {
        match self {
            Self::Embedded(s) => s.lpop(key, count),
            Self::Remote(c) => list_pop(c, b"LPOP", key, count),
        }
    }

    /// `RPOP key count`. Symmetric to [`Self::lpop`] from the tail.
    pub fn rpop(&mut self, key: &[u8], count: usize) -> io::Result<Vec<Vec<u8>>> {
        match self {
            Self::Embedded(s) => s.rpop(key, count),
            Self::Remote(c) => list_pop(c, b"RPOP", key, count),
        }
    }

    /// `LLEN key`. 0 when the key is absent.
    pub fn llen(&mut self, key: &[u8]) -> io::Result<usize> {
        match self {
            Self::Embedded(s) => s.llen(key),
            Self::Remote(c) => match c.request(&vec2(b"LLEN", key))? {
                Reply::Int(n) if n >= 0 => Ok(n as usize),
                Reply::Error(e) => Err(io::Error::other(string(e))),
                other => Err(unexpected(other)),
            },
        }
    }

    /// `LRANGE key start stop`. Redis-style indexing — negative offsets
    /// count from the tail (`-1` = last element).
    pub fn lrange(&mut self, key: &[u8], start: i64, stop: i64) -> io::Result<Vec<Vec<u8>>> {
        match self {
            Self::Embedded(s) => s
                .with(|inner| inner.lrange(key, start, stop))
                .map_err(store_err),
            Self::Remote(c) => {
                let args = vec![
                    b"LRANGE".to_vec(),
                    key.to_vec(),
                    start.to_string().into_bytes(),
                    stop.to_string().into_bytes(),
                ];
                match c.request(&args)? {
                    Reply::Array(items) => array_to_bulks(items),
                    Reply::Error(e) => Err(io::Error::other(string(e))),
                    other => Err(unexpected(other)),
                }
            }
        }
    }

    // ===== Set =====

    /// `SADD key member [member ...]`. Returns count of newly added members.
    pub fn sadd(&mut self, key: &[u8], members: &[&[u8]]) -> io::Result<usize> {
        match self {
            Self::Embedded(s) => s.sadd(key, members),
            Self::Remote(c) => set_multi(c, b"SADD", key, members),
        }
    }

    /// `SREM key member [member ...]`. Returns count of removed members.
    pub fn srem(&mut self, key: &[u8], members: &[&[u8]]) -> io::Result<usize> {
        match self {
            Self::Embedded(s) => s.srem(key, members),
            Self::Remote(c) => set_multi(c, b"SREM", key, members),
        }
    }

    /// `SMEMBERS key`. Order is implementation-defined; empty if absent.
    pub fn smembers(&mut self, key: &[u8]) -> io::Result<Vec<Vec<u8>>> {
        match self {
            Self::Embedded(s) => s.smembers(key),
            Self::Remote(c) => match c.request(&vec2(b"SMEMBERS", key))? {
                Reply::Array(items) => array_to_bulks(items),
                Reply::Error(e) => Err(io::Error::other(string(e))),
                other => Err(unexpected(other)),
            },
        }
    }

    /// `SCARD key`. 0 when the key is absent.
    pub fn scard(&mut self, key: &[u8]) -> io::Result<usize> {
        match self {
            Self::Embedded(s) => s.scard(key),
            Self::Remote(c) => match c.request(&vec2(b"SCARD", key))? {
                Reply::Int(n) if n >= 0 => Ok(n as usize),
                Reply::Error(e) => Err(io::Error::other(string(e))),
                other => Err(unexpected(other)),
            },
        }
    }

    /// `SISMEMBER key member`. `false` when key or member absent.
    pub fn sismember(&mut self, key: &[u8], member: &[u8]) -> io::Result<bool> {
        match self {
            Self::Embedded(s) => s
                .with(|inner| inner.sismember(key, member))
                .map_err(store_err),
            Self::Remote(c) => match c.request(&vec3(b"SISMEMBER", key, member))? {
                Reply::Int(1) => Ok(true),
                Reply::Int(0) => Ok(false),
                Reply::Error(e) => Err(io::Error::other(string(e))),
                other => Err(unexpected(other)),
            },
        }
    }

    /// `SINTER key [key ...]` — intersection of all sets.
    pub fn sinter(&mut self, keys: &[&[u8]]) -> io::Result<Vec<Vec<u8>>> {
        match self {
            Self::Embedded(s) => embed_set_combine(s, keys, SetOp::Inter),
            Self::Remote(c) => remote_set_combine(c, b"SINTER", keys),
        }
    }

    /// `SUNION key [key ...]` — union of all sets.
    pub fn sunion(&mut self, keys: &[&[u8]]) -> io::Result<Vec<Vec<u8>>> {
        match self {
            Self::Embedded(s) => embed_set_combine(s, keys, SetOp::Union),
            Self::Remote(c) => remote_set_combine(c, b"SUNION", keys),
        }
    }

    /// `SDIFF key [key ...]` — members of the first set absent from the rest.
    pub fn sdiff(&mut self, keys: &[&[u8]]) -> io::Result<Vec<Vec<u8>>> {
        match self {
            Self::Embedded(s) => embed_set_combine(s, keys, SetOp::Diff),
            Self::Remote(c) => remote_set_combine(c, b"SDIFF", keys),
        }
    }

    // ===== Sorted set =====

    /// `ZADD key score member [score member ...]`. Returns count of newly
    /// added members (overwrites don't count).
    pub fn zadd(&mut self, key: &[u8], pairs: &[(f64, &[u8])]) -> io::Result<usize> {
        match self {
            Self::Embedded(s) => s.zadd(key, pairs),
            Self::Remote(c) => {
                let mut args = Vec::with_capacity(2 + pairs.len() * 2);
                args.push(b"ZADD".to_vec());
                args.push(key.to_vec());
                for (score, m) in pairs {
                    args.push(score.to_string().into_bytes());
                    args.push(m.to_vec());
                }
                match c.request(&args)? {
                    Reply::Int(n) if n >= 0 => Ok(n as usize),
                    Reply::Error(e) => Err(io::Error::other(string(e))),
                    other => Err(unexpected(other)),
                }
            }
        }
    }

    /// `ZREM key member [member ...]`. Returns count of removed members.
    pub fn zrem(&mut self, key: &[u8], members: &[&[u8]]) -> io::Result<usize> {
        match self {
            Self::Embedded(s) => s.zrem(key, members),
            Self::Remote(c) => set_multi(c, b"ZREM", key, members),
        }
    }

    /// `ZSCORE key member`. `None` if absent.
    pub fn zscore(&mut self, key: &[u8], member: &[u8]) -> io::Result<Option<f64>> {
        match self {
            Self::Embedded(s) => s.zscore(key, member),
            Self::Remote(c) => match c.request(&vec3(b"ZSCORE", key, member))? {
                Reply::Bulk(v) => {
                    let s = std::str::from_utf8(&v)
                        .map_err(|_| io::Error::other("non-utf8 score reply"))?;
                    let n: f64 = s
                        .parse()
                        .map_err(|_| io::Error::other(format!("bad score float: {s}")))?;
                    Ok(Some(n))
                }
                Reply::Nil => Ok(None),
                Reply::Error(e) => Err(io::Error::other(string(e))),
                other => Err(unexpected(other)),
            },
        }
    }

    /// `ZCARD key`. Number of members; 0 if absent.
    pub fn zcard(&mut self, key: &[u8]) -> io::Result<usize> {
        match self {
            Self::Embedded(s) => s.zcard(key),
            Self::Remote(c) => match c.request(&vec2(b"ZCARD", key))? {
                Reply::Int(n) if n >= 0 => Ok(n as usize),
                Reply::Error(e) => Err(io::Error::other(string(e))),
                other => Err(unexpected(other)),
            },
        }
    }

    /// `ZRANGE key start stop`. Ascending-score order; negative indices
    /// count from the tail.
    pub fn zrange(&mut self, key: &[u8], start: i64, stop: i64) -> io::Result<Vec<Vec<u8>>> {
        match self {
            Self::Embedded(s) => s
                .with(|inner| inner.zrange(key, start, stop))
                .map(|pairs| pairs.into_iter().map(|(m, _score)| m).collect())
                .map_err(store_err),
            Self::Remote(c) => {
                let args = vec![
                    b"ZRANGE".to_vec(),
                    key.to_vec(),
                    start.to_string().into_bytes(),
                    stop.to_string().into_bytes(),
                ];
                match c.request(&args)? {
                    Reply::Array(items) => array_to_bulks(items),
                    Reply::Error(e) => Err(io::Error::other(string(e))),
                    other => Err(unexpected(other)),
                }
            }
        }
    }
}

// ─────────────────────────────────────────────────────────────────────────
// Shared request builders. Both backends accept a slice of byte-slices,
// but the RESP path needs to splat them into a single argv vector.
// ─────────────────────────────────────────────────────────────────────────

fn list_push(
    c: &mut RespClient,
    verb: &[u8],
    key: &[u8],
    values: &[&[u8]],
) -> io::Result<usize> {
    let mut args = Vec::with_capacity(values.len() + 2);
    args.push(verb.to_vec());
    args.push(key.to_vec());
    args.extend(values.iter().map(|v| v.to_vec()));
    match c.request(&args)? {
        Reply::Int(n) if n >= 0 => Ok(n as usize),
        Reply::Error(e) => Err(io::Error::other(string(e))),
        other => Err(unexpected(other)),
    }
}

fn list_pop(
    c: &mut RespClient,
    verb: &[u8],
    key: &[u8],
    count: usize,
) -> io::Result<Vec<Vec<u8>>> {
    let args = vec![verb.to_vec(), key.to_vec(), count.to_string().into_bytes()];
    match c.request(&args)? {
        Reply::Array(items) => array_to_bulks(items),
        Reply::Bulk(v) => Ok(vec![v]),
        Reply::Nil => Ok(Vec::new()),
        Reply::Error(e) => Err(io::Error::other(string(e))),
        other => Err(unexpected(other)),
    }
}

fn set_multi(
    c: &mut RespClient,
    verb: &[u8],
    key: &[u8],
    members: &[&[u8]],
) -> io::Result<usize> {
    let mut args = Vec::with_capacity(members.len() + 2);
    args.push(verb.to_vec());
    args.push(key.to_vec());
    args.extend(members.iter().map(|m| m.to_vec()));
    match c.request(&args)? {
        Reply::Int(n) if n >= 0 => Ok(n as usize),
        Reply::Error(e) => Err(io::Error::other(string(e))),
        other => Err(unexpected(other)),
    }
}

// Set-combine plumbing: each backend's path computes the intersection /
// union / difference of N sets identified by `keys`.

#[derive(Clone, Copy)]
enum SetOp {
    Inter,
    Union,
    Diff,
}

fn embed_set_combine(
    s: &kevy_embedded::Store,
    keys: &[&[u8]],
    op: SetOp,
) -> io::Result<Vec<Vec<u8>>> {
    use std::collections::HashSet;
    if keys.is_empty() {
        return Ok(Vec::new());
    }
    let snapshots: Vec<Vec<Vec<u8>>> = keys
        .iter()
        .map(|k| s.smembers(k))
        .collect::<io::Result<_>>()?;
    let mut iter = snapshots.into_iter();
    let mut acc: HashSet<Vec<u8>> = iter.next().unwrap_or_default().into_iter().collect();
    for rest in iter {
        let other: HashSet<Vec<u8>> = rest.into_iter().collect();
        acc = match op {
            SetOp::Inter => acc.intersection(&other).cloned().collect(),
            SetOp::Union => acc.union(&other).cloned().collect(),
            SetOp::Diff => acc.difference(&other).cloned().collect(),
        };
    }
    Ok(acc.into_iter().collect())
}

fn remote_set_combine(
    c: &mut RespClient,
    verb: &[u8],
    keys: &[&[u8]],
) -> io::Result<Vec<Vec<u8>>> {
    let mut args = Vec::with_capacity(keys.len() + 1);
    args.push(verb.to_vec());
    args.extend(keys.iter().map(|k| k.to_vec()));
    match c.request(&args)? {
        Reply::Array(items) => array_to_bulks(items),
        Reply::Error(e) => Err(io::Error::other(string(e))),
        other => Err(unexpected(other)),
    }
}

#[cfg(test)]
#[path = "collections_tests.rs"]
mod tests;