ruvio-client 0.2.2

use std::collections::HashMap;

use std::time::{Duration, SystemTime};

use crate::client::{Client, duration_millis, duration_secs, join, unexpected, unix_millis};

use crate::error::Error;
use crate::models::{
    HashScanPage, PubSubMessage, ScanPage, SortedSetAddOptions, SortedSetEntry, StreamClaimOptions,
    StreamEntry, StreamPendingEntry, StreamPendingFilter, StreamReadOptions, StreamReadResult,
};

use crate::value::RespValue;

macro_rules! args {
    ($($argument:expr),* $(,)?) => {
        vec![$($argument.to_string()),*]
    };
}

impl Client {
    /// Sends `MGET`. A missing key is `None`.
    pub fn mget(&mut self, keys: &[&str]) -> Result<Vec<Option<String>>, Error> {
        self.optional_strings(join("MGET", keys))
    }

    /// Sends `MSET`.
    pub fn mset(&mut self, pairs: &[(&str, &str)]) -> Result<(), Error> {
        self.ok(with_pairs(args!["MSET"], pairs))
    }

    /// Sends `GETSET` and returns the previous value.
    pub fn getset(&mut self, key: &str, value: &str) -> Result<Option<String>, Error> {
        self.bulk_or_null(["GETSET", key, value])
    }

    /// Sends `APPEND` and returns the new length.
    pub fn append(&mut self, key: &str, value: &str) -> Result<i64, Error> {
        self.integer(["APPEND", key, value])
    }

    /// Sends `STRLEN`.
    pub fn strlen(&mut self, key: &str) -> Result<i64, Error> {
        self.integer(["STRLEN", key])
    }

    /// Sends `DECR`.
    pub fn decr(&mut self, key: &str) -> Result<i64, Error> {
        self.integer(["DECR", key])
    }

    /// Sends `INCRBY`.
    pub fn incr_by(&mut self, key: &str, delta: i64) -> Result<i64, Error> {
        self.integer(args!["INCRBY", key, delta])
    }

    /// Sends `DECRBY`.
    pub fn decr_by(&mut self, key: &str, delta: i64) -> Result<i64, Error> {
        self.integer(args!["DECRBY", key, delta])
    }

    /// Sends `UNLINK` for one key.
    pub fn unlink_key(&mut self, key: &str) -> Result<i64, Error> {
        self.unlink(&[key])
    }

    /// Sends `UNLINK` and returns how many keys were removed.
    pub fn unlink(&mut self, keys: &[&str]) -> Result<i64, Error> {
        self.integer(join("UNLINK", keys))
    }

    /// Sends `EXISTS` for one key.
    pub fn exists_key(&mut self, key: &str) -> Result<i64, Error> {
        self.exists(&[key])
    }

    /// Sends `EXISTS` and returns how many of the keys exist.
    pub fn exists(&mut self, keys: &[&str]) -> Result<i64, Error> {
        self.integer(join("EXISTS", keys))
    }

    /// Sends `TYPE`: `string`, `list`, `set`, `hash`, `zset`, `stream`, or `none`.
    pub fn key_type(&mut self, key: &str) -> Result<String, Error> {
        self.text(["TYPE", key])
    }

    /// Sends `RENAME`.
    pub fn rename(&mut self, key: &str, new_key: &str) -> Result<(), Error> {
        self.ok(["RENAME", key, new_key])
    }

    /// Sends `SCAN` with an optional `MATCH` pattern and `COUNT` hint.
    pub fn scan(
        &mut self,
        cursor: u64,
        pattern: Option<&str>,
        count: Option<u64>,
    ) -> Result<ScanPage, Error> {
        let mut arguments = args!["SCAN", cursor];

        push_scan_options(&mut arguments, pattern, count);

        let (cursor, keys) = parse_scan("SCAN", self.run(arguments)?)?;

        Ok(ScanPage { cursor, keys })
    }

    /// Sends `DBSIZE` for the selected database.
    pub fn dbsize(&mut self) -> Result<i64, Error> {
        self.integer(["DBSIZE"])
    }

    /// Sends `EXPIREAT`, or `PEXPIREAT` when `when` is not a whole second.
    pub fn expire_at(&mut self, key: &str, when: SystemTime) -> Result<bool, Error> {
        let millis = unix_millis(when)?;

        if millis % 1000 == 0 {
            return self.flag(args!["EXPIREAT", key, millis / 1000]);
        }

        self.flag(args!["PEXPIREAT", key, millis])
    }

    /// Sends `PEXPIRE`. False when the key does not exist.
    pub fn pexpire(&mut self, key: &str, ttl: Duration) -> Result<bool, Error> {
        self.flag(args!["PEXPIRE", key, duration_millis(ttl)])
    }

    /// Sends `TTL`: seconds left, -1 without an expiry, -2 when the key is missing.
    pub fn ttl(&mut self, key: &str) -> Result<i64, Error> {
        self.integer(["TTL", key])
    }

    /// Sends `PTTL`: milliseconds left, -1 without an expiry, -2 when the key is missing.
    pub fn pttl(&mut self, key: &str) -> Result<i64, Error> {
        self.integer(["PTTL", key])
    }

    /// Sends `PERSIST`. False when the key is missing or has no expiry.
    pub fn persist(&mut self, key: &str) -> Result<bool, Error> {
        self.flag(["PERSIST", key])
    }

    /// Sends `LPUSH` and returns the new length.
    pub fn lpush(&mut self, key: &str, values: &[&str]) -> Result<i64, Error> {
        self.integer(with(args!["LPUSH", key], values))
    }

    /// Sends `RPUSH` and returns the new length.
    pub fn rpush(&mut self, key: &str, values: &[&str]) -> Result<i64, Error> {
        self.integer(with(args!["RPUSH", key], values))
    }

    /// Sends `LPOP` for one element.
    pub fn lpop(&mut self, key: &str) -> Result<Option<String>, Error> {
        self.bulk_or_null(["LPOP", key])
    }

    /// Sends `LPOP` with a count. A missing list returns an empty vector.
    pub fn lpop_count(&mut self, key: &str, count: u64) -> Result<Vec<String>, Error> {
        self.strings(args!["LPOP", key, count])
    }

    /// Sends `RPOP` for one element.
    pub fn rpop(&mut self, key: &str) -> Result<Option<String>, Error> {
        self.bulk_or_null(["RPOP", key])
    }

    /// Sends `RPOP` with a count. A missing list returns an empty vector.
    pub fn rpop_count(&mut self, key: &str, count: u64) -> Result<Vec<String>, Error> {
        self.strings(args!["RPOP", key, count])
    }

    /// Sends `LLEN`.
    pub fn llen(&mut self, key: &str) -> Result<i64, Error> {
        self.integer(["LLEN", key])
    }

    /// Sends `LINDEX`.
    pub fn lindex(&mut self, key: &str, index: i64) -> Result<Option<String>, Error> {
        self.bulk_or_null(args!["LINDEX", key, index])
    }

    /// Sends `LRANGE`. `stop` is inclusive and negative indexes count from the tail.
    pub fn lrange(&mut self, key: &str, start: i64, stop: i64) -> Result<Vec<String>, Error> {
        self.strings(args!["LRANGE", key, start, stop])
    }

    /// Sends `LTRIM`.
    pub fn ltrim(&mut self, key: &str, start: i64, stop: i64) -> Result<(), Error> {
        self.ok(args!["LTRIM", key, start, stop])
    }

    /// Sends `BLPOP` for one key. Returns `(key, value)`, or `None` when
    /// `timeout` passes. `Duration::ZERO` waits forever.
    pub fn blpop_key(
        &mut self,
        key: &str,
        timeout: Duration,
    ) -> Result<Option<(String, String)>, Error> {
        self.blpop(&[key], timeout)
    }

    /// Sends `BLPOP`. Returns `(key, value)`, or `None` when `timeout` passes.
    /// `Duration::ZERO` waits forever. The timeout rounds up to whole seconds.
    pub fn blpop(
        &mut self,
        keys: &[&str],
        timeout: Duration,
    ) -> Result<Option<(String, String)>, Error> {
        self.blocking_pop("BLPOP", keys, timeout)
    }

    /// Sends `BRPOP` for one key. Returns `(key, value)`, or `None` when
    /// `timeout` passes.
    pub fn brpop_key(
        &mut self,
        key: &str,
        timeout: Duration,
    ) -> Result<Option<(String, String)>, Error> {
        self.brpop(&[key], timeout)
    }

    /// Sends `BRPOP`. Returns `(key, value)`, or `None` when `timeout` passes.
    pub fn brpop(
        &mut self,
        keys: &[&str],
        timeout: Duration,
    ) -> Result<Option<(String, String)>, Error> {
        self.blocking_pop("BRPOP", keys, timeout)
    }

    /// Sends `SADD` and returns how many members were new.
    pub fn sadd(&mut self, key: &str, members: &[&str]) -> Result<i64, Error> {
        self.integer(with(args!["SADD", key], members))
    }

    /// Sends `SREM` and returns how many members were removed.
    pub fn srem(&mut self, key: &str, members: &[&str]) -> Result<i64, Error> {
        self.integer(with(args!["SREM", key], members))
    }

    /// Sends `SISMEMBER`.
    pub fn sismember(&mut self, key: &str, member: &str) -> Result<bool, Error> {
        self.flag(["SISMEMBER", key, member])
    }

    /// Sends `SCARD`.
    pub fn scard(&mut self, key: &str) -> Result<i64, Error> {
        self.integer(["SCARD", key])
    }

    /// Sends `SMEMBERS`.
    pub fn smembers(&mut self, key: &str) -> Result<Vec<String>, Error> {
        self.strings(["SMEMBERS", key])
    }

    /// Sends `SINTER`.
    pub fn sinter(&mut self, keys: &[&str]) -> Result<Vec<String>, Error> {
        self.strings(join("SINTER", keys))
    }

    /// Sends `SUNION`.
    pub fn sunion(&mut self, keys: &[&str]) -> Result<Vec<String>, Error> {
        self.strings(join("SUNION", keys))
    }

    /// Sends `SDIFF`.
    pub fn sdiff(&mut self, keys: &[&str]) -> Result<Vec<String>, Error> {
        self.strings(join("SDIFF", keys))
    }

    /// Sends `SINTERSTORE` and returns the size of `destination`.
    pub fn sinterstore(&mut self, destination: &str, keys: &[&str]) -> Result<i64, Error> {
        self.integer(with(args!["SINTERSTORE", destination], keys))
    }

    /// Sends `SUNIONSTORE` and returns the size of `destination`.
    pub fn sunionstore(&mut self, destination: &str, keys: &[&str]) -> Result<i64, Error> {
        self.integer(with(args!["SUNIONSTORE", destination], keys))
    }

    /// Sends `SDIFFSTORE` and returns the size of `destination`.
    pub fn sdiffstore(&mut self, destination: &str, keys: &[&str]) -> Result<i64, Error> {
        self.integer(with(args!["SDIFFSTORE", destination], keys))
    }

    /// Sends `SMOVE`. False when `member` was not in `source`.
    pub fn smove(&mut self, source: &str, destination: &str, member: &str) -> Result<bool, Error> {
        self.flag(["SMOVE", source, destination, member])
    }

    /// Sends `SPOP` for one member.
    pub fn spop(&mut self, key: &str) -> Result<Option<String>, Error> {
        self.bulk_or_null(["SPOP", key])
    }

    /// Sends `SPOP` with a count and returns the removed members.
    pub fn spop_count(&mut self, key: &str, count: u64) -> Result<Vec<String>, Error> {
        self.strings(args!["SPOP", key, count])
    }

    /// Sends `SRANDMEMBER` for one member.
    pub fn srandmember(&mut self, key: &str) -> Result<Option<String>, Error> {
        self.bulk_or_null(["SRANDMEMBER", key])
    }

    /// Sends `SRANDMEMBER` with a count. A positive count returns distinct members;
    /// a negative count may repeat members.
    pub fn srandmember_count(&mut self, key: &str, count: i64) -> Result<Vec<String>, Error> {
        self.strings(args!["SRANDMEMBER", key, count])
    }

    /// Sends `HSET` and returns how many fields were new.
    pub fn hset(&mut self, key: &str, fields: &[(&str, &str)]) -> Result<i64, Error> {
        self.integer(with_pairs(args!["HSET", key], fields))
    }

    /// Sends `HGET`.
    pub fn hget(&mut self, key: &str, field: &str) -> Result<Option<String>, Error> {
        self.bulk_or_null(["HGET", key, field])
    }

    /// Sends `HDEL` and returns how many fields were removed.
    pub fn hdel(&mut self, key: &str, fields: &[&str]) -> Result<i64, Error> {
        self.integer(with(args!["HDEL", key], fields))
    }

    /// Sends `HLEN`.
    pub fn hlen(&mut self, key: &str) -> Result<i64, Error> {
        self.integer(["HLEN", key])
    }

    /// Sends `HGETALL`.
    pub fn hgetall(&mut self, key: &str) -> Result<HashMap<String, String>, Error> {
        let pairs = pairs(&self.run(["HGETALL", key])?)?;

        Ok(pairs.into_iter().collect())
    }

    /// Sends `HMGET`. A missing field is `None`.
    pub fn hmget(&mut self, key: &str, fields: &[&str]) -> Result<Vec<Option<String>>, Error> {
        self.optional_strings(with(args!["HMGET", key], fields))
    }

    /// Sends `HEXISTS`.
    pub fn hexists(&mut self, key: &str, field: &str) -> Result<bool, Error> {
        self.flag(["HEXISTS", key, field])
    }

    /// Sends `HKEYS`.
    pub fn hkeys(&mut self, key: &str) -> Result<Vec<String>, Error> {
        self.strings(["HKEYS", key])
    }

    /// Sends `HVALS`.
    pub fn hvals(&mut self, key: &str) -> Result<Vec<String>, Error> {
        self.strings(["HVALS", key])
    }

    /// Sends `HINCRBY` and returns the new value.
    pub fn hincr_by(&mut self, key: &str, field: &str, delta: i64) -> Result<i64, Error> {
        self.integer(args!["HINCRBY", key, field, delta])
    }

    /// Sends `HSETNX`. False when the field already existed.
    pub fn hsetnx(&mut self, key: &str, field: &str, value: &str) -> Result<bool, Error> {
        self.flag(["HSETNX", key, field, value])
    }

    /// Sends `HSTRLEN`.
    pub fn hstrlen(&mut self, key: &str, field: &str) -> Result<i64, Error> {
        self.integer(["HSTRLEN", key, field])
    }

    /// Sends `HSCAN` with an optional `MATCH` pattern and `COUNT` hint.
    pub fn hscan(
        &mut self,
        key: &str,
        cursor: u64,
        pattern: Option<&str>,
        count: Option<u64>,
    ) -> Result<HashScanPage, Error> {
        let mut arguments = args!["HSCAN", key, cursor];

        push_scan_options(&mut arguments, pattern, count);

        let (cursor, flat) = parse_scan("HSCAN", self.run(arguments)?)?;
        let mut fields = Vec::with_capacity(flat.len() / 2);
        let mut items = flat.into_iter();

        while let (Some(field), Some(value)) = (items.next(), items.next()) {
            fields.push((field, value));
        }

        Ok(HashScanPage { cursor, fields })
    }

    /// Sends `ZADD` with `(member, score)` pairs and returns how many members were new.
    ///
    /// ```no_run
    /// # let mut db = ruvio_client::Client::connect("127.0.0.1", 6379)?;
    /// db.zadd("board", &[("ada", 100.0), ("bob", 50.0)])?;
    /// # Ok::<(), ruvio_client::Error>(())
    /// ```
    pub fn zadd(&mut self, key: &str, entries: &[(&str, f64)]) -> Result<i64, Error> {
        self.zadd_with(key, SortedSetAddOptions::default(), entries)
    }

    /// Sends `ZADD` with `NX`, `XX`, `GT`, `LT`, or `CH`.
    /// Without [`SortedSetAddOptions::changed`] the reply counts new members.
    pub fn zadd_with(
        &mut self,
        key: &str,
        options: SortedSetAddOptions,
        entries: &[(&str, f64)],
    ) -> Result<i64, Error> {
        let mut arguments = zadd_arguments(key, options);

        for (member, score) in entries {
            arguments.push(score.to_string());
            arguments.push((*member).to_owned());
        }

        self.integer(arguments)
    }

    /// Sends `ZADD ... INCR`. Returns the new score, or `None` when an option skipped the update.
    pub fn zadd_incr(
        &mut self,
        key: &str,
        member: &str,
        delta: f64,
        options: SortedSetAddOptions,
    ) -> Result<Option<f64>, Error> {
        let mut arguments = zadd_arguments(key, options);

        arguments.extend(args!["INCR", delta, member]);
        self.bulk_or_null(arguments)?
            .map(|text| parse_score(&text))
            .transpose()
    }

    /// Sends `ZINCRBY` and returns the new score.
    pub fn zincr_by(&mut self, key: &str, member: &str, delta: f64) -> Result<f64, Error> {
        parse_score(&self.text(args!["ZINCRBY", key, delta, member])?)
    }

    /// Sends `ZRANGE` by rank. `stop` is inclusive.
    pub fn zrange(&mut self, key: &str, start: i64, stop: i64) -> Result<Vec<String>, Error> {
        self.strings(args!["ZRANGE", key, start, stop])
    }

    /// Sends `ZREVRANGE` by rank. `stop` is inclusive.
    pub fn zrevrange(&mut self, key: &str, start: i64, stop: i64) -> Result<Vec<String>, Error> {
        self.strings(args!["ZREVRANGE", key, start, stop])
    }

    /// Sends `ZRANGE ... WITHSCORES`.
    pub fn zrange_with_scores(
        &mut self,
        key: &str,
        start: i64,
        stop: i64,
    ) -> Result<Vec<SortedSetEntry>, Error> {
        sorted_set(&self.run(args!["ZRANGE", key, start, stop, "WITHSCORES"])?)
    }

    /// Sends `ZREVRANGE ... WITHSCORES`.
    pub fn zrevrange_with_scores(
        &mut self,
        key: &str,
        start: i64,
        stop: i64,
    ) -> Result<Vec<SortedSetEntry>, Error> {
        sorted_set(&self.run(args!["ZREVRANGE", key, start, stop, "WITHSCORES"])?)
    }

    /// Sends `ZRANGEBYSCORE ... WITHSCORES`. `min` and `max` accept `-inf`, `+inf`, and `(` for exclusive bounds.
    pub fn zrange_by_score_with_scores(
        &mut self,
        key: &str,
        min: &str,
        max: &str,
    ) -> Result<Vec<SortedSetEntry>, Error> {
        sorted_set(&self.run(["ZRANGEBYSCORE", key, min, max, "WITHSCORES"])?)
    }

    /// Sends `ZREM` and returns how many members were removed.
    pub fn zrem(&mut self, key: &str, members: &[&str]) -> Result<i64, Error> {
        self.integer(with(args!["ZREM", key], members))
    }

    /// Sends `ZCARD`.
    pub fn zcard(&mut self, key: &str) -> Result<i64, Error> {
        self.integer(["ZCARD", key])
    }

    /// Sends `ZSCORE`.
    pub fn zscore(&mut self, key: &str, member: &str) -> Result<Option<f64>, Error> {
        self.bulk_or_null(["ZSCORE", key, member])?
            .map(|text| parse_score(&text))
            .transpose()
    }

    /// Sends `ZRANK`: the 0-based rank from the lowest score.
    pub fn zrank(&mut self, key: &str, member: &str) -> Result<Option<i64>, Error> {
        self.optional_integer(["ZRANK", key, member])
    }

    /// Sends `ZREVRANK`: the 0-based rank from the highest score.
    pub fn zrevrank(&mut self, key: &str, member: &str) -> Result<Option<i64>, Error> {
        self.optional_integer(["ZREVRANK", key, member])
    }

    /// Sends `BF.RESERVE`.
    pub fn bf_reserve(&mut self, key: &str, error_rate: f64, capacity: u64) -> Result<(), Error> {
        self.ok(args!["BF.RESERVE", key, error_rate, capacity])
    }

    /// Sends `BF.ADD`. False when the item may already have been added.
    pub fn bf_add(&mut self, key: &str, item: &str) -> Result<bool, Error> {
        self.flag(["BF.ADD", key, item])
    }

    /// Sends `BF.EXISTS`. True means "maybe added"; false means "never added".
    pub fn bf_exists(&mut self, key: &str, item: &str) -> Result<bool, Error> {
        self.flag(["BF.EXISTS", key, item])
    }

    /// Sends `XADD` and returns the entry id. Use `*` for `id` to let the server pick one.
    pub fn xadd(&mut self, key: &str, id: &str, fields: &[(&str, &str)]) -> Result<String, Error> {
        self.text(with_pairs(args!["XADD", key, id], fields))
    }

    /// Sends `XADD key MAXLEN n`: the stream keeps exactly the newest `max_len` entries.
    pub fn xadd_maxlen(
        &mut self,
        key: &str,
        max_len: u64,
        id: &str,
        fields: &[(&str, &str)],
    ) -> Result<String, Error> {
        self.text(with_pairs(
            args!["XADD", key, "MAXLEN", max_len, id],
            fields,
        ))
    }

    /// Sends `XLEN`.
    pub fn xlen(&mut self, key: &str) -> Result<i64, Error> {
        self.integer(["XLEN", key])
    }

    /// Sends `XRANGE` from `start` to `end` (`-` and `+` for the ends), with an optional `COUNT`.
    pub fn xrange(
        &mut self,
        key: &str,
        start: &str,
        end: &str,
        count: Option<u64>,
    ) -> Result<Vec<StreamEntry>, Error> {
        let mut arguments = args!["XRANGE", key, start, end];

        if let Some(count) = count {
            arguments.extend(args!["COUNT", count]);
        }

        stream_entries(&self.run(arguments)?)
    }

    /// Sends `XREVRANGE` from `end` down to `start`, with an optional `COUNT`.
    pub fn xrevrange(
        &mut self,
        key: &str,
        end: &str,
        start: &str,
        count: Option<u64>,
    ) -> Result<Vec<StreamEntry>, Error> {
        let mut arguments = args!["XREVRANGE", key, end, start];

        if let Some(count) = count {
            arguments.extend(args!["COUNT", count]);
        }

        stream_entries(&self.run(arguments)?)
    }

    /// Sends `XDEL` and returns how many entries were removed.
    pub fn xdel(&mut self, key: &str, ids: &[&str]) -> Result<i64, Error> {
        self.integer(with(args!["XDEL", key], ids))
    }

    /// Sends `XTRIM MAXLEN` and returns how many entries were removed.
    pub fn xtrim_maxlen(&mut self, key: &str, max_len: u64) -> Result<i64, Error> {
        self.integer(args!["XTRIM", key, "MAXLEN", max_len])
    }

    /// Sends `XREAD` for `(key, id)` pairs. A block that times out returns an empty vector.
    pub fn xread(
        &mut self,
        options: &StreamReadOptions,
        streams: &[(&str, &str)],
    ) -> Result<Vec<StreamReadResult>, Error> {
        if options.no_ack {
            return Err(Error::Protocol(
                "NOACK applies only to XREADGROUP".to_owned(),
            ));
        }

        let mut arguments = args!["XREAD"];

        push_stream_read_options(&mut arguments, options);
        push_streams(&mut arguments, streams);
        stream_read(&self.run(arguments)?)
    }

    /// Sends `XGROUP CREATE`. With `make_stream` it adds `MKSTREAM`.
    pub fn xgroup_create(
        &mut self,
        key: &str,
        group: &str,
        id: &str,
        make_stream: bool,
    ) -> Result<(), Error> {
        let mut arguments = args!["XGROUP", "CREATE", key, group, id];

        if make_stream {
            arguments.push("MKSTREAM".to_owned());
        }

        self.ok(arguments)
    }

    /// Sends `XREADGROUP GROUP` with `COUNT`, `BLOCK`, and `NOACK` from `options`.
    /// Use `>` as the id for entries never delivered to the group.
    pub fn xreadgroup(
        &mut self,
        group: &str,
        consumer: &str,
        options: &StreamReadOptions,
        streams: &[(&str, &str)],
    ) -> Result<Vec<StreamReadResult>, Error> {
        let mut arguments = args!["XREADGROUP", "GROUP", group, consumer];

        push_stream_read_options(&mut arguments, options);

        if options.no_ack {
            arguments.push("NOACK".to_owned());
        }

        push_streams(&mut arguments, streams);
        stream_read(&self.run(arguments)?)
    }

    /// Sends `XGROUP DESTROY`. False when the group did not exist.
    pub fn xgroup_destroy(&mut self, key: &str, group: &str) -> Result<bool, Error> {
        self.flag(["XGROUP", "DESTROY", key, group])
    }

    /// Sends `XGROUP SETID`.
    pub fn xgroup_setid(&mut self, key: &str, group: &str, id: &str) -> Result<(), Error> {
        self.ok(["XGROUP", "SETID", key, group, id])
    }

    /// Sends `XGROUP DELCONSUMER` and returns how many pending entries the consumer had.
    pub fn xgroup_delconsumer(
        &mut self,
        key: &str,
        group: &str,
        consumer: &str,
    ) -> Result<i64, Error> {
        self.integer(["XGROUP", "DELCONSUMER", key, group, consumer])
    }

    /// Sends `XACK` and returns how many entries were acknowledged.
    pub fn xack(&mut self, key: &str, group: &str, ids: &[&str]) -> Result<i64, Error> {
        self.integer(with(args!["XACK", key, group], ids))
    }

    /// Sends the summary form of `XPENDING`: count, smallest id, largest id, and per-consumer counts.
    pub fn xpending_summary(&mut self, key: &str, group: &str) -> Result<RespValue, Error> {
        self.run(["XPENDING", key, group])
    }

    /// Sends the extended `XPENDING` form: ids from `start` to `end` (inclusive), at most `count`,
    /// narrowed by `filter`.
    pub fn xpending(
        &mut self,
        key: &str,
        group: &str,
        start: &str,
        end: &str,
        count: u64,
        filter: &StreamPendingFilter,
    ) -> Result<Vec<StreamPendingEntry>, Error> {
        let mut arguments = args!["XPENDING", key, group];

        if let Some(idle) = filter.min_idle {
            arguments.extend(args!["IDLE", duration_millis(idle)]);
        }

        arguments.extend(args![start, end, count]);

        if let Some(consumer) = &filter.consumer {
            arguments.push(consumer.clone());
        }

        let reply = self.run(arguments)?;
        let mut rows = Vec::new();

        for row in reply.as_array().unwrap_or_default() {
            let Some([id, owner, idle, deliveries]) = row.as_array() else {
                return Err(unexpected("XPENDING", row));
            };

            rows.push(StreamPendingEntry {
                id: id.as_string()?.unwrap_or_default(),
                consumer: owner.as_string()?.unwrap_or_default(),
                idle_millis: idle.as_integer().unwrap_or_default(),
                delivery_count: deliveries.as_integer().unwrap_or_default(),
            });
        }

        Ok(rows)
    }

    /// Sends `XCLAIM` with `IDLE`, `TIME`, `RETRYCOUNT`, `FORCE`, or `LASTID` from `options`.
    pub fn xclaim(
        &mut self,
        key: &str,
        group: &str,
        consumer: &str,
        min_idle: Duration,
        ids: &[&str],
        options: &StreamClaimOptions,
    ) -> Result<Vec<StreamEntry>, Error> {
        let arguments = claim_arguments(key, group, consumer, min_idle, ids, options)?;

        stream_entries(&self.run(arguments)?)
    }

    /// Sends `XCLAIM ... JUSTID` and returns only the claimed ids.
    pub fn xclaim_ids(
        &mut self,
        key: &str,
        group: &str,
        consumer: &str,
        min_idle: Duration,
        ids: &[&str],
        options: &StreamClaimOptions,
    ) -> Result<Vec<String>, Error> {
        let mut arguments = claim_arguments(key, group, consumer, min_idle, ids, options)?;

        arguments.push("JUSTID".to_owned());
        self.strings(arguments)
    }

    /// Sends `INFO`, or `INFO section`. The `prometheus` section returns Prometheus text.
    pub fn info(&mut self, section: Option<&str>) -> Result<String, Error> {
        match section {
            Some(section) => self.text(["INFO", section]),
            None => self.text(["INFO"]),
        }
    }

    /// Sends `CONFIG GET` and returns `(parameter, value)` pairs.
    pub fn config_get(&mut self, parameter: &str) -> Result<Vec<(String, String)>, Error> {
        pairs(&self.run(["CONFIG", "GET", parameter])?)
    }

    /// Sends `SAVE`.
    pub fn save(&mut self) -> Result<(), Error> {
        self.ok(["SAVE"])
    }

    /// Sends `BGSAVE`.
    pub fn bgsave(&mut self) -> Result<(), Error> {
        self.ok(["BGSAVE"])
    }

    /// Sends `FLUSHDB`. Only the selected database is emptied.
    pub fn flushdb(&mut self) -> Result<(), Error> {
        self.ok(["FLUSHDB"])
    }

    /// Sends `FLUSHALL`. Every database is emptied.
    pub fn flushall(&mut self) -> Result<(), Error> {
        self.ok(["FLUSHALL"])
    }

    /// Sends `SWAPDB`.
    pub fn swapdb(&mut self, first: u32, second: u32) -> Result<(), Error> {
        self.ok(args!["SWAPDB", first, second])
    }

    /// Sends `MOVE`. False when the key is missing or already exists in `database`.
    pub fn move_key(&mut self, key: &str, database: u32) -> Result<bool, Error> {
        self.flag(args!["MOVE", key, database])
    }

    /// Sends `MULTI`. Until [`Self::exec`], the server answers `QUEUED`, so queue commands
    /// with [`Client::execute`] rather than the typed methods.
    pub fn multi(&mut self) -> Result<(), Error> {
        self.ok(["MULTI"])
    }

    /// Sends `EXEC`. `None` when a watched key changed and the transaction was aborted.
    pub fn exec(&mut self) -> Result<Option<Vec<RespValue>>, Error> {
        match self.run(["EXEC"])? {
            RespValue::Null => Ok(None),
            RespValue::Array(items) => Ok(Some(items)),
            other => Err(unexpected("EXEC", &other)),
        }
    }

    /// Sends `DISCARD`.
    pub fn discard(&mut self) -> Result<(), Error> {
        self.ok(["DISCARD"])
    }

    /// Sends `WATCH` for one key.
    pub fn watch_key(&mut self, key: &str) -> Result<(), Error> {
        self.watch(&[key])
    }

    /// Sends `WATCH`.
    pub fn watch(&mut self, keys: &[&str]) -> Result<(), Error> {
        self.ok(join("WATCH", keys))
    }

    /// Sends `UNWATCH`.
    pub fn unwatch(&mut self) -> Result<(), Error> {
        self.ok(["UNWATCH"])
    }

    /// Sends `EVAL`. The key count is `keys.len()`.
    pub fn eval(
        &mut self,
        script: &str,
        keys: &[&str],
        arguments: &[&str],
    ) -> Result<RespValue, Error> {
        self.run(script_arguments("EVAL", script, keys, arguments))
    }

    /// Sends `EVALSHA`. The key count is `keys.len()`.
    pub fn evalsha(
        &mut self,
        sha: &str,
        keys: &[&str],
        arguments: &[&str],
    ) -> Result<RespValue, Error> {
        self.run(script_arguments("EVALSHA", sha, keys, arguments))
    }

    /// Sends `SCRIPT LOAD` and returns the SHA1 for [`Self::evalsha`].
    pub fn script_load(&mut self, script: &str) -> Result<String, Error> {
        self.text(["SCRIPT", "LOAD", script])
    }

    /// Sends `SCRIPT EXISTS`, one flag per hash.
    pub fn script_exists(&mut self, hashes: &[&str]) -> Result<Vec<bool>, Error> {
        let reply = self.run(with(args!["SCRIPT", "EXISTS"], hashes))?;

        Ok(reply
            .as_array()
            .unwrap_or_default()
            .iter()
            .map(|item| item.as_integer().unwrap_or_default() > 0)
            .collect())
    }

    /// Sends `SCRIPT FLUSH`.
    pub fn script_flush(&mut self) -> Result<(), Error> {
        self.ok(["SCRIPT", "FLUSH"])
    }

    /// Sends `SCRIPT KILL`.
    pub fn script_kill(&mut self) -> Result<(), Error> {
        self.ok(["SCRIPT", "KILL"])
    }

    /// Sends `FCALL`. The key count is `keys.len()`.
    pub fn fcall(
        &mut self,
        name: &str,
        keys: &[&str],
        arguments: &[&str],
    ) -> Result<RespValue, Error> {
        self.run(script_arguments("FCALL", name, keys, arguments))
    }

    /// Sends `FUNCTION LOAD name body`. The body is plain Lua, not a Redis library shebang,
    /// and is lost when the server restarts.
    pub fn function_load(&mut self, name: &str, body: &str) -> Result<(), Error> {
        self.ok(["FUNCTION", "LOAD", name, body])
    }

    /// Sends `FUNCTION LIST`.
    pub fn function_list(&mut self) -> Result<RespValue, Error> {
        self.run(["FUNCTION", "LIST"])
    }

    /// Sends `FUNCTION DELETE`. False when no function had that name.
    pub fn function_delete(&mut self, name: &str) -> Result<bool, Error> {
        self.flag(["FUNCTION", "DELETE", name])
    }

    /// Sends `ACL WHOAMI`.
    pub fn acl_whoami(&mut self) -> Result<String, Error> {
        self.text(["ACL", "WHOAMI"])
    }

    /// Sends `ACL USERS`.
    pub fn acl_users(&mut self) -> Result<Vec<String>, Error> {
        self.strings(["ACL", "USERS"])
    }

    /// Sends `ACL LIST`.
    pub fn acl_list(&mut self) -> Result<Vec<String>, Error> {
        self.strings(["ACL", "LIST"])
    }

    /// Sends `ACL GETUSER`.
    pub fn acl_getuser(&mut self, user: &str) -> Result<RespValue, Error> {
        self.run(["ACL", "GETUSER", user])
    }

    /// Sends `ACL CAT`.
    pub fn acl_cat(&mut self) -> Result<Vec<String>, Error> {
        self.strings(["ACL", "CAT"])
    }

    /// Sends `ACL SETUSER user rule...`.
    pub fn acl_setuser(&mut self, user: &str, rules: &[&str]) -> Result<(), Error> {
        self.ok(with(args!["ACL", "SETUSER", user], rules))
    }

    /// Sends `CLUSTER SLOTS`.
    pub fn cluster_slots(&mut self) -> Result<RespValue, Error> {
        self.run(["CLUSTER", "SLOTS"])
    }

    /// Sends `CLUSTER NODES`.
    pub fn cluster_nodes(&mut self) -> Result<String, Error> {
        self.text(["CLUSTER", "NODES"])
    }

    /// Sends `CLUSTER SHARDS`.
    pub fn cluster_shards(&mut self) -> Result<RespValue, Error> {
        self.run(["CLUSTER", "SHARDS"])
    }

    /// Sends `CLUSTER INFO`.
    pub fn cluster_info(&mut self) -> Result<String, Error> {
        self.text(["CLUSTER", "INFO"])
    }

    /// Sends `CLUSTER MYID`.
    pub fn cluster_myid(&mut self) -> Result<String, Error> {
        self.text(["CLUSTER", "MYID"])
    }

    /// Sends `CLUSTER KEYSLOT`.
    pub fn cluster_keyslot(&mut self, key: &str) -> Result<i64, Error> {
        self.integer(["CLUSTER", "KEYSLOT", key])
    }

    /// Sends `READONLY`. Ruvio has no replicas; the reply is still `OK` in cluster mode.
    pub fn readonly(&mut self) -> Result<(), Error> {
        self.ok(["READONLY"])
    }

    /// Sends `READWRITE`.
    pub fn readwrite(&mut self) -> Result<(), Error> {
        self.ok(["READWRITE"])
    }

    /// Sends `CLIENT GETNAME`.
    pub fn client_getname(&mut self) -> Result<Option<String>, Error> {
        self.bulk_or_null(["CLIENT", "GETNAME"])
    }

    /// Sends `CLIENT SETNAME`.
    pub fn client_setname(&mut self, name: &str) -> Result<(), Error> {
        self.ok(["CLIENT", "SETNAME", name])
    }

    /// Sends `CLIENT TRACKING ON` or `OFF`.
    pub fn client_tracking(&mut self, enabled: bool) -> Result<(), Error> {
        self.ok(["CLIENT", "TRACKING", if enabled { "ON" } else { "OFF" }])
    }

    /// Sends `HELLO protocol`. Connecting itself does not send this, and this client
    /// only decodes RESP2 replies.
    pub fn hello(&mut self, protocol: u8) -> Result<RespValue, Error> {
        self.run(args!["HELLO", protocol])
    }

    /// Sends `PUBLISH` and returns how many subscribers received the message.
    pub fn publish(&mut self, channel: &str, message: &str) -> Result<i64, Error> {
        self.integer(["PUBLISH", channel, message])
    }

    /// Sends `SPUBLISH` and returns how many subscribers received the message.
    pub fn spublish(&mut self, channel: &str, message: &str) -> Result<i64, Error> {
        self.integer(["SPUBLISH", channel, message])
    }

    /// Sends `SUBSCRIBE` and reads one confirmation per channel.
    /// Then read deliveries with [`Self::next_message`].
    pub fn subscribe(&mut self, channels: &[&str]) -> Result<Vec<RespValue>, Error> {
        self.subscription("SUBSCRIBE", channels, false)
    }

    /// Sends `UNSUBSCRIBE`. No channels drops every exact subscription on this connection.
    pub fn unsubscribe(&mut self, channels: &[&str]) -> Result<Vec<RespValue>, Error> {
        self.subscription("UNSUBSCRIBE", channels, true)
    }

    /// Sends `PSUBSCRIBE` and reads one confirmation per pattern.
    pub fn psubscribe(&mut self, patterns: &[&str]) -> Result<Vec<RespValue>, Error> {
        self.subscription("PSUBSCRIBE", patterns, false)
    }

    /// Sends `PUNSUBSCRIBE`. No patterns drops every pattern subscription on this connection.
    pub fn punsubscribe(&mut self, patterns: &[&str]) -> Result<Vec<RespValue>, Error> {
        self.subscription("PUNSUBSCRIBE", patterns, true)
    }

    /// Sends `SSUBSCRIBE` and reads one confirmation per channel.
    pub fn ssubscribe(&mut self, channels: &[&str]) -> Result<Vec<RespValue>, Error> {
        self.subscription("SSUBSCRIBE", channels, false)
    }

    /// Sends `SUNSUBSCRIBE`. No channels drops every sharded subscription on this connection.
    pub fn sunsubscribe(&mut self, channels: &[&str]) -> Result<Vec<RespValue>, Error> {
        self.subscription("SUNSUBSCRIBE", channels, true)
    }

    /// Reads replies until a `message`, `pmessage`, or `smessage` arrives, skipping
    /// subscribe confirmations. Blocks until [`Self::set_read_timeout`] passes.
    pub fn next_message(&mut self) -> Result<PubSubMessage, Error> {
        loop {
            let reply = self.read_message()?;

            if let Some(message) = pubsub_message(&reply)? {
                return Ok(message);
            }
        }
    }

    fn subscription(
        &mut self,
        command: &str,
        names: &[&str],
        allow_empty: bool,
    ) -> Result<Vec<RespValue>, Error> {
        if names.is_empty() && !allow_empty {
            return Err(Error::Protocol(format!(
                "{command} needs at least one channel or pattern"
            )));
        }

        self.run_replies(join(command, names), names.len().max(1))
    }

    fn optional_integer<I, S>(&mut self, arguments: I) -> Result<Option<i64>, Error>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        let (command, reply) = self.run_named(arguments)?;

        match reply {
            RespValue::Null => Ok(None),
            RespValue::Integer(value) => Ok(Some(value)),
            other => Err(unexpected(&command, &other)),
        }
    }

    fn blocking_pop(
        &mut self,
        command: &str,
        keys: &[&str],
        timeout: Duration,
    ) -> Result<Option<(String, String)>, Error> {
        let mut arguments = join(command, keys);

        arguments.push(duration_secs(timeout).to_string());

        match self.run(arguments)? {
            RespValue::Null => Ok(None),
            RespValue::Array(items) if items.len() == 2 => Ok(Some((
                items[0].as_string()?.unwrap_or_default(),
                items[1].as_string()?.unwrap_or_default(),
            ))),
            other => Err(unexpected(command, &other)),
        }
    }
}

fn with(mut arguments: Vec<String>, rest: &[&str]) -> Vec<String> {
    arguments.extend(rest.iter().map(|argument| (*argument).to_owned()));

    arguments
}

fn with_pairs(mut arguments: Vec<String>, pairs: &[(&str, &str)]) -> Vec<String> {
    for (left, right) in pairs {
        arguments.push((*left).to_owned());
        arguments.push((*right).to_owned());
    }

    arguments
}

fn push_scan_options(arguments: &mut Vec<String>, pattern: Option<&str>, count: Option<u64>) {
    if let Some(pattern) = pattern {
        arguments.extend(args!["MATCH", pattern]);
    }

    if let Some(count) = count {
        arguments.extend(args!["COUNT", count]);
    }
}

fn push_stream_read_options(arguments: &mut Vec<String>, options: &StreamReadOptions) {
    if let Some(count) = options.count {
        arguments.extend(args!["COUNT", count]);
    }

    if let Some(block) = options.block {
        arguments.extend(args!["BLOCK", duration_millis(block)]);
    }
}

fn push_streams(arguments: &mut Vec<String>, streams: &[(&str, &str)]) {
    arguments.push("STREAMS".to_owned());
    arguments.extend(streams.iter().map(|(key, _)| (*key).to_owned()));
    arguments.extend(streams.iter().map(|(_, id)| (*id).to_owned()));
}

fn zadd_arguments(key: &str, options: SortedSetAddOptions) -> Vec<String> {
    let mut arguments = args!["ZADD", key];
    let flags = [
        (options.if_not_exists, "NX"),
        (options.if_exists, "XX"),
        (options.greater_than, "GT"),
        (options.less_than, "LT"),
        (options.changed, "CH"),
    ];

    for (enabled, flag) in flags {
        if enabled {
            arguments.push(flag.to_owned());
        }
    }

    arguments
}

fn claim_arguments(
    key: &str,
    group: &str,
    consumer: &str,
    min_idle: Duration,
    ids: &[&str],
    options: &StreamClaimOptions,
) -> Result<Vec<String>, Error> {
    if options.idle.is_some() && options.time.is_some() {
        return Err(Error::Protocol(
            "IDLE and TIME cannot be combined".to_owned(),
        ));
    }

    let mut arguments = with(
        args!["XCLAIM", key, group, consumer, duration_millis(min_idle)],
        ids,
    );

    if let Some(idle) = options.idle {
        arguments.extend(args!["IDLE", duration_millis(idle)]);
    }

    if let Some(time) = options.time {
        arguments.extend(args!["TIME", unix_millis(time)?]);
    }

    if let Some(retry_count) = options.retry_count {
        arguments.extend(args!["RETRYCOUNT", retry_count]);
    }

    if options.force {
        arguments.push("FORCE".to_owned());
    }

    if let Some(last_id) = &options.last_id {
        arguments.extend(args!["LASTID", last_id]);
    }

    Ok(arguments)
}

fn script_arguments(command: &str, script: &str, keys: &[&str], arguments: &[&str]) -> Vec<String> {
    let values = with(args![command, script, keys.len()], keys);

    with(values, arguments)
}

fn parse_scan(command: &str, reply: RespValue) -> Result<(u64, Vec<String>), Error> {
    let Some([cursor, items]) = reply.as_array() else {
        return Err(unexpected(command, &reply));
    };

    let cursor = cursor
        .as_string()?
        .unwrap_or_default()
        .parse()
        .map_err(|_| Error::Protocol(format!("{command} returned a bad cursor")))?;
    let mut keys = Vec::new();

    for item in items.as_array().unwrap_or_default() {
        keys.push(item.as_string()?.unwrap_or_default());
    }

    Ok((cursor, keys))
}

fn parse_score(text: &str) -> Result<f64, Error> {
    text.parse()
        .map_err(|_| Error::Protocol(format!("{text:?} is not a score")))
}

fn pairs(reply: &RespValue) -> Result<Vec<(String, String)>, Error> {
    let items = reply.as_array().unwrap_or_default();
    let mut values = Vec::with_capacity(items.len() / 2);

    for pair in items.chunks_exact(2) {
        values.push((
            pair[0].as_string()?.unwrap_or_default(),
            pair[1].as_string()?.unwrap_or_default(),
        ));
    }

    Ok(values)
}

fn sorted_set(reply: &RespValue) -> Result<Vec<SortedSetEntry>, Error> {
    pairs(reply)?
        .into_iter()
        .map(|(member, score)| Ok(SortedSetEntry::new(member, parse_score(&score)?)))
        .collect()
}

fn stream_entries(reply: &RespValue) -> Result<Vec<StreamEntry>, Error> {
    let mut entries = Vec::new();

    for item in reply.as_array().unwrap_or_default() {
        let Some([id, fields]) = item.as_array() else {
            continue;
        };

        entries.push(StreamEntry {
            id: id.as_string()?.unwrap_or_default(),
            fields: pairs(fields)?,
        });
    }

    Ok(entries)
}

fn stream_read(reply: &RespValue) -> Result<Vec<StreamReadResult>, Error> {
    let mut results = Vec::new();

    for item in reply.as_array().unwrap_or_default() {
        let Some([key, entries]) = item.as_array() else {
            continue;
        };

        results.push(StreamReadResult {
            key: key.as_string()?.unwrap_or_default(),
            entries: stream_entries(entries)?,
        });
    }

    Ok(results)
}

fn pubsub_message(reply: &RespValue) -> Result<Option<PubSubMessage>, Error> {
    let items = reply.as_array().unwrap_or_default();
    let kind = match items.first() {
        Some(first) => first.as_string()?.unwrap_or_default(),
        None => return Ok(None),
    };

    let (pattern, channel, payload) = match (kind.as_str(), items) {
        ("message" | "smessage", [_, channel, payload]) => (None, channel, payload),
        ("pmessage", [_, pattern, channel, payload]) => (pattern.as_string()?, channel, payload),

        _ => return Ok(None),
    };

    let payload = match payload {
        RespValue::Bulk(bytes) => bytes.clone(),
        other => other.as_string()?.unwrap_or_default().into_bytes(),
    };

    Ok(Some(PubSubMessage {
        kind,
        pattern,
        channel: channel.as_string()?.unwrap_or_default(),
        payload,
    }))
}