memcached-async 0.0.2

use bytes::{Buf, BufMut, Bytes, BytesMut};

use crate::codec::DecodeOutcome;
use crate::error::{Error, ErrorKind};
use crate::response::Response;
use crate::types::{
    MetaCode, MetaFlags, MetaResponse, Op, Protocol, ReplyMode, Request, RequestMeta, StatLine,
    ValueEntry,
};

/// ASCII/meta decode limits.
#[derive(Debug, Clone, Copy)]
pub struct AsciiLimits {
    pub max_line_len: usize,
    pub max_blob_len: usize,
}

impl Default for AsciiLimits {
    fn default() -> Self {
        Self {
            max_line_len: 4096,
            max_blob_len: 1 << 20,
        }
    }
}

#[derive(Debug, Clone)]
struct BlobState {
    len: usize,
    req: Request,
    meta: RequestMeta,
}

#[derive(Debug, Clone)]
enum AsciiState {
    Line,
    Blob(Box<BlobState>),
}

/// Streaming ASCII/meta decoder.
#[derive(Debug, Clone)]
pub struct AsciiDecoder {
    state: AsciiState,
    discarding_line: bool,
}

impl Default for AsciiDecoder {
    fn default() -> Self {
        Self::new()
    }
}

impl AsciiDecoder {
    pub fn new() -> Self {
        Self {
            state: AsciiState::Line,
            discarding_line: false,
        }
    }

    pub fn decode(&mut self, buf: &mut BytesMut, limits: AsciiLimits) -> Option<DecodeOutcome> {
        let state = std::mem::replace(&mut self.state, AsciiState::Line);
        match state {
            AsciiState::Line => {
                self.state = AsciiState::Line;
                self.decode_line(buf, limits)
            }
            AsciiState::Blob(blob) => {
                let BlobState { len, mut req, meta } = *blob;
                if buf.len() < len + 2 {
                    self.state = AsciiState::Blob(Box::new(BlobState { len, req, meta }));
                    return None;
                }
                let payload = buf.split_to(len).freeze();
                let cr = buf.get_u8();
                let lf = buf.get_u8();
                if cr != b'\r' || lf != b'\n' {
                    let response = Response::Error(Error::client("bad data chunk"));
                    self.state = AsciiState::Line;
                    return Some(DecodeOutcome::Response(meta, response));
                }
                req.value = Some(payload);
                self.state = AsciiState::Line;
                Some(DecodeOutcome::Request(req, meta))
            }
        }
    }

    fn decode_line(&mut self, buf: &mut BytesMut, limits: AsciiLimits) -> Option<DecodeOutcome> {
        if self.discarding_line {
            if let Some(pos) = find_crlf(buf) {
                buf.advance(pos + 2);
                self.discarding_line = false;
                let meta = RequestMeta::ascii();
                let response = Response::Error(Error::client("line too long"));
                return Some(DecodeOutcome::Response(meta, response));
            }
            if buf.len() > limits.max_line_len {
                buf.clear();
            }
            return None;
        }

        let pos = match find_crlf(buf) {
            Some(pos) => pos,
            None => {
                if buf.len() > limits.max_line_len {
                    self.discarding_line = true;
                }
                return None;
            }
        };

        if pos > limits.max_line_len {
            buf.advance(pos + 2);
            let meta = RequestMeta::ascii();
            let response = Response::Error(Error::client("line too long"));
            return Some(DecodeOutcome::Response(meta, response));
        }

        let line = buf.split_to(pos).freeze();
        buf.advance(2);

        if line.is_empty() {
            let meta = RequestMeta::ascii();
            let response = Response::Error(Error::client("empty command"));
            return Some(DecodeOutcome::Response(meta, response));
        }

        match parse_line(line, limits.max_blob_len) {
            Ok(LineParse::Line(req, meta)) => Some(DecodeOutcome::Request(req, meta)),
            Ok(LineParse::Blob { len, mut req, meta }) => {
                if buf.len() >= len + 2 {
                    let payload = buf.split_to(len).freeze();
                    let cr = buf.get_u8();
                    let lf = buf.get_u8();
                    if cr != b'\r' || lf != b'\n' {
                        let response = Response::Error(Error::client("bad data chunk"));
                        return Some(DecodeOutcome::Response(meta, response));
                    }
                    req.value = Some(payload);
                    return Some(DecodeOutcome::Request(req, meta));
                }
                self.state = AsciiState::Blob(Box::new(BlobState { len, req, meta }));
                None
            }
            Err(err) => Some(DecodeOutcome::Response(
                RequestMeta::ascii(),
                Response::Error(err),
            )),
        }
    }
}

enum LineParse {
    Line(Request, RequestMeta),
    Blob {
        len: usize,
        req: Request,
        meta: RequestMeta,
    },
}

fn parse_line(line: Bytes, max_blob_len: usize) -> Result<LineParse, Error> {
    let tokens = split_tokens(&line);
    if tokens.is_empty() {
        return Err(Error::client("empty command"));
    }

    let cmd = tokens[0].as_ref();
    let op = parse_op(cmd);
    let mut meta = RequestMeta::ascii();
    let mut req = Request::new(op);

    match op {
        Op::Get | Op::Gets => {
            if tokens.len() < 2 {
                return Err(Error::client("missing key"));
            }
            let keys = parse_keys(&tokens[1..])?;
            if keys.len() == 1 {
                req.key = Some(keys[0].clone());
            } else {
                req.keys = keys;
            }
        }
        Op::Gat | Op::Gats => {
            if tokens.len() < 3 {
                return Err(Error::client("missing exptime or key"));
            }
            let exptime =
                parse_i64(tokens[1].as_ref()).ok_or_else(|| Error::client("invalid exptime"))?;
            req.exptime = Some(exptime);
            let keys = parse_keys(&tokens[2..])?;
            if keys.len() == 1 {
                req.key = Some(keys[0].clone());
            } else {
                req.keys = keys;
            }
        }
        Op::Set | Op::Add | Op::Replace | Op::Append | Op::Prepend | Op::Cas => {
            let min = if matches!(op, Op::Cas) { 6 } else { 5 };
            if tokens.len() < min {
                return Err(Error::client("missing arguments"));
            }
            let mut end = tokens.len();
            let mut reply = ReplyMode::Always;
            if tokens.len() > min && tokens.last().map(|t| t.as_ref()) == Some(b"noreply") {
                reply = ReplyMode::SuppressSuccess;
                end -= 1;
            }
            if end != min {
                return Err(Error::client("invalid arguments"));
            }
            let key = tokens[1].clone();
            validate_key(&key)?;
            req.key = Some(key);
            req.flags =
                Some(parse_u32(tokens[2].as_ref()).ok_or_else(|| Error::client("invalid flags"))?);
            req.exptime = Some(
                parse_i64(tokens[3].as_ref()).ok_or_else(|| Error::client("invalid exptime"))?,
            );
            let bytes =
                parse_usize(tokens[4].as_ref()).ok_or_else(|| Error::client("invalid bytes"))?;
            if bytes > max_blob_len {
                return Err(Error::server("value too large"));
            }
            if matches!(op, Op::Cas) {
                req.cas = Some(
                    parse_u64(tokens[5].as_ref()).ok_or_else(|| Error::client("invalid cas"))?,
                );
            }
            meta.reply = reply;
            return Ok(LineParse::Blob {
                len: bytes,
                req,
                meta,
            });
        }
        Op::Delete => {
            if tokens.len() < 2 {
                return Err(Error::client("missing key"));
            }
            let mut reply = ReplyMode::Always;
            if tokens.len() > 2 && tokens.last().map(|t| t.as_ref()) == Some(b"noreply") {
                reply = ReplyMode::SuppressSuccess;
                if tokens.len() != 3 {
                    return Err(Error::client("invalid arguments"));
                }
            } else if tokens.len() != 2 {
                return Err(Error::client("invalid arguments"));
            }
            let key = tokens[1].clone();
            validate_key(&key)?;
            req.key = Some(key);
            meta.reply = reply;
        }
        Op::Flush => {
            let mut reply = ReplyMode::Always;
            let mut delay: Option<i64> = None;
            match tokens.len() {
                1 => {}
                2 => {
                    if tokens[1].as_ref() == b"noreply" {
                        reply = ReplyMode::SuppressSuccess;
                    } else {
                        delay = Some(
                            parse_i64(tokens[1].as_ref())
                                .ok_or_else(|| Error::client("invalid exptime"))?,
                        );
                    }
                }
                3 => {
                    if tokens[2].as_ref() != b"noreply" {
                        return Err(Error::client("invalid arguments"));
                    }
                    reply = ReplyMode::SuppressSuccess;
                    delay = Some(
                        parse_i64(tokens[1].as_ref())
                            .ok_or_else(|| Error::client("invalid exptime"))?,
                    );
                }
                _ => return Err(Error::client("invalid arguments")),
            }
            if let Some(value) = delay {
                if value < 0 {
                    return Err(Error::client("invalid exptime"));
                }
                req.exptime = Some(value);
            }
            meta.reply = reply;
        }
        Op::Incr | Op::Decr => {
            if tokens.len() < 3 {
                return Err(Error::client("missing arguments"));
            }
            let mut reply = ReplyMode::Always;
            if tokens.len() > 3 && tokens.last().map(|t| t.as_ref()) == Some(b"noreply") {
                reply = ReplyMode::SuppressSuccess;
                if tokens.len() != 4 {
                    return Err(Error::client("invalid arguments"));
                }
            } else if tokens.len() != 3 {
                return Err(Error::client("invalid arguments"));
            }
            let key = tokens[1].clone();
            validate_key(&key)?;
            req.key = Some(key);
            req.delta =
                Some(parse_u64(tokens[2].as_ref()).ok_or_else(|| Error::client("invalid delta"))?);
            meta.reply = reply;
        }
        Op::Touch => {
            if tokens.len() < 3 {
                return Err(Error::client("missing arguments"));
            }
            let mut reply = ReplyMode::Always;
            if tokens.len() > 3 && tokens.last().map(|t| t.as_ref()) == Some(b"noreply") {
                reply = ReplyMode::SuppressSuccess;
                if tokens.len() != 4 {
                    return Err(Error::client("invalid arguments"));
                }
            } else if tokens.len() != 3 {
                return Err(Error::client("invalid arguments"));
            }
            let key = tokens[1].clone();
            validate_key(&key)?;
            req.key = Some(key);
            req.exptime = Some(
                parse_i64(tokens[2].as_ref()).ok_or_else(|| Error::client("invalid exptime"))?,
            );
            meta.reply = reply;
        }
        Op::Stats => {
            if tokens.len() > 1 {
                let keys = parse_keys(&tokens[1..])?;
                req.keys = keys;
            }
        }
        Op::Version | Op::Quit => {
            if tokens.len() != 1 {
                return Err(Error::client("invalid arguments"));
            }
        }
        Op::MetaGet
        | Op::MetaSet
        | Op::MetaDelete
        | Op::MetaArithmetic
        | Op::MetaDebug
        | Op::MetaNoop => {
            meta.protocol = Protocol::Meta;
            if op == Op::MetaNoop {
                if tokens.len() != 1 {
                    return Err(Error::client("invalid arguments"));
                }
                return Ok(LineParse::Line(req, meta));
            }
            if tokens.len() < 2 {
                return Err(Error::client("missing key"));
            }
            let mut idx = 2;
            let mut datalen = None;
            if op == Op::MetaSet {
                if tokens.len() < 3 {
                    return Err(Error::client("missing data length"));
                }
                datalen = Some(
                    parse_usize(tokens[2].as_ref())
                        .ok_or_else(|| Error::client("invalid data length"))?,
                );
                idx = 3;
            }
            let meta_flags = parse_meta_flags(&tokens[idx..]);
            if meta_flags.has(b'q') {
                meta.reply = if op == Op::MetaGet {
                    ReplyMode::SuppressMiss
                } else {
                    ReplyMode::SuppressSuccess
                };
            }
            let mut key = tokens[1].clone();
            if meta_flags.has(b'b') {
                let decoded = decode_base64(key.as_ref())?;
                if decoded.len() > 250 {
                    return Err(Error::client("key too long"));
                }
                key = Bytes::from(decoded);
            } else {
                validate_key(&key)?;
            }
            req.key = Some(key);
            if let Some(delta) = meta_flags
                .token(b'D')
                .and_then(|token| parse_u64(token.as_ref()))
            {
                req.delta = Some(delta);
            }
            if let Some(initial) = meta_flags
                .token(b'J')
                .and_then(|token| parse_u64(token.as_ref()))
            {
                req.initial = Some(initial);
            }
            if let Some(cas) = meta_flags
                .token(b'C')
                .and_then(|token| parse_u64(token.as_ref()))
            {
                req.cas = Some(cas);
            }
            req.meta = Some(meta_flags.clone());
            if op == Op::MetaSet {
                let mut length = datalen.unwrap_or(0);
                if let Some(slen) = meta_flags
                    .token(b'S')
                    .and_then(|token| parse_usize(token.as_ref()))
                {
                    length = slen;
                }
                if length > max_blob_len {
                    return Err(Error::server("value too large"));
                }
                if length == 0 {
                    return Ok(LineParse::Line(req, meta));
                }
                return Ok(LineParse::Blob {
                    len: length,
                    req,
                    meta,
                });
            }
        }
        Op::SaslListMechs | Op::SaslAuth | Op::SaslStep | Op::Unknown => {
            req.op = Op::Unknown;
        }
        Op::Noop => {}
    }

    Ok(LineParse::Line(req, meta))
}

fn parse_op(cmd: &[u8]) -> Op {
    match cmd {
        b"get" => Op::Get,
        b"gets" => Op::Gets,
        b"gat" => Op::Gat,
        b"gats" => Op::Gats,
        b"set" => Op::Set,
        b"add" => Op::Add,
        b"replace" => Op::Replace,
        b"append" => Op::Append,
        b"prepend" => Op::Prepend,
        b"cas" => Op::Cas,
        b"delete" => Op::Delete,
        b"flush_all" => Op::Flush,
        b"incr" => Op::Incr,
        b"decr" => Op::Decr,
        b"touch" => Op::Touch,
        b"stats" => Op::Stats,
        b"version" => Op::Version,
        b"quit" => Op::Quit,
        b"mg" => Op::MetaGet,
        b"ms" => Op::MetaSet,
        b"md" => Op::MetaDelete,
        b"ma" => Op::MetaArithmetic,
        b"me" => Op::MetaDebug,
        b"mn" => Op::MetaNoop,
        _ => Op::Unknown,
    }
}

fn parse_keys(tokens: &[Bytes]) -> Result<Vec<Bytes>, Error> {
    let mut keys = Vec::with_capacity(tokens.len());
    for token in tokens {
        validate_key(token)?;
        keys.push(token.clone());
    }
    Ok(keys)
}

fn validate_key(key: &Bytes) -> Result<(), Error> {
    if key.is_empty() {
        return Err(Error::client("empty key"));
    }
    if key.len() > 250 {
        return Err(Error::client("key too long"));
    }
    for &b in key.as_ref() {
        if b <= b' ' || b == 0x7f {
            return Err(Error::client("invalid key"));
        }
    }
    Ok(())
}

fn split_tokens(line: &Bytes) -> Vec<Bytes> {
    let mut tokens = Vec::new();
    let mut start = 0;
    let bytes = line.as_ref();
    while start < bytes.len() {
        while start < bytes.len() && bytes[start] == b' ' {
            start += 1;
        }
        if start >= bytes.len() {
            break;
        }
        let mut end = start;
        while end < bytes.len() && bytes[end] != b' ' {
            end += 1;
        }
        tokens.push(line.slice(start..end));
        start = end;
    }
    tokens
}

fn find_crlf(buf: &BytesMut) -> Option<usize> {
    let bytes = buf.as_ref();
    if bytes.len() < 2 {
        return None;
    }
    let mut i = 0;
    while i + 1 < bytes.len() {
        if bytes[i] == b'\r' && bytes[i + 1] == b'\n' {
            return Some(i);
        }
        i += 1;
    }
    None
}

fn parse_u32(token: &[u8]) -> Option<u32> {
    parse_u64(token).and_then(|value| u32::try_from(value).ok())
}

fn parse_usize(token: &[u8]) -> Option<usize> {
    parse_u64(token).and_then(|value| usize::try_from(value).ok())
}

fn parse_u64(token: &[u8]) -> Option<u64> {
    if token.is_empty() {
        return None;
    }
    let mut value: u64 = 0;
    for &b in token {
        if !b.is_ascii_digit() {
            return None;
        }
        value = value.checked_mul(10)?;
        value = value.checked_add((b - b'0') as u64)?;
    }
    Some(value)
}

fn parse_i64(token: &[u8]) -> Option<i64> {
    if token.is_empty() {
        return None;
    }
    let (neg, rest) = if token[0] == b'-' {
        (true, &token[1..])
    } else {
        (false, token)
    };
    let value = parse_u64(rest)? as i64;
    if neg { Some(-value) } else { Some(value) }
}

fn parse_meta_flags(tokens: &[Bytes]) -> MetaFlags {
    let mut ordered = Vec::with_capacity(tokens.len());
    for token in tokens {
        if token.is_empty() {
            continue;
        }
        let code = token.as_ref()[0];
        let rest = if token.len() > 1 {
            Some(token.slice(1..))
        } else {
            None
        };
        ordered.push(crate::types::MetaFlag { code, token: rest });
    }
    MetaFlags::new(ordered)
}

fn decode_base64(input: &[u8]) -> Result<Vec<u8>, Error> {
    if !input.len().is_multiple_of(4) {
        return Err(Error::client("invalid base64"));
    }
    let mut out = Vec::with_capacity(input.len() / 4 * 3);
    let mut i = 0;
    while i < input.len() {
        let a = decode_base64_val(input[i])?;
        let b = decode_base64_val(input[i + 1])?;
        let c = input[i + 2];
        let d = input[i + 3];
        let c_val = if c == b'=' {
            None
        } else {
            Some(decode_base64_val(c)?)
        };
        let d_val = if d == b'=' {
            None
        } else {
            Some(decode_base64_val(d)?)
        };

        out.push((a << 2) | (b >> 4));
        if let Some(c_val) = c_val {
            out.push((b << 4) | (c_val >> 2));
            if let Some(d_val) = d_val {
                out.push((c_val << 6) | d_val);
            }
        }
        i += 4;
    }
    Ok(out)
}

fn decode_base64_val(byte: u8) -> Result<u8, Error> {
    match byte {
        b'A'..=b'Z' => Ok(byte - b'A'),
        b'a'..=b'z' => Ok(byte - b'a' + 26),
        b'0'..=b'9' => Ok(byte - b'0' + 52),
        b'+' => Ok(62),
        b'/' => Ok(63),
        b'=' => Ok(0),
        _ => Err(Error::client("invalid base64")),
    }
}

pub fn should_suppress_ascii(meta: RequestMeta, response: &Response) -> bool {
    match meta.reply {
        ReplyMode::Always => false,
        ReplyMode::SuppressMiss => {
            if let Response::Meta(meta_resp) = response {
                meta_resp.code.is_miss()
            } else {
                false
            }
        }
        ReplyMode::SuppressSuccess => !matches!(response, Response::Error(_)),
        ReplyMode::QuietBuffered => false,
    }
}

pub fn should_suppress_meta(meta: RequestMeta, response: &Response) -> bool {
    match meta.reply {
        ReplyMode::Always => false,
        ReplyMode::SuppressMiss => {
            if let Response::Meta(meta_resp) = response {
                meta_resp.code.is_miss()
            } else {
                false
            }
        }
        ReplyMode::SuppressSuccess => {
            if let Response::Meta(meta_resp) = response {
                meta_resp.code.is_success()
            } else {
                false
            }
        }
        ReplyMode::QuietBuffered => false,
    }
}

pub fn encode_ascii_response(
    req: &Request,
    meta: RequestMeta,
    response: &Response,
    out: &mut BytesMut,
) -> bool {
    if should_suppress_ascii(meta, response) {
        return false;
    }
    match response {
        Response::Stored => out.extend_from_slice(b"STORED\r\n"),
        Response::NotStored => out.extend_from_slice(b"NOT_STORED\r\n"),
        Response::Exists => out.extend_from_slice(b"EXISTS\r\n"),
        Response::NotFound => out.extend_from_slice(b"NOT_FOUND\r\n"),
        Response::Deleted => out.extend_from_slice(b"DELETED\r\n"),
        Response::Touched => out.extend_from_slice(b"TOUCHED\r\n"),
        Response::Ok => out.extend_from_slice(b"OK\r\n"),
        Response::Numeric(value) => {
            write_u64(out, *value);
            out.extend_from_slice(b"\r\n");
        }
        Response::Value(entry) => {
            let include_cas = matches!(req.op, Op::Gets | Op::Gats);
            encode_value_entry(entry, include_cas, out);
            out.extend_from_slice(b"END\r\n");
        }
        Response::Values(entries) => {
            let include_cas = matches!(req.op, Op::Gets | Op::Gats);
            for entry in entries {
                encode_value_entry(entry, include_cas, out);
            }
            out.extend_from_slice(b"END\r\n");
        }
        Response::Stats(lines) => {
            for line in lines {
                encode_stat_line(line, out);
            }
            out.extend_from_slice(b"END\r\n");
        }
        Response::Version(version) => {
            out.extend_from_slice(b"VERSION ");
            out.extend_from_slice(version);
            out.extend_from_slice(b"\r\n");
        }
        Response::Noop => {
            if req.op == Op::MetaNoop {
                out.extend_from_slice(b"MN\r\n");
            }
        }
        Response::Error(err) => match err.kind {
            ErrorKind::UnknownCommand => out.extend_from_slice(b"ERROR\r\n"),
            ErrorKind::Client | ErrorKind::Auth => {
                out.extend_from_slice(b"CLIENT_ERROR ");
                out.extend_from_slice(err.message.as_ref());
                out.extend_from_slice(b"\r\n");
            }
            ErrorKind::Server => {
                out.extend_from_slice(b"SERVER_ERROR ");
                out.extend_from_slice(err.message.as_ref());
                out.extend_from_slice(b"\r\n");
            }
        },
        Response::Meta(_) | Response::ValuesStream(_) | Response::StatsStream(_) => {}
    }
    true
}

pub fn encode_meta_response(
    req: &Request,
    meta: RequestMeta,
    response: &MetaResponse,
    out: &mut BytesMut,
) {
    if should_suppress_meta(meta, &Response::Meta(response.clone())) {
        return;
    }
    out.extend_from_slice(response.code.as_bytes());
    if response.code == MetaCode::Va {
        out.extend_from_slice(b" ");
        let size = response
            .size
            .or_else(|| response.value.as_ref().map(|value| value.len()))
            .unwrap_or(0);
        write_usize(out, size);
    }

    if let Some(flags) = req.meta.as_ref() {
        for flag in &flags.ordered {
            append_meta_token(req, response, flag.code, out);
        }
    }

    if let Some(win) = response.extra.won {
        out.extend_from_slice(match win {
            crate::types::WinState::Won => b" W",
            crate::types::WinState::AlreadyWon => b" Z",
        });
    }
    if response.extra.stale {
        out.extend_from_slice(b" X");
    }

    out.extend_from_slice(b"\r\n");

    let want_value = req
        .meta
        .as_ref()
        .map(|flags| flags.has(b'v'))
        .unwrap_or(false);
    if want_value && let Some(value) = response.value.as_ref() {
        out.extend_from_slice(value);
        out.extend_from_slice(b"\r\n");
    }
}

pub fn encode_meta_debug(
    req: &Request,
    lines: impl IntoIterator<Item = StatLine>,
    out: &mut BytesMut,
) {
    let key = match req.key.as_ref() {
        Some(key) => key.as_ref(),
        None => b"",
    };
    out.extend_from_slice(b"ME ");
    out.extend_from_slice(key);
    for line in lines {
        out.extend_from_slice(b" ");
        out.extend_from_slice(line.key.as_ref());
        out.extend_from_slice(b"=");
        out.extend_from_slice(line.value.as_ref());
    }
    out.extend_from_slice(b"\r\n");
}

pub fn encode_value_entry(entry: &ValueEntry, include_cas: bool, out: &mut BytesMut) {
    out.extend_from_slice(b"VALUE ");
    out.extend_from_slice(entry.key.as_ref());
    out.extend_from_slice(b" ");
    write_u32(out, entry.flags);
    out.extend_from_slice(b" ");
    write_usize(out, entry.value.len());
    if include_cas {
        out.extend_from_slice(b" ");
        write_u64(out, entry.cas.unwrap_or(0));
    }
    out.extend_from_slice(b"\r\n");
    out.extend_from_slice(entry.value.as_ref());
    out.extend_from_slice(b"\r\n");
}

pub fn encode_stat_line(line: &StatLine, out: &mut BytesMut) {
    out.extend_from_slice(b"STAT ");
    out.extend_from_slice(line.key.as_ref());
    out.extend_from_slice(b" ");
    out.extend_from_slice(line.value.as_ref());
    out.extend_from_slice(b"\r\n");
}

fn append_meta_token(req: &Request, response: &MetaResponse, code: u8, out: &mut BytesMut) {
    match code {
        b'O' => {
            if let Some(token) = req.meta.as_ref().and_then(|flags| flags.token(b'O')) {
                out.extend_from_slice(b" O");
                out.extend_from_slice(token.as_ref());
            }
        }
        b'k' => {
            if let Some(key) = req.key.as_ref() {
                out.extend_from_slice(b" k");
                out.extend_from_slice(key.as_ref());
            }
        }
        b'c' => {
            if let Some(cas) = response.cas {
                out.extend_from_slice(b" c");
                write_u64(out, cas);
            }
        }
        b't' => {
            if let Some(ttl) = response.ttl {
                out.extend_from_slice(b" t");
                write_i64(out, ttl);
            }
        }
        b'f' => {
            if let Some(flags) = response.flags {
                out.extend_from_slice(b" f");
                write_u32(out, flags);
            }
        }
        b's' => {
            if let Some(size) = response.size {
                out.extend_from_slice(b" s");
                write_usize(out, size);
            }
        }
        b'h' => {
            if let Some(hit) = response.hit {
                out.extend_from_slice(b" h");
                out.extend_from_slice(if hit { b"1" } else { b"0" });
            }
        }
        b'l' => {
            if let Some(last) = response.last_access {
                out.extend_from_slice(b" l");
                write_u64(out, last);
            }
        }
        _ => {}
    }
}

fn write_u32(out: &mut BytesMut, value: u32) {
    write_u64(out, value as u64)
}

fn write_usize(out: &mut BytesMut, value: usize) {
    write_u64(out, value as u64)
}

fn write_u64(out: &mut BytesMut, mut value: u64) {
    let mut buf = [0u8; 20];
    let mut i = buf.len();
    if value == 0 {
        out.put_u8(b'0');
        return;
    }
    while value > 0 {
        i -= 1;
        buf[i] = b'0' + (value % 10) as u8;
        value /= 10;
    }
    out.extend_from_slice(&buf[i..]);
}

fn write_i64(out: &mut BytesMut, value: i64) {
    if value < 0 {
        out.put_u8(b'-');
        write_u64(out, (-value) as u64);
    } else {
        write_u64(out, value as u64);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use bytes::BytesMut;

    #[test]
    fn decode_set_with_value() {
        let mut decoder = AsciiDecoder::new();
        let mut buf = BytesMut::from("set key 1 10 5\r\nhello\r\n");
        let limits = AsciiLimits::default();

        let outcome = decoder.decode(&mut buf, limits);
        let (req, meta) = match outcome {
            Some(DecodeOutcome::Request(req, meta)) => (req, meta),
            _ => panic!("unexpected decode outcome"),
        };
        assert_eq!(req.op, Op::Set);
        assert_eq!(req.value.unwrap(), Bytes::from_static(b"hello"));
        assert_eq!(meta.reply, ReplyMode::Always);
    }

    #[test]
    fn decode_multi_get() {
        let mut decoder = AsciiDecoder::new();
        let mut buf = BytesMut::from("get k1 k2\r\n");
        let outcome = decoder.decode(&mut buf, AsciiLimits::default());
        let (req, _) = match outcome {
            Some(DecodeOutcome::Request(req, meta)) => (req, meta),
            _ => panic!("unexpected decode outcome"),
        };
        assert_eq!(req.op, Op::Get);
        assert_eq!(req.keys.len(), 2);
    }

    #[test]
    fn decode_meta_set_s_token() {
        let mut decoder = AsciiDecoder::new();
        let mut buf = BytesMut::from("ms key 5 S3\r\nabc\r\n");
        let limits = AsciiLimits::default();

        let outcome = decoder.decode(&mut buf, limits);
        let (req, meta) = match outcome {
            Some(DecodeOutcome::Request(req, meta)) => (req, meta),
            _ => panic!("unexpected decode outcome"),
        };
        assert_eq!(meta.protocol, Protocol::Meta);
        assert_eq!(req.value.unwrap(), Bytes::from_static(b"abc"));
    }

    #[test]
    fn decode_flush_all_delay_noreply() {
        let mut decoder = AsciiDecoder::new();
        let mut buf = BytesMut::from("flush_all 10 noreply\r\n");
        let outcome = decoder.decode(&mut buf, AsciiLimits::default());
        let (req, meta) = match outcome {
            Some(DecodeOutcome::Request(req, meta)) => (req, meta),
            _ => panic!("unexpected decode outcome"),
        };
        assert_eq!(req.op, Op::Flush);
        assert_eq!(req.exptime, Some(10));
        assert_eq!(meta.reply, ReplyMode::SuppressSuccess);
    }
}