wedb_embed 0.1.0

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

/// 列表结构元数据（对标 Apache Kvrocks ListMetadata 42字节 / 紧凑41字节）
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct ListMeta {
    pub base: KeyMeta,
    pub head: u64,
    pub tail: u64,
}

impl ListMeta {
    pub const INITIAL_INDEX: u64 = u64::MAX / 2; // 0x7fff_ffff_ffff_ffff
    pub const ENCODED_SIZE: usize = KeyMeta::ENCODED_SIZE + 16; // 26 + 16 = 42
    pub const KVROCKS_ENCODED_SIZE: usize = KeyMeta::KVROCKS_COMPLEX_ENCODED_SIZE + 16; // 25 + 16 = 41

    #[inline]
    pub fn new(expire_at: u64, version: u64) -> Self {
        Self {
            base: KeyMeta::new(RedisType::List, expire_at, version, 0),
            head: Self::INITIAL_INDEX,
            tail: Self::INITIAL_INDEX,
        }
    }

    #[inline]
    pub fn new_with_version(expire_at: u64) -> Self {
        Self {
            base: KeyMeta::new_with_version(RedisType::List, expire_at, 0),
            head: Self::INITIAL_INDEX,
            tail: Self::INITIAL_INDEX,
        }
    }

    #[inline]
    pub fn size(&self) -> u64 {
        self.base.size
    }

    #[inline]
    pub fn version(&self) -> u64 {
        self.base.version
    }

    #[inline]
    pub fn expire_at(&self) -> u64 {
        self.base.expire_at
    }

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

    #[inline]
    pub fn is_empty(&self) -> bool {
        self.base.size == 0
    }

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

    #[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());
        buf[KeyMeta::ENCODED_SIZE..KeyMeta::ENCODED_SIZE + 8]
            .copy_from_slice(&self.head.to_be_bytes());
        buf[KeyMeta::ENCODED_SIZE + 8..Self::ENCODED_SIZE]
            .copy_from_slice(&self.tail.to_be_bytes());
        buf
    }

    #[inline]
    pub fn encode_kvrocks(&self) -> Vec<u8> {
        let mut out = Vec::with_capacity(Self::KVROCKS_ENCODED_SIZE);
        out.extend_from_slice(&self.base.encode_kvrocks());
        out.extend_from_slice(&self.head.to_be_bytes());
        out.extend_from_slice(&self.tail.to_be_bytes());
        out
    }

    #[inline]
    pub fn decode(bytes: &[u8]) -> Option<Self> {
        if bytes.len() >= Self::ENCODED_SIZE
            && bytes[0] <= 14
            && (bytes[1] == 0 || bytes[1] == 0x80)
        {
            let base = KeyMeta::decode(&bytes[..KeyMeta::ENCODED_SIZE])?;
            let head = read_u64_be(bytes, KeyMeta::ENCODED_SIZE)?;
            let tail = read_u64_be(bytes, KeyMeta::ENCODED_SIZE + 8)?;
            Some(Self { base, head, tail })
        } else if bytes.len() >= Self::KVROCKS_ENCODED_SIZE
            && (bytes[0] & KeyMeta::META_64BIT_ENCODING_MASK != 0)
        {
            let base = KeyMeta::decode(&bytes[..KeyMeta::KVROCKS_COMPLEX_ENCODED_SIZE])?;
            let head = read_u64_be(bytes, KeyMeta::KVROCKS_COMPLEX_ENCODED_SIZE)?;
            let tail = read_u64_be(bytes, KeyMeta::KVROCKS_COMPLEX_ENCODED_SIZE + 8)?;
            Some(Self { base, head, tail })
        } else if bytes.len() >= 33 && (bytes[0] & KeyMeta::META_64BIT_ENCODING_MASK == 0) {
            // Kvrocks 32-bit 紧凑格式 (17字节 base + 16字节 head/tail)
            let base = KeyMeta::decode(&bytes[..17])?;
            let head = read_u64_be(bytes, 17)?;
            let tail = read_u64_be(bytes, 25)?;
            Some(Self { base, head, tail })
        } else if bytes.len() >= Self::ENCODED_SIZE {
            let base = KeyMeta::decode(&bytes[..KeyMeta::ENCODED_SIZE])?;
            let head = read_u64_be(bytes, KeyMeta::ENCODED_SIZE)?;
            let tail = read_u64_be(bytes, KeyMeta::ENCODED_SIZE + 8)?;
            Some(Self { base, head, tail })
        } else if bytes.len() == 16 {
            let head = read_u64_be(bytes, 0)?;
            let tail = read_u64_be(bytes, 8)?;
            let size = if tail >= head {
                tail.wrapping_sub(head)
            } else {
                0
            };
            Some(Self {
                base: KeyMeta::new(RedisType::List, 0, 0, size),
                head,
                tail,
            })
        } else {
            None
        }
    }
}

#[inline(always)]
fn read_u64_be(bytes: &[u8], offset: usize) -> Option<u64> {
    bytes
        .get(offset..offset + 8)?
        .try_into()
        .ok()
        .map(u64::from_be_bytes)
}

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

    #[test]
    fn test_list_meta_initial_index() {
        let meta = ListMeta::new(0, 100);
        assert_eq!(meta.head, ListMeta::INITIAL_INDEX);
        assert_eq!(meta.tail, ListMeta::INITIAL_INDEX);
        assert_eq!(meta.size(), 0);
        assert!(meta.is_empty());
    }

    #[test]
    fn test_list_meta_roundtrip_standard() {
        let mut meta = ListMeta::new(1234567890, 42);
        meta.base.size = 10;
        meta.head = ListMeta::INITIAL_INDEX - 5;
        meta.tail = ListMeta::INITIAL_INDEX + 5;

        let enc = meta.encode();
        assert_eq!(enc.len(), ListMeta::ENCODED_SIZE);

        let dec = ListMeta::decode(&enc).expect("decode failed");
        assert_eq!(dec.base.rtype, RedisType::List);
        assert_eq!(dec.base.expire_at, 1234567890);
        assert_eq!(dec.base.version, 42);
        assert_eq!(dec.base.size, 10);
        assert_eq!(dec.head, ListMeta::INITIAL_INDEX - 5);
        assert_eq!(dec.tail, ListMeta::INITIAL_INDEX + 5);
    }

    #[test]
    fn test_list_meta_roundtrip_kvrocks() {
        let mut meta = ListMeta::new(9876543210, 100);
        meta.base.size = 20;
        meta.head = ListMeta::INITIAL_INDEX - 10;
        meta.tail = ListMeta::INITIAL_INDEX + 10;

        let kv_enc = meta.encode_kvrocks();
        assert_eq!(kv_enc.len(), ListMeta::KVROCKS_ENCODED_SIZE);

        let dec = ListMeta::decode(&kv_enc).expect("decode kvrocks failed");
        assert_eq!(dec.base.rtype, RedisType::List);
        assert_eq!(dec.base.expire_at, 9876543210);
        assert_eq!(dec.base.version, 100);
        assert_eq!(dec.base.size, 20);
        assert_eq!(dec.head, ListMeta::INITIAL_INDEX - 10);
        assert_eq!(dec.tail, ListMeta::INITIAL_INDEX + 10);
    }

    #[test]
    fn test_list_meta_new_with_version_and_raw16() {
        let meta = ListMeta::new_with_version(5000);
        assert_eq!(meta.head, ListMeta::INITIAL_INDEX);
        assert_eq!(meta.tail, ListMeta::INITIAL_INDEX);
        assert_eq!(meta.base.expire_at, 5000);
        assert!(meta.base.version > 0);

        // 16 字节原始 head/tail 兼容解码
        let mut raw = [0u8; 16];
        raw[..8].copy_from_slice(&100u64.to_be_bytes());
        raw[8..].copy_from_slice(&105u64.to_be_bytes());
        let dec = ListMeta::decode(&raw).expect("decode 16-byte raw failed");
        assert_eq!(dec.head, 100);
        assert_eq!(dec.tail, 105);
        assert_eq!(dec.size(), 5);
    }
}