yo-resp 0.3.25

//! The string commands, from the wire.
//!
//! Every body here parses its arguments, calls one method on [`Keyspace`], and
//! writes one reply. There is no logic about what a command means in this file,
//! because that logic is in `yo-kv` where the embedded API reaches it too (Y23).
//! What is here is the part that only the wire has: which keyword goes where,
//! which combinations Redis refuses, and which of the two protocols the answer
//! is spelled in.
//!
//! Two rules hold everywhere in this file.
//!
//! Nothing is written to the reply until the arguments are known to be good.
//! The dispatcher rolls the buffer back when a body returns an error, so a body
//! that broke this rule would be caught rather than corrupting the stream, but
//! the rule is what makes the rollback a safety net rather than the mechanism.
//!
//! Nothing allocates. The arguments are slices of the read buffer, the keywords
//! are compared in place, and the numbers are written straight into the reply.
//! A shard thread that allocates aborts, and the four commands M2 is measured
//! on all go through here.
//!
//! The option rules are Redis's, checked against a running 8.8 rather than read
//! off the documentation, because the documentation does not say that `SET k v
//! EX 5 EX 5` is accepted and `SET k v EX 5 PX 5` is not.

use super::args::{self, Args, is, syntax};
use super::notify::{self, class};
use super::table::Spec;
use crate::reply::Out;
use yo_common::num::{parse_f64, parse_i64};
use yo_common::{Code, Error, Result, xxh3};
use yo_kv::strings::check_len;
use yo_kv::{Compare, Db, Exists, Expire, IncrEx, IncrExpire, Keyspace, Num, SetOptions, Str};

/// What Redis says when a digest is not sixteen hexadecimal characters.
const BAD_DIGEST: &str = "must be exactly 16 hexadecimal characters";
/// What Redis says when `SETRANGE` is given a negative offset.
const BAD_OFFSET: &str = "offset is out of range";
/// What Redis says when `MSETEX` is given a count it cannot use.
const BAD_NUMKEYS: &str = "invalid numkeys value";
/// What Redis says when `MSETEX`'s count does not match what follows it.
const BAD_PAIRS: &str = "wrong number of key-value pairs";
/// What Redis says when `LCS` is asked for the length and the indexes at once.
const LEN_AND_IDX: &str = "If you want both the length and indexes, please just use IDX.";

/// Run one string command.
///
/// The name has already been looked up and the arity has already been checked,
/// so this matches on the table's own spelling of the name rather than on what
/// the client sent. The match is over twenty six short strings, which compiles
/// to a switch on the length and then a compare; the table grows to about two
/// hundred and fifty commands by M8 and this becomes a jump through an index
/// stored in the [`Spec`], which does not change any of the bodies.
pub(super) fn execute(
    db: &Db,
    on: usize,
    spec: &Spec,
    args: Args<'_>,
    out: &mut Out,
) -> Result<()> {
    // The five that name more than one key, taken before a stripe is chosen
    // because there is no one stripe to choose. Everything below this names
    // exactly one key and the table says it is at argument one.
    match spec.name {
        "mset" => return mset(db, on, args, out),
        "msetnx" => return msetnx(db, on, args, out),
        "mget" => return mget(db, args, out),
        "msetex" => return msetex(db, on, args, out),
        "lcs" => return lcs(db, args, out),
        _ => {}
    }
    let mut held = db.hold(args.get(1));
    let db = &mut *held;
    match spec.name {
        "get" => match db.get(args.get(1))? {
            Some(v) => write_str(out, v),
            None => out.nil(),
        },
        "set" => set(db, on, args, out)?,
        // Both of these have an owning form the embedded caller wants and a
        // `_with` form that hands the old value over where it lies. On the wire
        // the value is written into the reply and never looked at again, so the
        // owning form would be a malloc and a free per command.
        "getset" => {
            let mut had = false;
            db.set_with(
                args.get(1),
                args.get(2),
                SetOptions::PLAIN.returning(),
                |v| {
                    had = true;
                    write_str(out, v);
                },
            )?;
            if !had {
                out.nil();
            }
            // `GETSET` always stores, so there is no condition on this the way
            // there is on `SET`. It says `set` and not a name of its own.
            notify::fire(on, class::STRING, "set", args.get(1));
        }
        "getdel" => {
            if db.getdel_with(args.get(1), |v| write_str(out, v))? {
                notify::fire(on, class::GENERIC, "del", args.get(1));
            } else {
                out.nil();
            }
        }
        "getex" => getex(db, on, args, out)?,
        "setnx" => {
            let stored = db.setnx(args.get(1), args.get(2))?;
            if stored {
                notify::fire(on, class::STRING, "set", args.get(1));
            }
            out.int(i64::from(stored));
        }
        "setex" => {
            db.setex(args.get(1), args.int(2)?, args.get(3))?;
            timed_set(on, args.get(1));
            out.ok();
        }
        "psetex" => {
            db.psetex(args.get(1), args.int(2)?, args.get(3))?;
            timed_set(on, args.get(1));
            out.ok();
        }
        // Even `APPEND k ''`, which appends nothing and creates the key if it
        // was not there. Redis says `append` for that one too.
        "append" => {
            out.int(count(db.append(args.get(1), args.get(2))?));
            notify::fire(on, class::STRING, "append", args.get(1));
        }
        "strlen" => out.int(count(db.strlen(args.get(1))?)),
        "setrange" => {
            let offset =
                usize::try_from(args.int(2)?).map_err(|_| Error::new(Code::Invalid, BAD_OFFSET))?;
            out.int(count(db.setrange(args.get(1), offset, args.get(3))?));
            // Nothing to say when there was nothing to write. `SETRANGE k 5 ""`
            // answers the length and leaves the key exactly as it found it, so
            // a real server does not announce it and neither does this.
            if !args.get(3).is_empty() {
                notify::fire(on, class::STRING, "setrange", args.get(1));
            }
        }
        // `SUBSTR` is `GETRANGE` under the name it had before 2.0, and Redis
        // still ships both as separate entries in its table.
        "getrange" | "substr" => {
            let (start, end) = (args.int(2)?, args.int(3)?);
            out.bulk(&db.getrange(args.get(1), start, end)?);
        }
        // All four say `incrby`, including the two that take away. The event
        // is named after the one function underneath rather than after the
        // command, which is Redis's own wording and not a shortcut taken here.
        "incr" => {
            out.int(db.incr(args.get(1))?);
            notify::fire(on, class::STRING, "incrby", args.get(1));
        }
        "decr" => {
            out.int(db.decr(args.get(1))?);
            notify::fire(on, class::STRING, "incrby", args.get(1));
        }
        "incrby" => {
            out.int(db.incrby(args.get(1), args.int(2)?)?);
            notify::fire(on, class::STRING, "incrby", args.get(1));
        }
        "decrby" => {
            out.int(db.decrby(args.get(1), args.int(2)?)?);
            notify::fire(on, class::STRING, "incrby", args.get(1));
        }
        // A bulk string on both protocols, not a RESP3 double. Redis has never
        // changed this one and a client that parses the digits would break.
        "incrbyfloat" => {
            out.human_double(db.incrbyfloat(args.get(1), args.float(2)?)?);
            notify::fire(on, class::STRING, "incrbyfloat", args.get(1));
        }
        "delex" => delex(db, on, args, out)?,
        "digest" => match db.digest(args.get(1))? {
            Some(h) => out.bulk(&xxh3::hex(h)),
            None => out.nil(),
        },
        "increx" => increx(db, on, args, out)?,
        // Unreachable: the dispatcher only sends this function commands whose
        // group is `string`, and every one of those is above. An error rather
        // than a panic, because a table that has grown a row nobody wrote a
        // body for should be a command that does not work and not a server
        // that stops answering.
        _ => return Err(args::unknown_command(args)),
    }
    Ok(())
}

/// A stored value, as the bulk string the client is expecting.
///
/// An int encoded value never has its digits written down anywhere until
/// somebody asks for them, and this is where they get written.
fn write_str(out: &mut Out, v: Str<'_>) {
    match v {
        Str::Int(n) => out.bulk_int(n),
        Str::Bytes(b) => out.bulk(b),
    }
}

/// A length or a count as the integer the reply carries.
///
/// Saturating rather than wrapping. Nothing this counts can reach `i64::MAX`,
/// since a value is capped well below it, and a length that came back wrong is
/// better reported as an implausible number than as a negative one.
fn count(n: usize) -> i64 {
    i64::try_from(n).unwrap_or(i64::MAX)
}

/// How many key and value pairs `MSET` and `MSETNX` were given.
///
/// An even argument count means a value without its key, which Redis reports as
/// a wrong number of arguments even though the arity in its table is satisfied.
fn pair_count(args: Args<'_>, name: &str) -> Result<usize> {
    if args.len() % 2 != 1 {
        return Err(args::wrong_arity(name));
    }
    Ok((args.len() - 1) / 2)
}

/// `count` key and value pairs starting at argument `from`.
///
/// Borrowed straight out of the read buffer and never collected, which is what
/// [`Keyspace::mset`] takes an iterator for.
fn pairs<'a>(
    args: Args<'a>,
    from: usize,
    count: usize,
) -> impl Iterator<Item = (&'a [u8], &'a [u8])> + Clone {
    (0..count).map(move |i| (args.get(from + 2 * i), args.get(from + 2 * i + 1)))
}

/// A store that came with a deadline, as the two events it is.
///
/// `SETEX`, `PSETEX` and `SET ... EX` all say `set` and then `expire`, in that
/// order and on two different classes, because to a subscriber they are a write
/// followed by a deadline being put on what was written. `SET ... KEEPTTL` is
/// only the first of the two, since it did not touch the deadline.
fn timed_set(on: usize, key: &[u8]) {
    notify::fire(on, class::STRING, "set", key);
    notify::fire(on, class::GENERIC, "expire", key);
}

// ------------------------------------------------------------------ expiry

/// Which of the four expiration keywords was given.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Unit {
    /// `EX`, a number of seconds from now.
    Sec,
    /// `PX`, a number of milliseconds from now.
    Millis,
    /// `EXAT`, an absolute unix second.
    SecAt,
    /// `PXAT`, an absolute unix millisecond.
    MillisAt,
}

impl Unit {
    /// The keyword, or `None` if this argument is not one.
    fn parse(arg: &[u8]) -> Option<Unit> {
        if is(arg, b"EX") {
            Some(Unit::Sec)
        } else if is(arg, b"PX") {
            Some(Unit::Millis)
        } else if is(arg, b"EXAT") {
            Some(Unit::SecAt)
        } else if is(arg, b"PXAT") {
            Some(Unit::MillisAt)
        } else {
            None
        }
    }

    /// Whether the number is counted in seconds.
    fn is_seconds(self) -> bool {
        matches!(self, Unit::Sec | Unit::SecAt)
    }

    /// Whether the number is measured from now rather than from the epoch.
    fn is_relative(self) -> bool {
        matches!(self, Unit::Sec | Unit::Millis)
    }
}

/// The absolute unix millisecond an expiration option means.
///
/// Redis's range rule is one check over the number before it is scaled: zero or
/// less is refused, and a number of seconds that would not fit in a signed
/// millisecond is refused. A deadline in the past is not refused, it is
/// accepted and the key is gone immediately, which is how `SET k v EXAT 1`
/// behaves and is worth keeping since it is how a client deletes on a schedule.
fn deadline(unit: Unit, n: i64, now: u64, name: &str) -> Result<u64> {
    if n <= 0 || (unit.is_seconds() && n > i64::MAX / 1000) {
        return Err(args::invalid_expire(name));
    }
    let ms = if unit.is_seconds() { n * 1000 } else { n };
    // Both halves are positive and neither is anywhere near the top of the
    // range, so the sum cannot overflow: `now` is about 2^41 and `ms` is at
    // most 2^63 divided by a thousand.
    let ms = ms as u64;
    Ok(if unit.is_relative() { now + ms } else { ms })
}

// --------------------------------------------------------------------- SET

/// The option bits `SET` tracks, which are Redis's `OBJ_SET_*` flags.
///
/// A keyword given twice is accepted everywhere in `SET` and the last one wins,
/// which is why these are only ever tested against the keywords they conflict
/// with and never against themselves.
mod bits {
    /// `NX`.
    pub const NX: u16 = 1 << 0;
    /// `XX`.
    pub const XX: u16 = 1 << 1;
    /// `EX`.
    pub const EX: u16 = 1 << 2;
    /// `PX`.
    pub const PX: u16 = 1 << 3;
    /// `EXAT`.
    pub const EXAT: u16 = 1 << 4;
    /// `PXAT`.
    pub const PXAT: u16 = 1 << 5;
    /// `KEEPTTL`.
    pub const KEEPTTL: u16 = 1 << 6;
    /// `PERSIST`, which `GETEX` and `INCREX` take and `SET` does not.
    pub const PERSIST: u16 = 1 << 7;
    /// Any of the four `IF` conditions.
    pub const IF: u16 = 1 << 8;
    /// Every way of naming a deadline.
    pub const ANY_EXPIRE: u16 = EX | PX | EXAT | PXAT;
}

/// The bit an expiration keyword sets.
fn unit_bit(unit: Unit) -> u16 {
    match unit {
        Unit::Sec => bits::EX,
        Unit::Millis => bits::PX,
        Unit::SecAt => bits::EXAT,
        Unit::MillisAt => bits::PXAT,
    }
}

/// `SET key value [NX|XX] [GET] [EX s|PX ms|EXAT ts|PXAT ts|KEEPTTL]
/// [IFEQ v|IFNE v|IFDEQ d|IFDNE d]`.
fn set(db: &mut Keyspace, on: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
    let (key, val) = (args.get(1), args.get(2));
    let mut opts = SetOptions::PLAIN;
    let mut seen = 0u16;
    // The expiration and the condition are remembered as positions and read at
    // the end, because a keyword may be given twice and the last one wins, and
    // because Redis checks the whole option list for syntax before it looks at
    // any of the values.
    let mut expire: Option<(Unit, usize)> = None;
    let mut cond: Option<(&[u8], usize)> = None;
    let mut i = 3;
    while i < args.len() {
        let o = args.get(i);
        if is(o, b"NX") && seen & (bits::XX | bits::IF) == 0 {
            seen |= bits::NX;
            opts = opts.if_missing();
            i += 1;
        } else if is(o, b"XX") && seen & (bits::NX | bits::IF) == 0 {
            seen |= bits::XX;
            opts = opts.if_present();
            i += 1;
        } else if is(o, b"GET") {
            opts = opts.returning();
            i += 1;
        } else if is(o, b"KEEPTTL") && seen & bits::ANY_EXPIRE == 0 {
            seen |= bits::KEEPTTL;
            opts = opts.expiring(Expire::Keep);
            i += 1;
        } else if let Some(u) = Unit::parse(o)
            && seen & (bits::KEEPTTL | (bits::ANY_EXPIRE & !unit_bit(u))) == 0
            && i + 1 < args.len()
        {
            seen |= unit_bit(u);
            expire = Some((u, i + 1));
            i += 2;
        } else if is_condition(o)
            && seen & (bits::NX | bits::XX) == 0
            && cond.is_none_or(|(k, _)| is(o, k))
            && i + 1 < args.len()
        {
            seen |= bits::IF;
            cond = Some((o, i + 1));
            i += 2;
        } else {
            return Err(syntax());
        }
    }

    if let Some((u, at)) = expire {
        let ms = deadline(u, args.int(at)?, db.clock().now_ms(), "set")?;
        opts = opts.expiring(Expire::At(ms));
    }
    if let Some((keyword, at)) = cond {
        opts.compare = Some(condition(keyword, args.get(at))?);
    }

    // `set_with` and not `set`, so the old value goes straight into the reply
    // out of the record it is still sitting in. `set` would hand it back as a
    // `Vec` that this writes once and drops, which is a malloc and a free per
    // `SET ... GET`.
    let mut had = false;
    let done = db.set_with(key, val, opts, |v| {
        had = true;
        write_str(out, v);
    })?;
    // Said here rather than down among the three ways the reply is written,
    // because `SET ... GET` writes the old value and stores the new one and
    // both of those are true at once. A `SET` refused by `NX`, `XX` or one of
    // the compare and swap conditions stored nothing and says nothing.
    if done.stored {
        if expire.is_some() {
            timed_set(on, key);
        } else {
            notify::fire(on, class::STRING, "set", key);
        }
    }
    if opts.get {
        if !had {
            out.nil();
        }
    } else if done.stored {
        out.ok();
    } else {
        // `SET k v NX` on a key that is there is a null and not an error, on
        // both protocols. It is the RESP2 null bulk string, which is the one
        // `Out::nil` writes.
        out.nil();
    }
    Ok(())
}

/// Whether this argument is one of the four compare and swap keywords.
fn is_condition(arg: &[u8]) -> bool {
    is(arg, b"IFEQ") || is(arg, b"IFNE") || is(arg, b"IFDEQ") || is(arg, b"IFDNE")
}

/// The condition a keyword and its argument mean.
///
/// The digest forms take the sixteen hexadecimal characters `DIGEST` hands out
/// and nothing else, so a client that sends the digest as a number or with a
/// `0x` in front of it gets told exactly what is wrong with it.
fn condition<'a>(keyword: &[u8], arg: &'a [u8]) -> Result<Compare<'a>> {
    if is(keyword, b"IFEQ") {
        Ok(Compare::Equal(arg))
    } else if is(keyword, b"IFNE") {
        Ok(Compare::NotEqual(arg))
    } else {
        let d = xxh3::from_hex(arg).ok_or_else(|| Error::new(Code::Invalid, BAD_DIGEST))?;
        if is(keyword, b"IFDEQ") {
            Ok(Compare::DigestEqual(d))
        } else {
            Ok(Compare::DigestNotEqual(d))
        }
    }
}

// ------------------------------------------------------------------- GETEX

/// `GETEX key [EX s|PX ms|EXAT ts|PXAT ts|PERSIST]`.
fn getex(db: &mut Keyspace, on: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
    let key = args.get(1);
    let mut seen = 0u16;
    let mut expire: Option<(Unit, usize)> = None;
    let mut i = 2;
    while i < args.len() {
        let o = args.get(i);
        if is(o, b"PERSIST") && seen & bits::ANY_EXPIRE == 0 {
            seen |= bits::PERSIST;
            i += 1;
        } else if let Some(u) = Unit::parse(o)
            && seen & (bits::PERSIST | (bits::ANY_EXPIRE & !unit_bit(u))) == 0
            && i + 1 < args.len()
        {
            seen |= unit_bit(u);
            expire = Some((u, i + 1));
            i += 2;
        } else {
            return Err(syntax());
        }
    }

    // Redis looks the key up before it looks at the expiration value, so
    // `GETEX nosuch EX abc` and `GETEX nosuch EX 0` are both a null rather than
    // an error. The syntax of the option list is still checked first, which is
    // why that loop is above this and not below it.
    if !db.exists(key) {
        out.nil();
        return Ok(());
    }

    let wanted = match expire {
        Some((u, at)) => Expire::At(deadline(u, args.int(at)?, db.clock().now_ms(), "getex")?),
        None if seen & bits::PERSIST != 0 => Expire::Clear,
        None => Expire::Keep,
    };
    // Asked before the call, because afterwards there is no way to tell a
    // `PERSIST` that dropped a deadline from one that had nothing to drop, and
    // only the first of those says anything. The stripe is already held, so it
    // is a lookup and not a second acquisition.
    let had_deadline = wanted == Expire::Clear && matches!(db.deadline_of(key), yo_kv::Ask::At(_));
    match db.getex(key, wanted)? {
        Some(v) => write_str(out, v),
        None => out.nil(),
    }
    match wanted {
        Expire::At(_) => notify::fire(on, class::GENERIC, "expire", key),
        Expire::Clear if had_deadline => notify::fire(on, class::GENERIC, "persist", key),
        // Plain `GETEX` is a read and says nothing, and a `PERSIST` that found
        // no deadline changed nothing and says nothing either.
        _ => {}
    }
    Ok(())
}

// ------------------------------------------------------------------ MSETEX

/// `MSETEX numkeys key value [key value ...] [NX|XX]
/// [EX s|PX ms|EXAT ts|PXAT ts|KEEPTTL]`.
fn msetex(db: &Db, on: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
    let n = parse_i64(args.get(1))
        .filter(|&n| n > 0)
        .and_then(|n| usize::try_from(n).ok())
        .ok_or_else(|| Error::new(Code::Invalid, BAD_NUMKEYS))?;
    let end = n
        .checked_mul(2)
        .and_then(|pairs| pairs.checked_add(2))
        .filter(|&end| end <= args.len())
        .ok_or_else(|| Error::new(Code::Invalid, BAD_PAIRS))?;

    let mut seen = 0u16;
    let mut exists = Exists::Always;
    let mut expire = Expire::Clear;
    let mut at: Option<(Unit, usize)> = None;
    let mut i = end;
    while i < args.len() {
        let o = args.get(i);
        if is(o, b"NX") && seen & bits::XX == 0 {
            seen |= bits::NX;
            exists = Exists::IfMissing;
            i += 1;
        } else if is(o, b"XX") && seen & bits::NX == 0 {
            seen |= bits::XX;
            exists = Exists::IfPresent;
            i += 1;
        } else if is(o, b"KEEPTTL") && seen & bits::ANY_EXPIRE == 0 {
            seen |= bits::KEEPTTL;
            expire = Expire::Keep;
            i += 1;
        } else if let Some(u) = Unit::parse(o)
            && seen & (bits::KEEPTTL | (bits::ANY_EXPIRE & !unit_bit(u))) == 0
            && i + 1 < args.len()
        {
            seen |= unit_bit(u);
            at = Some((u, i + 1));
            i += 2;
        } else {
            return Err(syntax());
        }
    }
    if let Some((u, pos)) = at {
        expire = Expire::At(deadline(u, args.int(pos)?, db.now_ms(), "msetex")?);
    }

    // The same three passes [`Keyspace::msetex`] makes, for the same reason
    // [`mset`] makes them out here: the condition is over the whole set of keys
    // and it has to be decided before anything is written.
    for (k, v) in pairs(args, 2, n) {
        check_len(k, v.len())?;
    }
    // Every stripe the command names, held from before the condition is decided
    // until after the last pair is written, so that a key cannot appear or go
    // between being asked about and being written.
    let mut held = db.hold_keys(pairs(args, 2, n).map(|(k, _)| k));
    let mut asked = |k: &[u8]| held.stripe_mut(db.stripe_of(k)).exists(k);
    let allowed = match exists {
        Exists::Always => true,
        Exists::IfMissing => pairs(args, 2, n).all(|(k, _)| !asked(k)),
        Exists::IfPresent => pairs(args, 2, n).all(|(k, _)| asked(k)),
    };
    if !allowed {
        out.int(0);
        return Ok(());
    }
    // `Exists::Always` on each pair, because the condition has been decided
    // already and asking again per stripe would ask it of one key rather than
    // of all of them. `Expire::Keep` still has to go through, since what it
    // keeps is that key's own deadline and only its stripe knows it.
    for (k, v) in pairs(args, 2, n) {
        held.stripe_mut(db.stripe_of(k)).msetex(
            core::iter::once((k, v)),
            Exists::Always,
            expire,
        )?;
        // One key at a time and in the order they were given, which is what
        // `MSET` does too and what somebody watching several of these keys
        // would expect to see.
        //
        // The deadline comes first here, which is the other way round from
        // `SET ... EX` and from `SETEX`. That is measured against a real server
        // and not a slip: `MSETEX` is newer and its body sets the deadline on
        // the way past rather than after the write.
        if let Expire::At(_) = expire {
            notify::fire(on, class::GENERIC, "expire", k);
        }
        notify::fire(on, class::STRING, "set", k);
    }
    out.int(1);
    Ok(())
}

// -------------------------------------------------------- more than one key

/// `MSET key value [key value ...]`.
///
/// [`Keyspace::mset`] checks every pair before it writes any, so that a value
/// that is too long refuses the whole command rather than half of it. That
/// order has to be kept when the keys are on several stripes, and it cannot be
/// kept by handing each stripe its own pairs: the first stripe would write
/// before the second found the bad pair. So the check is a pass of its own out
/// here and the write is one pair at a time through the stripe that pair
/// belongs on, with every stripe the command names held for the whole of it so
/// that nobody sees half of an `MSET`.
fn mset(db: &Db, on: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
    let n = pair_count(args, "mset")?;
    for (k, v) in pairs(args, 1, n) {
        check_len(k, v.len())?;
    }
    let mut held = db.hold_keys(pairs(args, 1, n).map(|(k, _)| k));
    for (k, v) in pairs(args, 1, n) {
        held.stripe_mut(db.stripe_of(k))
            .mset(core::iter::once((k, v)))?;
        notify::fire(on, class::STRING, "set", k);
    }
    out.ok();
    Ok(())
}

/// `MSETNX key value [key value ...]`, which stores all of them or none.
///
/// Three passes for the same reason [`Keyspace::msetnx`] takes three: the
/// lengths, then whether any of the keys is already there, then the writes. The
/// middle pass is what makes a key named twice inside one call not defeat
/// itself, since nothing has been written when it runs.
fn msetnx(db: &Db, on: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
    let n = pair_count(args, "msetnx")?;
    for (k, v) in pairs(args, 1, n) {
        check_len(k, v.len())?;
    }
    let mut held = db.hold_keys(pairs(args, 1, n).map(|(k, _)| k));
    if pairs(args, 1, n).any(|(k, _)| held.stripe_mut(db.stripe_of(k)).exists(k)) {
        out.int(0);
        return Ok(());
    }
    for (k, v) in pairs(args, 1, n) {
        held.stripe_mut(db.stripe_of(k))
            .mset(core::iter::once((k, v)))?;
        notify::fire(on, class::STRING, "set", k);
    }
    out.int(1);
    Ok(())
}

/// `MGET key [key ...]`.
///
/// One key at a time rather than [`Keyspace::mget`], which collects the answers
/// into a `Vec` for a caller that wants them all at once. The wire wants them
/// one at a time and in order, and a `Vec` here would be an allocation per call
/// on a thread that must not.
fn mget(db: &Db, args: Args<'_>, out: &mut Out) -> Result<()> {
    // Every stripe at once, so the answer is these keys as they were at one
    // moment rather than each key as it was when the walk reached it.
    let mut held = db.hold_keys((1..args.len()).map(|i| args.get(i)));
    out.array(args.len() - 1);
    for i in 1..args.len() {
        let key = args.get(i);
        match held.stripe_mut(db.stripe_of(key)).mget_one(key) {
            Some(v) => write_str(out, v),
            None => out.nil(),
        }
    }
    Ok(())
}

// ------------------------------------------------------------------- DELEX

/// `DELEX key [IFEQ v|IFNE v|IFDEQ d|IFDNE d]`.
///
/// The arity in the table is a minimum of two, and a real server then refuses
/// anything that is not two or four arguments as a wrong number of them rather
/// than as a syntax error.
fn delex(db: &mut Keyspace, on: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
    if args.len() != 2 && args.len() != 4 {
        return Err(args::wrong_arity("delex"));
    }
    let compare = if args.len() == 4 {
        if !is_condition(args.get(2)) {
            return Err(syntax());
        }
        Some(condition(args.get(2), args.get(3))?)
    } else {
        None
    };
    let gone = db.delex(args.get(1), compare);
    if gone {
        // `del` and not a name of its own, for the reason `UNLINK` says `del`:
        // the event is what happened to the key.
        notify::fire(on, class::GENERIC, "del", args.get(1));
    }
    out.int(i64::from(gone));
    Ok(())
}

// ------------------------------------------------------------------ INCREX

/// `INCREX key [BYINT n|BYFLOAT f] [SATURATE] [LBOUND l] [UBOUND u]
/// [EX s|PX ms|EXAT ts|PXAT ts|PERSIST] [ENX]`.
///
/// Unlike `SET`, `INCREX` refuses a keyword it has already seen. It is a newer
/// command and it was written with a stricter parser, and a client that sends
/// `BYINT 1 BYINT 2` has a bug either way.
fn increx(db: &mut Keyspace, on: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
    /// `BYINT` or `BYFLOAT`.
    const BY: u16 = 1 << 9;
    /// `SATURATE`.
    const SATURATE: u16 = 1 << 10;
    /// `LBOUND`.
    const LBOUND: u16 = 1 << 11;
    /// `UBOUND`.
    const UBOUND: u16 = 1 << 12;
    /// `ENX`.
    const ENX: u16 = 1 << 13;

    let key = args.get(1);
    let mut seen = 0u16;
    let mut opts = IncrEx::PLAIN;
    // The bounds are read after the loop, because whether they are integers or
    // floats depends on a `BYFLOAT` that may not have been reached yet.
    let mut int_kind = true;
    let mut by: Option<usize> = None;
    let mut lower: Option<usize> = None;
    let mut upper: Option<usize> = None;
    let mut at: Option<(Unit, usize)> = None;
    let mut i = 2;
    while i < args.len() {
        let o = args.get(i);
        if (is(o, b"BYINT") || is(o, b"BYFLOAT")) && seen & BY == 0 && i + 1 < args.len() {
            seen |= BY;
            int_kind = is(o, b"BYINT");
            by = Some(i + 1);
            i += 2;
        } else if is(o, b"SATURATE") && seen & SATURATE == 0 {
            seen |= SATURATE;
            opts = opts.saturating();
            i += 1;
        } else if is(o, b"LBOUND") && seen & LBOUND == 0 && i + 1 < args.len() {
            seen |= LBOUND;
            lower = Some(i + 1);
            i += 2;
        } else if is(o, b"UBOUND") && seen & UBOUND == 0 && i + 1 < args.len() {
            seen |= UBOUND;
            upper = Some(i + 1);
            i += 2;
        } else if is(o, b"PERSIST") && seen & (bits::ANY_EXPIRE | bits::PERSIST) == 0 {
            seen |= bits::PERSIST;
            i += 1;
        } else if is(o, b"ENX") && seen & ENX == 0 {
            seen |= ENX;
            i += 1;
        } else if let Some(u) = Unit::parse(o)
            && seen & (bits::PERSIST | bits::ANY_EXPIRE) == 0
            && i + 1 < args.len()
        {
            seen |= unit_bit(u);
            at = Some((u, i + 1));
            i += 2;
        } else {
            return Err(syntax());
        }
    }
    if seen & ENX != 0 && at.is_none() {
        return Err(Error::new(Code::Invalid, "ENX flag requires an expiration"));
    }

    if let Some(pos) = by {
        opts = opts.by(number(args.get(pos), int_kind, "Increment")?);
    }
    let lower = lower
        .map(|pos| number(args.get(pos), int_kind, "LBOUND"))
        .transpose()?;
    let upper = upper
        .map(|pos| number(args.get(pos), int_kind, "UBOUND"))
        .transpose()?;
    opts = opts.between(lower, upper);
    if let Some((u, pos)) = at {
        let ms = deadline(u, args.int(pos)?, db.clock().now_ms(), "increx")?;
        opts = opts.expiring(if seen & ENX != 0 {
            IncrExpire::AtIfNone(ms)
        } else {
            IncrExpire::At(ms)
        });
    } else if seen & bits::PERSIST != 0 {
        opts = opts.expiring(IncrExpire::Persist);
    }

    // `ENX` only puts a deadline on a key that has none, so whether it did
    // anything cannot be read off the reply. Asked before the call and only
    // when somebody is listening, the same way `GETEX PERSIST` asks.
    let kept =
        seen & ENX != 0 && notify::armed() && matches!(db.deadline_of(key), yo_kv::Ask::At(_));
    let done = db.increx(key, opts)?;
    // The same two names `INCRBY` and `INCRBYFLOAT` use, chosen by the kind the
    // increment was counted in, and then the deadline if one was asked for. A
    // saturated increment that moved the value by nothing still moved it, so
    // there is no condition on this beyond the command having worked.
    notify::fire(
        on,
        class::STRING,
        if int_kind { "incrby" } else { "incrbyfloat" },
        key,
    );
    if at.is_some() && !kept {
        notify::fire(on, class::GENERIC, "expire", key);
    } else if seen & bits::PERSIST != 0 {
        notify::fire(on, class::GENERIC, "persist", key);
    }
    out.array(2);
    write_num(out, done.value);
    write_num(out, done.applied);
    Ok(())
}

/// An `INCREX` number, in the kind the increment is counted in.
///
/// The message names the option it came from, which is `Increment` for the
/// amount itself, because `ERR value is not an integer` in a command with four
/// numbers in it does not tell the client which one to look at.
fn number(arg: &[u8], int_kind: bool, what: &str) -> Result<Num> {
    if int_kind {
        parse_i64(arg).map(Num::Int).ok_or_else(|| {
            Error::fmt(
                Code::Invalid,
                format_args!("{what} is not an integer or out of range"),
            )
        })
    } else {
        parse_f64(arg)
            .map(Num::Float)
            .ok_or_else(|| Error::fmt(Code::Invalid, format_args!("{what} is not a valid float")))
    }
}

/// An `INCREX` number, as the client sees it.
///
/// The integer form is an integer on both protocols. The float form is a RESP3
/// double and a bulk string of the same digits on RESP2, which is what
/// [`Out::double`] already does.
fn write_num(out: &mut Out, n: Num) {
    match n {
        Num::Int(v) => out.int(v),
        Num::Float(v) => out.double(v),
    }
}

// --------------------------------------------------------------------- LCS

/// `LCS key1 key2 [LEN] [IDX] [MINMATCHLEN n] [WITHMATCHLEN]`.
///
/// `MINMATCHLEN` and `WITHMATCHLEN` without `IDX` are accepted and ignored,
/// which is what a real server does.
fn lcs(db: &Db, args: Args<'_>, out: &mut Out) -> Result<()> {
    let (a, b) = (args.get(1), args.get(2));
    let (mut want_len, mut want_idx, mut with_len) = (false, false, false);
    let mut minmatchlen = 0u32;
    let mut i = 3;
    while i < args.len() {
        let o = args.get(i);
        if is(o, b"LEN") {
            want_len = true;
            i += 1;
        } else if is(o, b"IDX") {
            want_idx = true;
            i += 1;
        } else if is(o, b"WITHMATCHLEN") {
            with_len = true;
            i += 1;
        } else if is(o, b"MINMATCHLEN") && i + 1 < args.len() {
            // A negative minimum is not an error, it is no minimum at all.
            minmatchlen = u32::try_from(args.int(i + 1)?.max(0)).unwrap_or(u32::MAX);
            i += 2;
        } else {
            return Err(syntax());
        }
    }
    if want_len && want_idx {
        return Err(Error::new(Code::Invalid, LEN_AND_IDX));
    }

    // Both values copied out before either is used, because the two keys can be
    // on two stripes and no one keyspace holds them both. That is what
    // [`Keyspace::lcs`] does inside itself anyway: it builds a table the size of
    // the product of the two lengths, so a pair of copies is not what makes it
    // expensive.
    // One at a time and not as a pair, because a tuple keeps everything in it
    // alive until the whole tuple is built, and the two stripes can be one.
    let x = db.hold(a).string_copy(a)?;
    let y = db.hold(b).string_copy(b)?;

    if want_idx {
        let idx = yo_kv::lcs::idx(&x, &y, minmatchlen)?;
        // A map on RESP3 and the same pairs flattened into an array on RESP2,
        // which is what `Out::map` writes and what a real server sends.
        out.map(2);
        out.bulk(b"matches");
        out.array(idx.matches.len());
        for m in &idx.matches {
            out.array(if with_len { 3 } else { 2 });
            out.array(2);
            out.int(i64::from(m.a.0));
            out.int(i64::from(m.a.1));
            out.array(2);
            out.int(i64::from(m.b.0));
            out.int(i64::from(m.b.1));
            if with_len {
                out.int(i64::from(m.len));
            }
        }
        out.bulk(b"len");
        out.int(count(idx.len));
    } else if want_len {
        out.int(count(yo_kv::lcs::len(&x, &y)?));
    } else {
        out.bulk(&yo_kv::lcs::string(&x, &y)?);
    }
    Ok(())
}