wedb_embed 0.1.2

pub mod conf;
pub mod r#impl;
pub mod key;
pub mod meta;
pub mod traits;

use std::{
  mem::{size_of, swap},
  str,
};

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;
pub use traits::Kv;
pub use wedb_resp::parse_i64_fast;

use crate::{
  error::{Error, Result},
  key::check_composite_meta_not_other_type,
  meta::{
    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,
  },
  traits::DbLike,
};

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

pub use key::{
  prefix as compose_string_prefix, raw as compose_string_key, raw_bytes as compose_string_key_bytes,
};

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);
  // SAFETY: u64_to_hex_16 仅生成 '0'..='9' 与 'a'..='f' 的 ASCII 字符，必为有效 UTF-8。
  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);
  str::from_utf8(bytes).unwrap_or("").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::default())),
      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 => Ok(StringLCSResult::Idx(StringLCSIdxResult {
        matches: vec![StringLCSMatchedRange::new(
          0,
          lcs_len - 1,
          0,
          lcs_len - 1,
          lcs_len,
        )],
        len: lcs_len,
      })),
      StringLCSType::None => {
        let s = match String::from_utf8(s1.to_vec()) {
          Ok(s) => s,
          Err(e) => String::from_utf8_lossy(e.as_bytes()).into_owned(),
        };
        Ok(StringLCSResult::Str(s))
      }
    };
  }

  // 针对仅计算长度的快速通道，采用双行滚动数组将空间复杂度由 O(M*N) 降至 O(min(M, N))
  if args.lcs_type == StringLCSType::Len {
    let (s1, s2, _alen, blen) = if alen < blen {
      (s2, s1, blen, alen)
    } else {
      (s1, s2, alen, blen)
    };
    let mut prev = vec![0u32; blen + 1];
    let mut curr = vec![0u32; blen + 1];
    for &b1 in s1 {
      for j in 1..=blen {
        if b1 == s2[j - 1] {
          curr[j] = prev[j - 1] + 1;
        } else {
          curr[j] = prev[j].max(curr[j - 1]);
        }
      }
      swap(&mut prev, &mut curr);
    }
    return Ok(StringLCSResult::Len(prev[blen]));
  }

  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(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;

  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 = match String::from_utf8(lcs_bytes) {
        Ok(s) => s,
        Err(e) => String::from_utf8_lossy(e.as_bytes()).into_owned(),
      };
      Ok(StringLCSResult::Str(s))
    }
  }
}

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

  let data_ks = db.data();

  if let Some(raw) = data_ks.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)));
    }
  }

  check_composite_meta_not_other_type(db, key_bytes, b"", now_ms)?;
  Ok(None)
}