wedb_embed 0.1.1

pub mod conf;
pub mod meta;

pub use conf::{
    DelEx, ERR_DIGEST_INVALID_LEN, ERR_INCREMENT_NAN_OR_INFINITY, ERR_INCREMENT_OVERFLOW,
    ERR_LCS_INSUFFICIENT_MEMORY, ERR_LCS_TOO_LONG, ERR_OFFSET_OUT_OF_RANGE,
    ERR_STRING_EXCEEDS_MAX_SIZE, ERR_VALUE_NOT_FLOAT, ERR_VALUE_NOT_INTEGER, ERR_WRONG_TYPE, 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_header, encode_string_value,
    encode_string_value_into, is_string_expired,
};

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

use crate::db::WeDb;
use crate::error::{Error, Result};
use crate::key_composer::{ALL_COMPOSITE_META_TAGS, KeyComposer};
use crate::meta::{
    KeyMeta, bytes_to_hex_16, current_now_ms, parse_redis_float as meta_parse_redis_float,
    parse_redis_integer as meta_parse_redis_integer, u64_to_hex_16,
};

pub use wedb_resp::parse_i64_fast;

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

pub use crate::meta::normalize_range;

/// 解析 Redis 整数（单次遍历零拷贝字节解析，严格校验空白符与数值合法性，对标 Kvrocks ParseInt）
#[inline]
pub fn parse_redis_integer(v: &[u8]) -> Result<i64> {
    meta_parse_redis_integer(v, ERR_VALUE_NOT_INTEGER)
}

/// 解析 Redis 浮点数（严格校验空白符与浮点合法性，对标 Kvrocks ParseFloat）
#[inline]
pub fn parse_redis_float(v: &[u8]) -> Result<f64> {
    meta_parse_redis_float(v, ERR_VALUE_NOT_FLOAT)
}

/// 计算字符串 64 位十六进制摘要（对标 Kvrocks util::StringDigest，单次分配）
#[inline]
pub fn string_digest(val: &[u8]) -> String {
    let hash = rapidhash_v3(val);
    let bytes = u64_to_hex_16(hash);
    unsafe { str::from_utf8_unchecked(&bytes).to_string() }
}

/// 计算字符串 16 字节十六进制摘要数组（零堆分配，对标 Kvrocks util::StringDigest）
#[inline]
pub fn string_digest_bytes(val: &[u8]) -> [u8; 16] {
    let hash = rapidhash_v3(val);
    bytes_to_hex_16(hash.to_be_bytes())
}

/// 紧凑浮点数字节序列化（基于 zmij 实现零堆分配切片生成，对标 Kvrocks util::Float2String）
#[inline]
pub fn format_float_bytes(val: f64, buf: &mut zmij::Buffer) -> &[u8] {
    if val.is_infinite() {
        if val.is_sign_positive() {
            b"inf"
        } else {
            b"-inf"
        }
    } else if val.is_nan() {
        b"nan"
    } else if val == 0.0 {
        b"0"
    } else {
        let s = buf.format_finite(val);
        if let Some(stripped) = s.strip_suffix(".0") {
            stripped.as_bytes()
        } else {
            s.as_bytes()
        }
    }
}

/// 紧凑浮点数字符串序列化（基于 zmij 实现高性能零堆分配格式化，对标 Kvrocks util::Float2String）
#[inline]
pub fn format_float(val: f64) -> String {
    let mut buf = zmij::Buffer::new();
    let bytes = format_float_bytes(val, &mut buf);
    unsafe { str::from_utf8_unchecked(bytes).to_string() }
}

/// 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(ERR_LCS_TOO_LONG));
    }

    // 快速路径：两个字符串完全相等
    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_lossy(s1).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]);
                }
            }
            swap(&mut prev, &mut curr);
            curr.fill(0);
        }
        return Ok(StringLCSResult::Len(prev[slen]));
    }

    // Idx / None 模式构建完整 DP 表并检查内存上限
    let dp_size = (alen + 1).saturating_mul(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(ERR_LCS_INSUFFICIENT_MEMORY));
    }

    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_lossy(&lcs_bytes).into_owned();
            Ok(StringLCSResult::Str(s))
        }
    }
}

impl WeDb {
    /// 零拷贝读取底层 SingleKV 原始切片与过期信息（内部核心辅助方法，支持严格 WRONGTYPE 校验）
    #[inline]
    pub(crate) fn get_string_raw(
        &self,
        key_bytes: &[u8],
    ) -> Result<Option<(fjall::Slice, u64, usize)>> {
        let kc = KeyComposer::new("default");
        let raw_k = kc.raw_key_bytes(key_bytes);
        let now_ms = current_now_ms();

        // 1. 优先检索 String 数据表
        if let Some(raw) = self.data.get(&raw_k)? {
            let (expire_at, payload) = decode_string_value(&raw);
            if !is_string_expired(expire_at, now_ms) {
                let offset = raw.len() - payload.len();
                return Ok(Some((raw, expire_at, offset)));
            }
        }

        // 2. 检查复合数据结构元数据（Hash/List/Set/ZSet 等），严格实现 WRONGTYPE 拦截（1:1 对标 Kvrocks ParseMetadataWithStats）
        // 快速判断：若 meta 表为空，则数据库内无任何复合数据结构，直接返回 None
        if self.meta.is_empty()? {
            return Ok(None);
        }

        // 复用单缓冲区迭代所有复合元数据标签，杜绝堆分配
        let mut buf = Vec::with_capacity(32 + key_bytes.len());
        for &tag in ALL_COMPOSITE_META_TAGS {
            kc.compose_meta_key_into(tag, key_bytes, &mut buf);
            if let Some(m_bytes) = self.meta.get(&buf)?
                && let Some(base_meta) = KeyMeta::decode(&m_bytes)
                && !base_meta.is_expired(now_ms)
            {
                return Err(Error::wrong_type(ERR_WRONG_TYPE));
            }
        }

        Ok(None)
    }

    /// 执行完整参数配置的 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_MAX_SIZE));
        }

        // 校验 IFDEQ / IFDNE 摘要格式（16位十六进制字符）
        if let Some(expected_digest) = args.cmp_value
            && matches!(args.set_type, StringSetType::IfDeq | StringSetType::IfDne)
            && (expected_digest.len() != 16
                || !expected_digest.iter().all(|b| b.is_ascii_hexdigit()))
        {
            return Err(Error::invalid_data(ERR_DIGEST_INVALID_LEN));
        }

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

        // 极速直写快速通道 (Fast Path)：
        // 当执行常规 SET key value（无 NX/XX/IFEQ/IFNE 等条件、无 GET、无 KEEPTTL，且无 cmp_value）时，
        // 零复合元数据表检索，零批处理分配开销，直接序列化并原子直写底层 SingleKV
        if args.is_fast_path() {
            let enc_val = encode_string_value(val_bytes, args.expire);
            self.data.insert(&*raw_k, enc_val)?;
            return Ok(Some(Vec::new()));
        }

        let need_old_value = args.set_type != StringSetType::None || args.get || args.keep_ttl;
        let old_raw_res = if need_old_value {
            Some(self.get_string_raw(key_bytes))
        } else {
            None
        };

        // 处理非 String 类型冲突与预期行为（1:1 对标 Kvrocks String::Set WRONGTYPE 处理）
        let (old_raw, old_is_wrong_type) = match old_raw_res {
            Some(Ok(v)) => (v, false),
            Some(Err(e)) => {
                if args.set_type == StringSetType::IfEq
                    || args.set_type == StringSetType::IfNe
                    || args.set_type == StringSetType::IfDeq
                    || args.set_type == StringSetType::IfDne
                    || args.get
                {
                    return Err(e);
                }
                if args.set_type == StringSetType::Nx {
                    return Ok(None);
                }
                // XX 或其他选项允许直接覆盖非 String 类型键
                (None, true)
            }
            None => (None, false),
        };

        // 条件检查（1:1 对标 Kvrocks String::Set 条件状态机）
        let condition_met = match args.set_type {
            StringSetType::None => true,
            StringSetType::Nx => old_raw.is_none() && !old_is_wrong_type,
            StringSetType::Xx => old_raw.is_some() || old_is_wrong_type,
            StringSetType::IfEq => {
                if let Some(expected) = args.cmp_value
                    && let Some((ref r, _, offset)) = old_raw
                {
                    &r[offset..] == expected
                } else {
                    false
                }
            }
            StringSetType::IfNe => {
                if let Some(expected) = args.cmp_value {
                    if let Some((ref r, _, offset)) = old_raw {
                        &r[offset..] != expected
                    } else {
                        true
                    }
                } else {
                    true
                }
            }
            StringSetType::IfDeq => {
                if let Some(expected) = args.cmp_value
                    && let Some((ref r, _, offset)) = old_raw
                {
                    string_digest_bytes(&r[offset..]).eq_ignore_ascii_case(expected)
                } else {
                    false
                }
            }
            StringSetType::IfDne => {
                if let Some(expected) = args.cmp_value {
                    if let Some((ref r, _, offset)) = old_raw {
                        !string_digest_bytes(&r[offset..]).eq_ignore_ascii_case(expected)
                    } else {
                        true
                    }
                } else {
                    true
                }
            }
        };

        let old_val = if args.get {
            old_raw.as_ref().map(|(r, _, offset)| r[*offset..].to_vec())
        } else {
            None
        };

        if !condition_met {
            return Ok(old_val);
        }

        // 计算最终过期时间
        let expire = if args.keep_ttl {
            old_raw.as_ref().map(|(_, exp, _)| *exp).unwrap_or(0)
        } else {
            args.expire
        };

        let enc_val = encode_string_value(val_bytes, expire);

        if old_is_wrong_type {
            // 仅在明确检测到复合类型冲突时进行批量原子清理与替换，杜绝悬挂元数据与子键
            let mut batch = self.db.batch();
            batch.insert(&self.data, &*raw_k, enc_val);
            self.cleanup_all_composite_data(&kc, key_bytes, &mut batch)?;
            batch.commit()?;
        } else {
            // 无类型冲突时直接原子写入 SingleKV
            self.data.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 old_raw = match self.get_string_raw(key_bytes)? {
            Some(v) => v,
            None => return Ok(None),
        };

        let (raw, _, offset) = old_raw;
        let payload = &raw[offset..];
        let res_vec = payload.to_vec();

        if let Some(opt_val) = opt {
            let now_ms = current_now_ms();
            let new_expire = opt_val.compute_expire(now_ms);
            let enc_val = encode_string_value(payload, new_expire);
            self.data.insert(&*raw_k, enc_val)?;
        }

        Ok(Some(res_vec))
    }

    /// 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 old_raw = match self.get_string_raw(key_bytes)? {
            Some(v) => v,
            None => return Ok(false),
        };

        let (raw, _, offset) = old_raw;
        let val_slice = &raw[offset..];

        let matched = match opt {
            DelEx::None => true,
            DelEx::IfEq(expected) => val_slice == expected,
            DelEx::IfNe(expected) => val_slice != expected,
            DelEx::IfDeq(expected) => {
                if expected.len() != 16 || !expected.iter().all(|b| b.is_ascii_hexdigit()) {
                    return Err(Error::invalid_data(ERR_DIGEST_INVALID_LEN));
                }
                string_digest_bytes(val_slice).eq_ignore_ascii_case(expected)
            }
            DelEx::IfDne(expected) => {
                if expected.len() != 16 || !expected.iter().all(|b| b.is_ascii_hexdigit()) {
                    return Err(Error::invalid_data(ERR_DIGEST_INVALID_LEN));
                }
                !string_digest_bytes(val_slice).eq_ignore_ascii_case(expected)
            }
        };

        if matched {
            self.data.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();
        if let Some((raw, _, offset)) = self.get_string_raw(key_bytes)? {
            let kc = KeyComposer::new("default");
            let raw_k = kc.raw_key_bytes(key_bytes);
            let val = raw[offset..].to_vec();
            self.data.remove(&*raw_k)?;
            Ok(Some(val))
        } else {
            Ok(None)
        }
    }

    /// INCRBYEX: 支持显式指定过期时间或保持 TTL 的原子递增操作（对标 Kvrocks IncrBy 机制拓展）
    pub fn incrby_ex<K: AsRef<[u8]>>(
        &self,
        key: K,
        increment: i64,
        expire_ms: u64,
        keep_ttl: bool,
    ) -> Result<i64> {
        let key_bytes = key.as_ref();
        let kc = KeyComposer::new("default");
        let raw_k = kc.raw_key_bytes(key_bytes);

        let old_raw = self.get_string_raw(key_bytes)?;
        let (cur_num, old_expire) = match old_raw {
            Some((ref raw, exp, offset)) => (parse_redis_integer(&raw[offset..])?, exp),
            None => (0i64, 0u64),
        };

        // 溢出校验（1:1 对标 Kvrocks LLONG 溢出判断）
        if (increment < 0 && cur_num <= 0 && increment < (i64::MIN - cur_num))
            || (increment > 0 && cur_num >= 0 && increment > (i64::MAX - cur_num))
        {
            return Err(Error::invalid_data(ERR_INCREMENT_OVERFLOW));
        }

        let new_num = cur_num
            .checked_add(increment)
            .ok_or_else(|| Error::invalid_data(ERR_INCREMENT_OVERFLOW))?;

        let final_expire = if keep_ttl { old_expire } else { expire_ms };

        let mut itoa_buf = itoa::Buffer::new();
        let formatted_bytes = itoa_buf.format(new_num).as_bytes();

        let enc_val = encode_string_value(formatted_bytes, final_expire);
        self.data.insert(&*raw_k, enc_val)?;
        Ok(new_num)
    }

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

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

    /// DECRBY key decrement（严格处理减法与边界，防止 i64::MIN 溢出）
    pub fn decrby<K: AsRef<[u8]>>(&self, key: K, decrement: i64) -> Result<i64> {
        if decrement == i64::MIN {
            return Err(Error::invalid_data(ERR_INCREMENT_OVERFLOW));
        }
        self.incrby_ex(key, -decrement, 0, true)
    }

    /// INCRBY key increment（严格检查空白符与数值溢出，对标 Kvrocks IncrBy）
    pub fn incrby<K: AsRef<[u8]>>(&self, key: K, increment: i64) -> Result<i64> {
        self.incrby_ex(key, increment, 0, true)
    }

    /// INCRBYFLOAT_EX: 支持显式指定过期时间或保持 TTL 的浮点原子递增
    pub fn incrbyfloat_ex<K: AsRef<[u8]>>(
        &self,
        key: K,
        increment: f64,
        expire_ms: u64,
        keep_ttl: bool,
    ) -> Result<f64> {
        if increment.is_nan() || increment.is_infinite() {
            return Err(Error::invalid_data(ERR_INCREMENT_NAN_OR_INFINITY));
        }

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

        let old_raw = self.get_string_raw(key_bytes)?;
        let (cur_num, old_expire) = match old_raw {
            Some((ref raw, exp, offset)) => (parse_redis_float(&raw[offset..])?, exp),
            None => (0.0f64, 0u64),
        };

        let mut new_num = cur_num + increment;
        if new_num.is_nan() || new_num.is_infinite() {
            return Err(Error::invalid_data(ERR_INCREMENT_NAN_OR_INFINITY));
        }
        if new_num == 0.0 {
            new_num = 0.0;
        }

        let final_expire = if keep_ttl { old_expire } else { expire_ms };
        let mut num_buf = zmij::Buffer::new();
        let formatted_bytes = format_float_bytes(new_num, &mut num_buf);
        let enc_val = encode_string_value(formatted_bytes, final_expire);
        self.data.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> {
        self.incrbyfloat_ex(key, increment, 0, true)
    }

    /// STRLEN key（零拷贝直接读取底层切片长度）
    pub fn strlen<K: AsRef<[u8]>>(&self, key: K) -> Result<usize> {
        match self.get_string_raw(key.as_ref())? {
            Some((raw, _, offset)) => Ok(raw.len() - offset),
            None => Ok(0),
        }
    }

    /// APPEND key value（1:1 对标 Kvrocks String::Append）
    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 old_raw = self.get_string_raw(key_bytes)?;
        let (cur_len, cur_expire) = match old_raw {
            Some((ref raw, exp, offset)) => (raw.len() - offset, exp),
            None => (0, 0),
        };

        let new_len = cur_len
            .checked_add(val_bytes.len())
            .ok_or_else(|| Error::invalid_data(ERR_STRING_EXCEEDS_MAX_SIZE))?;
        if new_len > MAX_STRING_SIZE {
            return Err(Error::invalid_data(ERR_STRING_EXCEEDS_MAX_SIZE));
        }

        let mut enc_val = Vec::with_capacity(STRING_HDR_SIZE + new_len);
        enc_val.extend_from_slice(&encode_string_header(cur_expire));
        if let Some((ref raw, _, offset)) = old_raw {
            enc_val.extend_from_slice(&raw[offset..]);
        }
        enc_val.extend_from_slice(val_bytes);

        self.data.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>> {
        match self.get_string_raw(key.as_ref())? {
            Some((raw, _, offset)) => {
                let payload = &raw[offset..];
                let len = payload.len() as i64;
                let (s, e) = normalize_range(start, end, len);
                if s > e || payload.is_empty() {
                    Ok(Vec::new())
                } else {
                    Ok(payload[s as usize..=e as usize].to_vec())
                }
            }
            None => Ok(Vec::new()),
        }
    }

    /// 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_OUT_OF_RANGE));
        }

        let required_len = offset
            .checked_add(val_bytes.len())
            .ok_or_else(|| Error::invalid_data(ERR_STRING_EXCEEDS_MAX_SIZE))?;
        if required_len > MAX_STRING_SIZE {
            return Err(Error::invalid_data(ERR_STRING_EXCEEDS_MAX_SIZE));
        }

        let old_raw = self.get_string_raw(key_bytes)?;
        if old_raw.is_none() && val_bytes.is_empty() {
            return Ok(0);
        }

        let (cur_slice, cur_expire) = match old_raw {
            Some((ref raw, exp, off)) => (&raw[off..], exp),
            None => (&[][..], 0),
        };

        let new_len = cur_slice.len().max(required_len);
        let mut enc_val = Vec::with_capacity(STRING_HDR_SIZE + new_len);
        enc_val.extend_from_slice(&encode_string_header(cur_expire));

        if cur_slice.len() < offset {
            enc_val.extend_from_slice(cur_slice);
            enc_val.resize(STRING_HDR_SIZE + offset, 0);
            enc_val.extend_from_slice(val_bytes);
        } else {
            enc_val.extend_from_slice(&cur_slice[..offset]);
            enc_val.extend_from_slice(val_bytes);
            if required_len < cur_slice.len() {
                enc_val.extend_from_slice(&cur_slice[required_len..]);
            }
        }

        self.data.insert(&*raw_k, enc_val)?;
        Ok(new_len)
    }

    /// MGET key [key ...]（批量读取，非 String 类型或不存在返回 None，1:1 对标 Kvrocks 与 Redis 规范）
    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 {
            match self.get_string_raw(k.as_ref()) {
                Ok(Some((raw, _, offset))) => results.push(Some(raw[offset..].to_vec())),
                Ok(None) => results.push(None),
                Err(ref e) if e.is_wrong_type() => results.push(None),
                Err(e) => return Err(e),
            }
        }
        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 参数控制版本（1:1 对标 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);
        }

        // 先校验所有 Value 大小，避免写批处理中途报错导致脏状态或不必要的计算
        for (_, v) in kvs {
            if v.as_ref().len() > MAX_STRING_SIZE {
                return Err(Error::invalid_data(ERR_STRING_EXCEEDS_MAX_SIZE));
            }
        }

        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 old_raw = self.get_string_raw(k.as_ref());
                match old_raw {
                    Ok(Some((_, exp, _))) => {
                        if args.set_type == StringSetType::Nx {
                            return Ok(false);
                        }
                        expires[i] = exp;
                    }
                    Ok(None) => {
                        if args.set_type == StringSetType::Xx {
                            return Ok(false);
                        }
                    }
                    Err(ref e) if e.is_wrong_type() => {
                        if args.set_type == StringSetType::Nx {
                            return Ok(false);
                        }
                        if args.set_type == StringSetType::Xx {
                            expires[i] = 0;
                        }
                    }
                    Err(e) => return Err(e),
                }
            }
        }

        let meta_is_empty = self.meta.is_empty()?;
        let mut batch = self.db.batch();

        for (i, (k, v)) in kvs.iter().enumerate() {
            let k_bytes = k.as_ref();
            let v_bytes = v.as_ref();
            let raw_k = kc.raw_key_bytes(k_bytes);
            let expire = if args.keep_ttl {
                expires[i]
            } else {
                args.expire
            };
            let enc_val = encode_string_value(v_bytes, expire);
            batch.insert(&self.data, &*raw_k, enc_val);

            // 仅在 meta 表非空时检查并清理潜在的复合结构元数据与前缀数据
            if !meta_is_empty {
                self.cleanup_all_composite_data(&kc, k_bytes, &mut batch)?;
            }
        }
        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_MAX_SIZE));
        }

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

        match self.get_string_raw(key_bytes)? {
            Some((raw, _, offset)) => {
                if &raw[offset..] == old_bytes {
                    let enc_val = encode_string_value(new_bytes, expire_ms);
                    self.data.insert(&*raw_k, enc_val)?;
                    Ok(1)
                } else {
                    Ok(0)
                }
            }
            None => Ok(-1),
        }
    }

    /// 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);

        match self.get_string_raw(key_bytes)? {
            Some((raw, _, offset)) => {
                if &raw[offset..] == val.as_ref() {
                    self.data.remove(&*raw_k)?;
                    Ok(1)
                } else {
                    Ok(0)
                }
            }
            None => Ok(-1),
        }
    }

    /// 计算指定 key 的哈希摘要（对标 Kvrocks String::Digest）
    pub fn digest<K: AsRef<[u8]>>(&self, key: K) -> Result<Option<String>> {
        match self.get_string_raw(key.as_ref())? {
            Some((raw, _, offset)) => Ok(Some(string_digest(&raw[offset..]))),
            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 old1 = self.get_string_raw(key1.as_ref())?;
        let old2 = self.get_string_raw(key2.as_ref())?;
        let s1 = old1.as_ref().map(|(r, _, o)| &r[*o..]).unwrap_or_default();
        let s2 = old2.as_ref().map(|(r, _, o)| &r[*o..]).unwrap_or_default();
        compute_lcs(s1, s2, args)
    }
}