wedb_embed 0.1.0

use crate::meta::{KeyMeta, RedisType};
use serde::{Deserialize, Serialize};

/// 将 f64 编码为保持大小比较顺序的 u64（对标 Apache Kvrocks EncodeDoubleToUInt64）
#[inline]
pub const fn encode_double_to_u64(value: f64) -> u64 {
    let bits = value.to_bits();
    if (bits >> 63) == 1 {
        bits ^ 0xffff_ffff_ffff_ffff
    } else {
        bits | 0x8000_0000_0000_0000
    }
}

/// 从有序 u64 解码出 f64（对标 Apache Kvrocks DecodeDoubleFromUInt64）
#[inline]
pub const fn decode_double_from_u64(mut value: u64) -> f64 {
    if (value >> 63) == 0 {
        value ^= 0xffff_ffff_ffff_ffff;
    } else {
        value &= 0x7fff_ffff_ffff_ffff;
    }
    f64::from_bits(value)
}

/// T-Digest 结构元数据（对标 Apache Kvrocks TDigestMetadata，98 字节定长二进制编码）
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct TDigestMeta {
    pub base: KeyMeta,
    pub compression: u32,
    pub capacity: u32,
    pub unmerged_nodes: u64,
    pub merged_nodes: u64,
    pub total_weight: u64,
    pub merged_weight: u64,
    pub minimum: f64,
    pub maximum: f64,
    pub total_observations: u64,
    pub merge_times: u64,
}

impl TDigestMeta {
    /// 26 字节 KeyMeta + 8 字节 (compression+capacity) + 32 字节 (4*u64) + 16 字节 (2*f64) + 16 字节 (2*u64) = 98 字节
    pub const ENCODED_SIZE: usize = KeyMeta::ENCODED_SIZE + 72; // 98
    pub const KVROCKS_ENCODED_SIZE: usize = KeyMeta::KVROCKS_COMPLEX_ENCODED_SIZE + 72; // 97

    #[inline]
    pub fn new(compression: u32, expire_at: u64, version: u64) -> Self {
        let comp = if compression == 0 {
            super::conf::DEFAULT_COMPRESSION
        } else {
            compression
        };
        let capacity = super::conf::calculate_capacity(comp) as u32;
        Self {
            base: KeyMeta::new(RedisType::TDigest, expire_at, version, 0),
            compression: comp,
            capacity,
            unmerged_nodes: 0,
            merged_nodes: 0,
            total_weight: 0,
            merged_weight: 0,
            minimum: f64::MAX,
            maximum: -f64::MAX,
            total_observations: 0,
            merge_times: 0,
        }
    }

    /// 重置元数据统计（对标 Apache Kvrocks TDigest::Reset）
    #[inline]
    pub fn reset(&mut self) {
        self.unmerged_nodes = 0;
        self.merged_nodes = 0;
        self.total_weight = 0;
        self.merged_weight = 0;
        self.minimum = f64::MAX;
        self.maximum = -f64::MAX;
        self.total_observations = 0;
        self.merge_times = 0;
    }

    #[inline]
    pub fn is_expired(&self, now_ms: u64) -> bool {
        self.base.is_expired(now_ms)
    }

    #[inline]
    pub fn total_nodes(&self) -> u64 {
        self.merged_nodes + self.unmerged_nodes
    }

    #[inline]
    pub fn delta(&self) -> f64 {
        1.0 / (self.compression as f64)
    }

    /// 零拷贝栈数组定长序列化
    #[inline]
    pub fn encode(&self) -> [u8; Self::ENCODED_SIZE] {
        let mut buf = [0u8; Self::ENCODED_SIZE];
        buf[..KeyMeta::ENCODED_SIZE].copy_from_slice(&self.base.encode());
        let mut offset = KeyMeta::ENCODED_SIZE;

        buf[offset..offset + 4].copy_from_slice(&self.compression.to_be_bytes());
        offset += 4;
        buf[offset..offset + 4].copy_from_slice(&self.capacity.to_be_bytes());
        offset += 4;

        buf[offset..offset + 8].copy_from_slice(&self.unmerged_nodes.to_be_bytes());
        offset += 8;
        buf[offset..offset + 8].copy_from_slice(&self.merged_nodes.to_be_bytes());
        offset += 8;
        buf[offset..offset + 8].copy_from_slice(&self.total_weight.to_be_bytes());
        offset += 8;
        buf[offset..offset + 8].copy_from_slice(&self.merged_weight.to_be_bytes());
        offset += 8;

        buf[offset..offset + 8].copy_from_slice(&encode_double_to_u64(self.minimum).to_be_bytes());
        offset += 8;
        buf[offset..offset + 8].copy_from_slice(&encode_double_to_u64(self.maximum).to_be_bytes());
        offset += 8;

        buf[offset..offset + 8].copy_from_slice(&self.total_observations.to_be_bytes());
        offset += 8;
        buf[offset..offset + 8].copy_from_slice(&self.merge_times.to_be_bytes());

        buf
    }

    /// 兼容 Apache Kvrocks 二进制格式
    #[inline]
    pub fn encode_kvrocks(&self) -> Vec<u8> {
        let mut out = self.base.encode_kvrocks();
        out.extend_from_slice(&self.compression.to_be_bytes());
        out.extend_from_slice(&self.capacity.to_be_bytes());
        out.extend_from_slice(&self.unmerged_nodes.to_be_bytes());
        out.extend_from_slice(&self.merged_nodes.to_be_bytes());
        out.extend_from_slice(&self.total_weight.to_be_bytes());
        out.extend_from_slice(&self.merged_weight.to_be_bytes());
        out.extend_from_slice(&encode_double_to_u64(self.minimum).to_be_bytes());
        out.extend_from_slice(&encode_double_to_u64(self.maximum).to_be_bytes());
        out.extend_from_slice(&self.total_observations.to_be_bytes());
        out.extend_from_slice(&self.merge_times.to_be_bytes());
        out
    }

    /// 统一反序列化（支持 98 字节本引擎格式与 97 字节 Kvrocks 格式）
    #[inline]
    pub fn decode(bytes: &[u8]) -> Option<Self> {
        if bytes.len() < Self::KVROCKS_ENCODED_SIZE {
            return None;
        }
        let (base, mut offset) = if bytes.len() >= Self::ENCODED_SIZE
            && (bytes[1] == 0 || bytes[1] == 0x80)
            && bytes[0] <= 14
        {
            (
                KeyMeta::decode(&bytes[..KeyMeta::ENCODED_SIZE])?,
                KeyMeta::ENCODED_SIZE,
            )
        } else if bytes.len() >= Self::KVROCKS_ENCODED_SIZE {
            (
                KeyMeta::decode(&bytes[..KeyMeta::KVROCKS_COMPLEX_ENCODED_SIZE])?,
                KeyMeta::KVROCKS_COMPLEX_ENCODED_SIZE,
            )
        } else {
            return None;
        };

        if bytes.len() < offset + 72 {
            return None;
        }

        let mut u32_buf = [0u8; 4];
        u32_buf.copy_from_slice(&bytes[offset..offset + 4]);
        let compression = u32::from_be_bytes(u32_buf);
        offset += 4;

        u32_buf.copy_from_slice(&bytes[offset..offset + 4]);
        let capacity = u32::from_be_bytes(u32_buf);
        offset += 4;

        let mut u64_buf = [0u8; 8];

        u64_buf.copy_from_slice(&bytes[offset..offset + 8]);
        let unmerged_nodes = u64::from_be_bytes(u64_buf);
        offset += 8;

        u64_buf.copy_from_slice(&bytes[offset..offset + 8]);
        let merged_nodes = u64::from_be_bytes(u64_buf);
        offset += 8;

        u64_buf.copy_from_slice(&bytes[offset..offset + 8]);
        let total_weight = u64::from_be_bytes(u64_buf);
        offset += 8;

        u64_buf.copy_from_slice(&bytes[offset..offset + 8]);
        let merged_weight = u64::from_be_bytes(u64_buf);
        offset += 8;

        u64_buf.copy_from_slice(&bytes[offset..offset + 8]);
        let minimum = decode_double_from_u64(u64::from_be_bytes(u64_buf));
        offset += 8;

        u64_buf.copy_from_slice(&bytes[offset..offset + 8]);
        let maximum = decode_double_from_u64(u64::from_be_bytes(u64_buf));
        offset += 8;

        u64_buf.copy_from_slice(&bytes[offset..offset + 8]);
        let total_observations = u64::from_be_bytes(u64_buf);
        offset += 8;

        u64_buf.copy_from_slice(&bytes[offset..offset + 8]);
        let merge_times = u64::from_be_bytes(u64_buf);

        Some(Self {
            base,
            compression,
            capacity,
            unmerged_nodes,
            merged_nodes,
            total_weight,
            merged_weight,
            minimum,
            maximum,
            total_observations,
            merge_times,
        })
    }
}