wedb_embed 0.1.0

pub mod conf;
pub mod meta;

pub use conf::{
    DelEx, GetEx, Lcs, Set, StringLCSArgs, StringLCSIdxResult, StringLCSMatchedRange,
    StringLCSRange, StringLCSResult, StringLCSType, StringMSetArgs, StringPair, StringSetArgs,
    StringSetType,
};
pub use meta::{
    STRING_HDR_SIZE, StringMeta, decode_string_value, encode_string_value, is_string_expired,
};

use rapidhash::v3::rapidhash_v3;
use std::mem::size_of;
use std::str;

use crate::db::WeDb;
use crate::error::{Error, Result};
use crate::key_composer::KeyComposer;

/// Redis 字符串最大支持 512MB
pub const MAX_STRING_SIZE: usize = 512 * 1024 * 1024;

/// 索引范围标准化（对标 Redis 规范，支持负数索引回环）
#[inline]
pub fn normalize_range(start: i64, end: i64, len: i64) -> (i64, i64) {
    if len <= 0 {
        return (0, -1);
    }
    let mut s = if start < 0 { len + start } else { start };
    let mut e = if end < 0 { len + end } else { end };
    if s < 0 {
        s = 0;
    }
    if e >= len {
        e = len - 1;
    }
    (s, e)
}

/// 解析 Redis 整数（严格校验空白符与数值合法性）
#[inline]
pub fn parse_redis_integer(v: &[u8]) -> Result<i64> {
    if v.is_empty()
        || v[0].is_ascii_whitespace()
        || v.last().is_some_and(|b| b.is_ascii_whitespace())
    {
        return Err(Error::invalid_data(
            "ERR value is not an integer or out of range",
        ));
    }
    str::from_utf8(v)
        .map_err(|_| Error::invalid_data("ERR value is not an integer or out of range"))?
        .parse::<i64>()
        .map_err(|_| Error::invalid_data("ERR value is not an integer or out of range"))
}

/// 解析 Redis 浮点数（严格校验空白符与浮点合法性）
#[inline]
pub fn parse_redis_float(v: &[u8]) -> Result<f64> {
    if v.is_empty()
        || v[0].is_ascii_whitespace()
        || v.last().is_some_and(|b| b.is_ascii_whitespace())
    {
        return Err(Error::invalid_data("ERR value is not a valid float"));
    }
    str::from_utf8(v)
        .map_err(|_| Error::invalid_data("ERR value is not a valid float"))?
        .parse::<f64>()
        .map_err(|_| Error::invalid_data("ERR value is not a valid float"))
}

/// 计算字符串 64 位十六进制摘要（对标 Kvrocks util::StringDigest）
#[inline]
pub fn string_digest(val: &[u8]) -> String {
    let hash = rapidhash_v3(val);
    format!("{hash:016x}")
}

/// 紧凑浮点数字符串序列化（对标 Kvrocks util::Float2String）
#[inline]
pub fn format_float(val: f64) -> String {
    if val == 0.0 {
        "0".to_string()
    } else {
        format!("{val}")
    }
}

/// LCS (最长公共子序列) 动态规划与匹配区间提取（1:1 对标 Kvrocks String::LCS 算法与回溯状态机）
pub fn compute_lcs(s1: &[u8], s2: &[u8], args: StringLCSArgs) -> Result<StringLCSResult> {
    let alen = s1.len();
    let blen = s2.len();

    if alen == 0 || blen == 0 {
        return match args.lcs_type {
            StringLCSType::Len => Ok(StringLCSResult::Len(0)),
            StringLCSType::Idx => Ok(StringLCSResult::Idx(StringLCSIdxResult {
                matches: Vec::new(),
                len: 0,
            })),
            StringLCSType::None => Ok(StringLCSResult::Str(String::new())),
        };
    }

    if alen >= (u32::MAX - 1) as usize || blen >= (u32::MAX - 1) as usize {
        return Err(Error::invalid_data("String too long for LCS"));
    }

    // 快速路径：两个字符串完全相等
    if s1 == s2 {
        let lcs_len = alen as u32;
        return match args.lcs_type {
            StringLCSType::Len => Ok(StringLCSResult::Len(lcs_len)),
            StringLCSType::Idx => {
                let match_len = lcs_len;
                let matches = if args.min_match_len <= 0 || match_len >= args.min_match_len as u32 {
                    vec![StringLCSMatchedRange::new(
                        0,
                        (alen - 1) as u32,
                        0,
                        (blen - 1) as u32,
                        match_len,
                    )]
                } else {
                    Vec::new()
                };
                Ok(StringLCSResult::Idx(StringLCSIdxResult {
                    matches,
                    len: lcs_len,
                }))
            }
            StringLCSType::None => {
                let s = String::from_utf8(s1.to_vec())
                    .unwrap_or_else(|e| String::from_utf8_lossy(e.as_bytes()).into_owned());
                Ok(StringLCSResult::Str(s))
            }
        };
    }

    // 仅计算长度时采用 2 行滚动 DP 数组，空间复杂度从 O(M*N) 极致优化至 O(min(M, N))
    if args.lcs_type == StringLCSType::Len {
        let (short, long) = if alen <= blen { (s1, s2) } else { (s2, s1) };
        let slen = short.len();
        let mut prev = vec![0u32; slen + 1];
        let mut curr = vec![0u32; slen + 1];

        for &c2 in long {
            for (j, &c1) in short.iter().enumerate() {
                if c1 == c2 {
                    curr[j + 1] = prev[j] + 1;
                } else {
                    curr[j + 1] = curr[j].max(prev[j + 1]);
                }
            }
            std::mem::swap(&mut prev, &mut curr);
            curr.fill(0);
        }
        return Ok(StringLCSResult::Len(prev[slen]));
    }

    // Idx / None 模式构建完整 DP 表并检查内存上限
    let dp_size = (alen + 1) * (blen + 1);
    let byte_size = dp_size.checked_mul(size_of::<u32>());
    if byte_size.is_none() || byte_size.unwrap_or(usize::MAX) > MAX_STRING_SIZE {
        return Err(Error::invalid_data(
            "Insufficient memory, transient memory for LCS exceeds proto-max-bulk-len",
        ));
    }

    let mut dp = vec![0u32; dp_size];
    let stride = blen + 1;
    let idx_fn = |i: usize, j: usize| -> usize { i * stride + j };

    for i in 1..=alen {
        let s1_c = s1[i - 1];
        let row_curr = i * stride;
        let row_prev = (i - 1) * stride;
        for j in 1..=blen {
            if s1_c == s2[j - 1] {
                dp[row_curr + j] = dp[row_prev + j - 1] + 1;
            } else {
                dp[row_curr + j] = dp[row_prev + j].max(dp[row_curr + j - 1]);
            }
        }
    }

    let lcs_len = dp[idx_fn(alen, blen)];

    let mut lcs_bytes = if args.lcs_type == StringLCSType::None {
        vec![0u8; lcs_len as usize]
    } else {
        Vec::new()
    };

    let mut matches = Vec::new();
    let mut idx = lcs_len as usize;
    let mut i = alen;
    let mut j = blen;
    let mut a_range_start = alen;
    let mut a_range_end = 0;
    let mut b_range_start = 0;
    let mut b_range_end = 0;

    // 回溯构造结果与区间（1:1 对标 Kvrocks String::LCS 回溯状态机）
    while i > 0 && j > 0 {
        let mut emit_range = false;
        if s1[i - 1] == s2[j - 1] {
            if args.lcs_type == StringLCSType::None && idx > 0 {
                lcs_bytes[idx - 1] = s1[i - 1];
            }

            if a_range_start == alen {
                a_range_start = i - 1;
                a_range_end = i - 1;
                b_range_start = j - 1;
                b_range_end = j - 1;
            } else if a_range_start == i && b_range_start == j {
                a_range_start -= 1;
                b_range_start -= 1;
            } else {
                emit_range = true;
            }

            if a_range_start == 0 || b_range_start == 0 {
                emit_range = true;
            }
            idx = idx.saturating_sub(1);
            i -= 1;
            j -= 1;
        } else {
            let lcs1 = dp[idx_fn(i - 1, j)];
            let lcs2 = dp[idx_fn(i, j - 1)];
            if lcs1 > lcs2 {
                i -= 1;
            } else {
                j -= 1;
            }
            if a_range_start != alen {
                emit_range = true;
            }
        }

        if emit_range {
            if args.lcs_type == StringLCSType::Idx {
                let match_len = (a_range_end - a_range_start + 1) as u32;
                if args.min_match_len <= 0 || match_len >= args.min_match_len as u32 {
                    matches.push(StringLCSMatchedRange::new(
                        a_range_start as u32,
                        a_range_end as u32,
                        b_range_start as u32,
                        b_range_end as u32,
                        match_len,
                    ));
                }
            }
            a_range_start = alen;
        }
    }

    match args.lcs_type {
        StringLCSType::Len => Ok(StringLCSResult::Len(lcs_len)),
        StringLCSType::Idx => Ok(StringLCSResult::Idx(StringLCSIdxResult {
            matches,
            len: lcs_len,
        })),
        StringLCSType::None => {
            let s = String::from_utf8(lcs_bytes)
                .unwrap_or_else(|e| String::from_utf8_lossy(e.as_bytes()).into_owned());
            Ok(StringLCSResult::Str(s))
        }
    }
}

impl WeDb {
    /// 执行完整参数配置的 String SET 命令（1:1 对标 Kvrocks String::Set）
    pub fn set_args<K: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        key: K,
        val: V,
        args: &StringSetArgs<'_>,
    ) -> Result<Option<Vec<u8>>> {
        let key_bytes = key.as_ref();
        let val_bytes = val.as_ref();

        if val_bytes.len() > MAX_STRING_SIZE {
            return Err(Error::invalid_data(
                "ERR string exceeds maximum allowed size (512MB)",
            ));
        }

        let kc = KeyComposer::new("default");
        let raw_k = kc.raw_key_bytes(key_bytes);

        let need_old_value = args.set_type != StringSetType::None || args.get || args.keep_ttl;
        let (old_val, old_expire) = if need_old_value {
            self.get_with_expire(key_bytes)?
        } else {
            (None, 0)
        };

        // 条件检查与 GET 返回处理（1:1 对标 Kvrocks String::Set）
        let condition_met = match args.set_type {
            StringSetType::None => true,
            StringSetType::Nx => old_val.is_none(),
            StringSetType::Xx => old_val.is_some(),
            StringSetType::IfEq => {
                if let Some(expected) = args.cmp_value {
                    old_val.as_deref() == Some(expected)
                } else {
                    false
                }
            }
            StringSetType::IfNe => {
                if let Some(expected) = args.cmp_value {
                    old_val.as_deref() != Some(expected)
                } else {
                    true
                }
            }
            StringSetType::IfDeq => {
                if let Some(expected) = args.cmp_value
                    && let Some(ref cur) = old_val
                {
                    string_digest(cur).as_bytes().eq_ignore_ascii_case(expected)
                } else {
                    false
                }
            }
            StringSetType::IfDne => {
                if let Some(expected) = args.cmp_value {
                    if let Some(ref cur) = old_val {
                        !string_digest(cur).as_bytes().eq_ignore_ascii_case(expected)
                    } else {
                        true
                    }
                } else {
                    true
                }
            }
        };

        if !condition_met {
            return Ok(if args.get { old_val } else { None });
        }

        // 计算最终过期时间
        let expire = if args.keep_ttl {
            old_expire
        } else {
            args.expire
        };

        let enc_val = encode_string_value(val_bytes, expire);
        self.data_ks.insert(&*raw_k, enc_val)?;

        if args.get {
            Ok(old_val)
        } else {
            Ok(Some(Vec::new()))
        }
    }

    /// SETEX key seconds value
    pub fn setex<K: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        key: K,
        val: V,
        expire_ms: u64,
    ) -> Result<()> {
        let args = StringSetArgs {
            expire: expire_ms,
            set_type: StringSetType::None,
            get: false,
            keep_ttl: false,
            cmp_value: None,
        };
        self.set_args(key, val, &args)?;
        Ok(())
    }

    /// SETNX key value
    pub fn setnx<K: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        key: K,
        val: V,
        expire_ms: u64,
    ) -> Result<bool> {
        let args = StringSetArgs {
            expire: expire_ms,
            set_type: StringSetType::Nx,
            get: false,
            keep_ttl: false,
            cmp_value: None,
        };
        let ret = self.set_args(key, val, &args)?;
        Ok(ret.is_some())
    }

    /// SETXX key value
    pub fn setxx<K: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        key: K,
        val: V,
        expire_ms: u64,
    ) -> Result<bool> {
        let args = StringSetArgs {
            expire: expire_ms,
            set_type: StringSetType::Xx,
            get: false,
            keep_ttl: false,
            cmp_value: None,
        };
        let ret = self.set_args(key, val, &args)?;
        Ok(ret.is_some())
    }

    /// GETEX key [EX seconds | PX milliseconds | EXAT timestamp | PXAT timestamp | PERSIST]
    pub fn getex<K: AsRef<[u8]>>(&self, key: K, opt: Option<GetEx>) -> Result<Option<Vec<u8>>> {
        let key_bytes = key.as_ref();
        let kc = KeyComposer::new("default");
        let raw_k = kc.raw_key_bytes(key_bytes);

        let (val, _) = self.get_with_expire(key_bytes)?;
        let val = match val {
            Some(v) => v,
            None => return Ok(None),
        };

        if let Some(opt_val) = opt {
            let now_ms = coarsetime::Clock::now_since_epoch().as_millis();
            let new_expire = match opt_val {
                GetEx::Persist => 0,
                GetEx::Ex(sec) => now_ms.saturating_add(sec.saturating_mul(1000)),
                GetEx::Px(ms) => now_ms.saturating_add(ms),
                GetEx::ExAt(sec) => sec.saturating_mul(1000),
                GetEx::PxAt(ms) => ms,
            };
            let enc_val = encode_string_value(&val, new_expire);
            self.data_ks.insert(&*raw_k, enc_val)?;
        }

        Ok(Some(val))
    }

    /// DELEX key [IFEQ val | IFNE val | IFDEQ digest | IFDNE digest]
    pub fn delex<K: AsRef<[u8]>>(&self, key: K, opt: DelEx<'_>) -> Result<bool> {
        let key_bytes = key.as_ref();
        let kc = KeyComposer::new("default");
        let raw_k = kc.raw_key_bytes(key_bytes);

        let (val, _) = self.get_with_expire(key_bytes)?;
        let val = match val {
            Some(v) => v,
            None => return Ok(false),
        };

        let matched = match opt {
            DelEx::None => true,
            DelEx::IfEq(expected) => val.as_slice() == expected,
            DelEx::IfNe(expected) => val.as_slice() != expected,
            DelEx::IfDeq(expected) => string_digest(&val)
                .as_bytes()
                .eq_ignore_ascii_case(expected),
            DelEx::IfDne(expected) => !string_digest(&val)
                .as_bytes()
                .eq_ignore_ascii_case(expected),
        };

        if matched {
            self.data_ks.remove(&*raw_k)?;
            Ok(true)
        } else {
            Ok(false)
        }
    }

    /// GETSET key value
    pub fn getset<K: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        key: K,
        val: V,
    ) -> Result<Option<Vec<u8>>> {
        let args = StringSetArgs {
            expire: 0,
            set_type: StringSetType::None,
            get: true,
            keep_ttl: false,
            cmp_value: None,
        };
        self.set_args(key, val, &args)
    }

    /// GETDEL key
    pub fn getdel<K: AsRef<[u8]>>(&self, key: K) -> Result<Option<Vec<u8>>> {
        let key_bytes = key.as_ref();
        let (val, _) = self.get_with_expire(key_bytes)?;
        if val.is_some() {
            let kc = KeyComposer::new("default");
            let raw_k = kc.raw_key_bytes(key_bytes);
            self.data_ks.remove(&*raw_k)?;
        }
        Ok(val)
    }

    /// INCR key
    pub fn incr<K: AsRef<[u8]>>(&self, key: K) -> Result<i64> {
        self.incrby(key, 1)
    }

    /// DECR key
    pub fn decr<K: AsRef<[u8]>>(&self, key: K) -> Result<i64> {
        self.decrby(key, 1)
    }

    /// DECRBY key decrement（严格处理减法与边界，防止 i64::MIN 溢出）
    pub fn decrby<K: AsRef<[u8]>>(&self, key: K, decrement: i64) -> Result<i64> {
        let key_bytes = key.as_ref();
        let kc = KeyComposer::new("default");
        let raw_k = kc.raw_key_bytes(key_bytes);

        let (current_val, cur_expire) = self.get_with_expire(key_bytes)?;
        let cur_num: i64 = match current_val {
            Some(ref v) => parse_redis_integer(v)?,
            None => 0,
        };
        let new_num = cur_num
            .checked_sub(decrement)
            .ok_or_else(|| Error::invalid_data("ERR increment or decrement would overflow"))?;
        let mut itoa_buf = itoa::Buffer::new();
        let new_val_str = itoa_buf.format(new_num);
        let enc_val = encode_string_value(new_val_str.as_bytes(), cur_expire);
        self.data_ks.insert(&*raw_k, enc_val)?;
        Ok(new_num)
    }

    /// INCRBY key increment（严格检查空白符与数值溢出，对标 Kvrocks IncrBy）
    pub fn incrby<K: AsRef<[u8]>>(&self, key: K, increment: i64) -> Result<i64> {
        let key_bytes = key.as_ref();
        let kc = KeyComposer::new("default");
        let raw_k = kc.raw_key_bytes(key_bytes);

        let (current_val, cur_expire) = self.get_with_expire(key_bytes)?;
        let cur_num: i64 = match current_val {
            Some(ref v) => parse_redis_integer(v)?,
            None => 0,
        };
        let new_num = cur_num
            .checked_add(increment)
            .ok_or_else(|| Error::invalid_data("ERR increment or decrement would overflow"))?;
        let mut itoa_buf = itoa::Buffer::new();
        let new_val_str = itoa_buf.format(new_num);
        let enc_val = encode_string_value(new_val_str.as_bytes(), cur_expire);
        self.data_ks.insert(&*raw_k, enc_val)?;
        Ok(new_num)
    }

    /// INCRBYFLOAT key increment（严格检查浮点合法性与溢出，对标 Kvrocks IncrByFloat）
    pub fn incrbyfloat<K: AsRef<[u8]>>(&self, key: K, increment: f64) -> Result<f64> {
        if increment.is_nan() || increment.is_infinite() {
            return Err(Error::invalid_data(
                "ERR increment would produce NaN or Infinity",
            ));
        }

        let key_bytes = key.as_ref();
        let kc = KeyComposer::new("default");
        let raw_k = kc.raw_key_bytes(key_bytes);

        let (current_val, cur_expire) = self.get_with_expire(key_bytes)?;
        let cur_num: f64 = match current_val {
            Some(ref v) => parse_redis_float(v)?,
            None => 0.0,
        };
        let mut new_num = cur_num + increment;
        if new_num.is_nan() || new_num.is_infinite() {
            return Err(Error::invalid_data(
                "ERR increment would produce NaN or Infinity",
            ));
        }
        if new_num == 0.0 {
            new_num = 0.0;
        }
        let new_val_str = format_float(new_num);
        let enc_val = encode_string_value(new_val_str.as_bytes(), cur_expire);
        self.data_ks.insert(&*raw_k, enc_val)?;
        Ok(new_num)
    }

    /// STRLEN key
    pub fn strlen<K: AsRef<[u8]>>(&self, key: K) -> Result<usize> {
        match self.get(key)? {
            Some(v) => Ok(v.len()),
            None => Ok(0),
        }
    }

    /// APPEND key value
    pub fn append<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, val: V) -> Result<usize> {
        let key_bytes = key.as_ref();
        let val_bytes = val.as_ref();
        let kc = KeyComposer::new("default");
        let raw_k = kc.raw_key_bytes(key_bytes);

        let (current_val, cur_expire) = self.get_with_expire(key_bytes)?;
        let mut current_bytes = current_val.unwrap_or_default();
        let new_len = current_bytes
            .len()
            .checked_add(val_bytes.len())
            .ok_or_else(|| {
                Error::invalid_data("ERR string exceeds maximum allowed size (512MB)")
            })?;
        if new_len > MAX_STRING_SIZE {
            return Err(Error::invalid_data(
                "ERR string exceeds maximum allowed size (512MB)",
            ));
        }

        current_bytes.extend_from_slice(val_bytes);
        let enc_val = encode_string_value(&current_bytes, cur_expire);
        self.data_ks.insert(&*raw_k, enc_val)?;
        Ok(new_len)
    }

    /// GETRANGE key start end
    pub fn getrange<K: AsRef<[u8]>>(&self, key: K, start: i64, end: i64) -> Result<Vec<u8>> {
        let current_bytes = self.get(key)?.unwrap_or_default();
        let len = current_bytes.len() as i64;
        let (s, e) = normalize_range(start, end, len);
        if s > e || current_bytes.is_empty() {
            return Ok(Vec::new());
        }
        Ok(current_bytes[s as usize..=e as usize].to_vec())
    }

    /// SETRANGE key offset value（对标 Kvrocks SetRange 边界与零填充）
    pub fn setrange<K: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        key: K,
        offset: usize,
        val: V,
    ) -> Result<usize> {
        let key_bytes = key.as_ref();
        let val_bytes = val.as_ref();
        let kc = KeyComposer::new("default");
        let raw_k = kc.raw_key_bytes(key_bytes);

        if offset > MAX_STRING_SIZE {
            return Err(Error::invalid_data("ERR offset is out of range"));
        }

        let required_len = offset.checked_add(val_bytes.len()).ok_or_else(|| {
            Error::invalid_data("ERR string exceeds maximum allowed size (512MB)")
        })?;
        if required_len > MAX_STRING_SIZE {
            return Err(Error::invalid_data(
                "ERR string exceeds maximum allowed size (512MB)",
            ));
        }

        let (current_val, cur_expire) = self.get_with_expire(key_bytes)?;
        if current_val.is_none() && val_bytes.is_empty() {
            return Ok(0);
        }

        let mut current_bytes = current_val.unwrap_or_default();

        if current_bytes.len() < required_len {
            current_bytes.resize(required_len, 0);
        }

        current_bytes[offset..required_len].copy_from_slice(val_bytes);
        let new_len = current_bytes.len();
        let enc_val = encode_string_value(&current_bytes, cur_expire);
        self.data_ks.insert(&*raw_k, enc_val)?;
        Ok(new_len)
    }

    /// MGET key [key ...]
    pub fn mget<K: AsRef<[u8]>>(&self, keys: &[K]) -> Result<Vec<Option<Vec<u8>>>> {
        let mut results = Vec::with_capacity(keys.len());
        for k in keys {
            results.push(self.get(k)?);
        }
        Ok(results)
    }

    /// MSET key value [key value ...]
    pub fn mset<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, kvs: &[(K, V)]) -> Result<()> {
        self.mset_args(kvs, StringMSetArgs::default())?;
        Ok(())
    }

    /// MSET 参数控制版本（对标 Kvrocks String::MSet）
    pub fn mset_args<K: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        kvs: &[(K, V)],
        args: StringMSetArgs,
    ) -> Result<bool> {
        if kvs.is_empty() {
            return Ok(true);
        }

        let kc = KeyComposer::new("default");
        let mut expires = vec![0u64; kvs.len()];

        if args.set_type != StringSetType::None || args.keep_ttl {
            for (i, (k, _)) in kvs.iter().enumerate() {
                let (cur_val, cur_exp) = self.get_with_expire(k)?;
                if cur_val.is_some() {
                    if args.set_type == StringSetType::Nx {
                        return Ok(false);
                    }
                    expires[i] = cur_exp;
                } else if args.set_type == StringSetType::Xx {
                    return Ok(false);
                }
            }
        }

        let mut batch = self.db.batch();
        for (i, (k, v)) in kvs.iter().enumerate() {
            let v_bytes = v.as_ref();
            if v_bytes.len() > MAX_STRING_SIZE {
                return Err(Error::invalid_data(
                    "ERR string exceeds maximum allowed size (512MB)",
                ));
            }
            let raw_k = kc.raw_key_bytes(k.as_ref());
            let expire = if args.keep_ttl {
                expires[i]
            } else {
                args.expire
            };
            let enc_val = encode_string_value(v_bytes, expire);
            batch.insert(&self.data_ks, &*raw_k, enc_val);
        }
        batch.commit()?;
        Ok(true)
    }

    /// MSETEX key value [key value ...]
    pub fn msetex<K: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        kvs: &[(K, V)],
        expire_ms: u64,
    ) -> Result<bool> {
        self.mset_args(
            kvs,
            StringMSetArgs {
                expire: expire_ms,
                set_type: StringSetType::None,
                keep_ttl: false,
            },
        )
    }

    /// MSETNX key value [key value ...]
    pub fn msetnx<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, kvs: &[(K, V)]) -> Result<bool> {
        self.mset_args(
            kvs,
            StringMSetArgs {
                expire: 0,
                set_type: StringSetType::Nx,
                keep_ttl: false,
            },
        )
    }

    /// CAS (Compare And Set)：当且仅当 key 当前值与 old_value 匹配时，将其更新为 new_value
    /// 返回值：1 = 成功更新；-1 = key 不存在；0 = 值不匹配（对标 Kvrocks String::CAS）
    pub fn cas<K: AsRef<[u8]>, V1: AsRef<[u8]>, V2: AsRef<[u8]>>(
        &self,
        key: K,
        old_val: V1,
        new_val: V2,
        expire_ms: u64,
    ) -> Result<i32> {
        let key_bytes = key.as_ref();
        let old_bytes = old_val.as_ref();
        let new_bytes = new_val.as_ref();
        if new_bytes.len() > MAX_STRING_SIZE {
            return Err(Error::invalid_data(
                "ERR string exceeds maximum allowed size (512MB)",
            ));
        }

        let kc = KeyComposer::new("default");
        let raw_k = kc.raw_key_bytes(key_bytes);

        let (cur_val, _) = self.get_with_expire(key_bytes)?;
        let cur_val = match cur_val {
            Some(v) => v,
            None => return Ok(-1),
        };

        if cur_val.as_slice() == old_bytes {
            let enc_val = encode_string_value(new_bytes, expire_ms);
            self.data_ks.insert(&*raw_k, enc_val)?;
            Ok(1)
        } else {
            Ok(0)
        }
    }

    /// CAD (Compare And Delete)：当且仅当 key 当前值与 val 匹配时删除该 key
    /// 返回值：1 = 成功删除；-1 = key 不存在；0 = 值不匹配（对标 Kvrocks String::CAD）
    pub fn cad<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, val: V) -> Result<i32> {
        let key_bytes = key.as_ref();
        let kc = KeyComposer::new("default");
        let raw_k = kc.raw_key_bytes(key_bytes);

        let (cur_val, _) = self.get_with_expire(key_bytes)?;
        let cur_val = match cur_val {
            Some(v) => v,
            None => return Ok(-1),
        };

        if cur_val.as_slice() == val.as_ref() {
            self.data_ks.remove(&*raw_k)?;
            Ok(1)
        } else {
            Ok(0)
        }
    }

    /// 计算指定 key 的哈希摘要（对标 Kvrocks String::Digest）
    pub fn digest<K: AsRef<[u8]>>(&self, key: K) -> Result<Option<String>> {
        match self.get(key)? {
            Some(v) => Ok(Some(string_digest(&v))),
            None => Ok(None),
        }
    }

    /// 计算两个 key 字符串的最长公共子序列（对标 Kvrocks String::LCS）
    pub fn lcs<K1: AsRef<[u8]>, K2: AsRef<[u8]>>(
        &self,
        key1: K1,
        key2: K2,
        args: StringLCSArgs,
    ) -> Result<StringLCSResult> {
        let s1 = self.get(key1)?.unwrap_or_default();
        let s2 = self.get(key2)?.unwrap_or_default();
        compute_lcs(&s1, &s2, args)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::tempdir;

    #[test]
    fn test_normalize_range() {
        assert_eq!(normalize_range(0, 4, 10), (0, 4));
        assert_eq!(normalize_range(-3, -1, 10), (7, 9));
        assert_eq!(normalize_range(5, 100, 10), (5, 9));
        assert_eq!(normalize_range(0, -1, 0), (0, -1));
        assert_eq!(normalize_range(-100, 2, 10), (0, 2));
        // 负数区间超出起始边界：-20 到 -15，应当产生 s > e 的空区间
        let (s, e) = normalize_range(-20, -15, 10);
        assert!(s > e);
        // start > end 产生空区间
        let (s, e) = normalize_range(5, 2, 10);
        assert!(s > e);
    }

    #[test]
    fn test_compute_lcs_algorithms() {
        let (s1, s2) = (b"abcdef", b"abzdef");
        let res_len = compute_lcs(
            s1,
            s2,
            StringLCSArgs {
                lcs_type: StringLCSType::Len,
                min_match_len: 0,
            },
        )
        .expect("lcs len failed");
        assert_eq!(res_len, StringLCSResult::Len(5));

        let res_str = compute_lcs(
            s1,
            s2,
            StringLCSArgs {
                lcs_type: StringLCSType::None,
                min_match_len: 0,
            },
        )
        .expect("lcs str failed");
        assert_eq!(res_str, StringLCSResult::Str("abdef".to_string()));

        let res_idx = compute_lcs(
            s1,
            s2,
            StringLCSArgs {
                lcs_type: StringLCSType::Idx,
                min_match_len: 2,
            },
        )
        .expect("lcs idx failed");
        if let StringLCSResult::Idx(idx_res) = res_idx {
            assert_eq!(idx_res.len, 5);
            assert!(!idx_res.matches.is_empty());
        } else {
            panic!("expected Idx result");
        }

        // 测试相同字符串的快速路径
        let res_same_len = compute_lcs(
            b"identical",
            b"identical",
            StringLCSArgs {
                lcs_type: StringLCSType::Len,
                min_match_len: 0,
            },
        )
        .expect("lcs same len failed");
        assert_eq!(res_same_len, StringLCSResult::Len(9));

        let res_same_str = compute_lcs(
            b"identical",
            b"identical",
            StringLCSArgs {
                lcs_type: StringLCSType::None,
                min_match_len: 0,
            },
        )
        .expect("lcs same str failed");
        assert_eq!(res_same_str, StringLCSResult::Str("identical".to_string()));

        let res_same_idx = compute_lcs(
            b"identical",
            b"identical",
            StringLCSArgs {
                lcs_type: StringLCSType::Idx,
                min_match_len: 0,
            },
        )
        .expect("lcs same idx failed");
        if let StringLCSResult::Idx(idx_res) = res_same_idx {
            assert_eq!(idx_res.len, 9);
            assert_eq!(idx_res.matches.len(), 1);
            assert_eq!(idx_res.matches[0].match_len, 9);
        } else {
            panic!("expected Idx result");
        }
    }

    #[test]
    fn test_parse_redis_number_helpers() {
        assert_eq!(parse_redis_integer(b"12345").unwrap(), 12345);
        assert_eq!(parse_redis_integer(b"-9876").unwrap(), -9876);
        assert_eq!(parse_redis_integer(b"0").unwrap(), 0);
        assert!(parse_redis_integer(b"").is_err());
        assert!(parse_redis_integer(b" 123").is_err());
        assert!(parse_redis_integer(b"123 ").is_err());
        assert!(parse_redis_integer(b"abc").is_err());

        assert!((parse_redis_float(b"3.14159").unwrap() - 3.14159).abs() < 1e-6);
        assert!((parse_redis_float(b"-0.5").unwrap() - (-0.5)).abs() < 1e-6);
        assert!(parse_redis_float(b"").is_err());
        assert!(parse_redis_float(b" 3.14").is_err());
        assert!(parse_redis_float(b"3.14 ").is_err());
        assert!(parse_redis_float(b"not_float").is_err());
    }

    #[test]
    fn test_format_float() {
        assert_eq!(format_float(0.0), "0");
        assert_eq!(format_float(-0.0), "0");
        assert_eq!(format_float(3.0), "3");
        assert_eq!(format_float(10.5), "10.5");
        assert_eq!(format_float(-1.25), "-1.25");
    }

    #[test]
    fn test_string_full_operations() -> Result<()> {
        let dir = tempdir()?;
        let db = WeDb::open(dir.path())?;

        // 基本 SET/GET
        db.set("k1", "v1", &[])?;
        assert_eq!(db.get("k1")?, Some(b"v1".to_vec()));

        // SET NX/XX
        let set_nx_res = db.setnx("k1", "v2", 0)?;
        assert!(!set_nx_res);
        assert_eq!(db.get("k1")?, Some(b"v1".to_vec()));

        let set_xx_res = db.setxx("k1", "v2", 0)?;
        assert!(set_xx_res);
        assert_eq!(db.get("k1")?, Some(b"v2".to_vec()));

        // GETSET
        let prev = db.getset("k1", "v3")?;
        assert_eq!(prev, Some(b"v2".to_vec()));
        assert_eq!(db.get("k1")?, Some(b"v3".to_vec()));

        // APPEND / STRLEN
        let len = db.append("k1", "_appended")?;
        assert_eq!(len, 11);
        assert_eq!(db.strlen("k1")?, 11);

        // GETRANGE / SETRANGE
        assert_eq!(db.getrange("k1", 0, 1)?, b"v3".to_vec());
        db.setrange("k1", 0, "V3")?;
        assert_eq!(db.getrange("k1", 0, 1)?, b"V3".to_vec());

        // INCR / INCRBY / DECR / DECRBY
        db.set("num", "10", &[])?;
        assert_eq!(db.incr("num")?, 11);
        assert_eq!(db.incrby("num", 5)?, 16);
        assert_eq!(db.decr("num")?, 15);
        assert_eq!(db.decrby("num", 5)?, 10);

        // DECRBY with i64::MIN
        db.set("num2", "-1", &[])?;
        assert_eq!(db.decrby("num2", i64::MIN)?, i64::MAX);

        // INCRBYFLOAT
        assert!((db.incrbyfloat("num", 2.5)? - 12.5).abs() < 1e-6);

        // MGET / MSET / MSETNX
        db.mset(&[("m1", "val1"), ("m2", "val2")])?;
        let mvals = db.mget(&["m1", "m2", "non_exist"])?;
        assert_eq!(
            mvals,
            vec![Some(b"val1".to_vec()), Some(b"val2".to_vec()), None]
        );

        let msetnx_fail = db.msetnx(&[("m1", "new1"), ("m3", "val3")])?;
        assert!(!msetnx_fail);
        assert_eq!(db.get("m3")?, None);

        // CAS / CAD
        db.set("cas_k", "old", &[])?;
        assert_eq!(db.cas("cas_k", "wrong", "new", 0)?, 0);
        assert_eq!(db.cas("cas_k", "old", "new", 0)?, 1);
        assert_eq!(db.get("cas_k")?, Some(b"new".to_vec()));

        assert_eq!(db.cad("cas_k", "wrong")?, 0);
        assert_eq!(db.cad("cas_k", "new")?, 1);
        assert_eq!(db.get("cas_k")?, None);

        // DELEX
        db.set("del_k", "my_secret", &[])?;
        let d = db.digest("del_k")?.expect("digest missing");
        assert!(db.delex("del_k", DelEx::IfDeq(d.as_bytes()))?);
        assert_eq!(db.get("del_k")?, None);

        // LCS
        db.set("lcs1", "ohmytext", &[])?;
        db.set("lcs2", "mynewtext", &[])?;
        let lcs_res = db.lcs(
            "lcs1",
            "lcs2",
            StringLCSArgs {
                lcs_type: StringLCSType::None,
                min_match_len: 0,
            },
        )?;
        assert_eq!(lcs_res, StringLCSResult::Str("mytext".to_string()));

        // Binary Key
        let bin_key = b"\x00\xff\xfe_binary_key";
        db.set(bin_key, "bin_val", &[])?;
        assert_eq!(db.get(bin_key)?, Some(b"bin_val".to_vec()));
        assert_eq!(db.del(&[bin_key])?, 1);
        assert_eq!(db.get(bin_key)?, None);

        Ok(())
    }
}