wedb_embed 0.1.1

pub mod conf;
pub mod meta;

pub use conf::{ERR_INDEX_OUT_OF_RANGE, ERR_NO_SUCH_KEY, ERR_RANK_ZERO, ERR_WRONG_TYPE, PosSpec};
pub use meta::ListMeta;

use crate::db::WeDb;
use crate::error::{Error, Result};
use crate::key_composer::{KeyComposer, KeyTag, SmallKey, SubkeyComposer};
use crate::meta::current_now_ms;

/// 栈上定长构造 List 元数据键（零堆分配）
#[inline]
fn compose_list_meta_key(kc: &KeyComposer<'_>, key: &[u8]) -> SmallKey {
    kc.compose_meta_key_stack(KeyTag::ListMeta.as_slice(), key)
}

/// 栈上定长构造 List 数据子键前缀（零堆分配）
#[inline]
fn compose_list_prefix_stack(kc: &KeyComposer<'_>, key: &[u8]) -> SmallKey {
    kc.compose_prefix_stack(KeyTag::ListData.as_slice(), key)
}

/// 列表项键快速生成器（单次分配前缀，后续寻址零堆内存分配）
#[derive(Debug, Clone)]
pub struct ListItemKeyComposer {
    composer: SubkeyComposer,
}

impl ListItemKeyComposer {
    #[inline]
    pub fn new(kc: &KeyComposer<'_>, key: &[u8]) -> Self {
        let prefix = compose_list_prefix_stack(kc, key);
        Self {
            composer: SubkeyComposer::from_slice(&prefix),
        }
    }

    #[inline(always)]
    pub fn key_for_idx(&mut self, idx: u64) -> &[u8] {
        self.composer.key_for_hex_u64(idx)
    }
}

impl WeDb {
    /// 获取列表元数据（严格校验类型与过期状态，若为其他类型报错 WRONGTYPE，若过期或不存在返回 None）
    #[inline]
    fn get_list_meta_checked(
        &self,
        kc: &KeyComposer<'_>,
        k_bytes: &[u8],
        meta_k: &[u8],
        now_ms: u64,
    ) -> Result<Option<ListMeta>> {
        self.get_meta_checked::<ListMeta>(kc, k_bytes, meta_k, now_ms)
    }

    /// 准备写入时的列表元数据（若旧列表已过期则自动在批处理中清理残留子键）
    #[inline]
    fn prepare_list_meta_for_write(
        &self,
        kc: &KeyComposer<'_>,
        k_bytes: &[u8],
        meta_k: &[u8],
        now_ms: u64,
        batch: &mut fjall::OwnedWriteBatch,
    ) -> Result<(ListMeta, bool)> {
        match self.meta.get(meta_k)? {
            Some(m_bytes) => match ListMeta::decode(&m_bytes) {
                Some(meta) => {
                    if !meta.is_expired(now_ms) {
                        Ok((meta, true))
                    } else {
                        // 已过期：清理旧残留子键，防止脏数据影响新列表
                        let mut composer = ListItemKeyComposer::new(kc, k_bytes);
                        for i in 0..meta.base.size {
                            let old_idx = meta.head.wrapping_add(i);
                            batch.remove(&self.data, composer.key_for_idx(old_idx));
                        }
                        batch.remove(&self.meta, meta_k);
                        Ok((ListMeta::new_with_version(0), false))
                    }
                }
                None => {
                    self.check_key_not_other_type(
                        kc,
                        k_bytes,
                        KeyTag::ListMeta.as_slice(),
                        now_ms,
                    )?;
                    Ok((ListMeta::new_with_version(0), false))
                }
            },
            None => {
                self.check_key_not_other_type(kc, k_bytes, KeyTag::ListMeta.as_slice(), now_ms)?;
                Ok((ListMeta::new_with_version(0), false))
            }
        }
    }

    /// LPUSH key element [element ...]
    pub fn lpush<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, values: &[V]) -> Result<u64> {
        self.lpush_with_kc(&KeyComposer::new("default"), key, values)
    }

    pub fn lpush_with_kc<K: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        kc: &KeyComposer<'_>,
        key: K,
        values: &[V],
    ) -> Result<u64> {
        self.list_push_internal(kc, key.as_ref(), values, true, true)
    }

    /// LPUSH 单元素便捷方法
    #[inline]
    pub fn lpush_one<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, value: V) -> Result<u64> {
        self.lpush(key, &[value])
    }

    /// RPUSH key element [element ...]
    pub fn rpush<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, values: &[V]) -> Result<u64> {
        self.rpush_with_kc(&KeyComposer::new("default"), key, values)
    }

    pub fn rpush_with_kc<K: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        kc: &KeyComposer<'_>,
        key: K,
        values: &[V],
    ) -> Result<u64> {
        self.list_push_internal(kc, key.as_ref(), values, true, false)
    }

    /// RPUSH 单元素便捷方法
    #[inline]
    pub fn rpush_one<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, value: V) -> Result<u64> {
        self.rpush(key, &[value])
    }

    /// LPUSHX key element [element ...] (仅在列表存在且未过期时压入)
    pub fn lpushx<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, values: &[V]) -> Result<u64> {
        self.lpushx_with_kc(&KeyComposer::new("default"), key, values)
    }

    pub fn lpushx_with_kc<K: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        kc: &KeyComposer<'_>,
        key: K,
        values: &[V],
    ) -> Result<u64> {
        self.list_push_internal(kc, key.as_ref(), values, false, true)
    }

    /// LPUSHX 单元素便捷方法
    #[inline]
    pub fn lpushx_one<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, value: V) -> Result<u64> {
        self.lpushx(key, &[value])
    }

    /// RPUSHX key element [element ...] (仅在列表存在且未过期时压入)
    pub fn rpushx<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, values: &[V]) -> Result<u64> {
        self.rpushx_with_kc(&KeyComposer::new("default"), key, values)
    }

    pub fn rpushx_with_kc<K: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        kc: &KeyComposer<'_>,
        key: K,
        values: &[V],
    ) -> Result<u64> {
        self.list_push_internal(kc, key.as_ref(), values, false, false)
    }

    /// RPUSHX 单元素便捷方法
    #[inline]
    pub fn rpushx_one<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, value: V) -> Result<u64> {
        self.rpushx(key, &[value])
    }

    fn list_push_internal<V: AsRef<[u8]>>(
        &self,
        kc: &KeyComposer<'_>,
        key: &[u8],
        values: &[V],
        create_if_missing: bool,
        left: bool,
    ) -> Result<u64> {
        if values.is_empty() {
            return self.llen_with_kc(kc, key);
        }
        let meta_k = compose_list_meta_key(kc, key);
        let now_ms = current_now_ms();

        let mut batch = self.db.batch();
        let (mut meta, exists) =
            self.prepare_list_meta_for_write(kc, key, &meta_k, now_ms, &mut batch)?;

        if !exists && !create_if_missing {
            if !batch.is_empty() {
                batch.commit()?;
            }
            return Ok(0);
        }

        let mut composer = ListItemKeyComposer::new(kc, key);
        if left {
            for val in values {
                meta.head = meta.head.wrapping_sub(1);
                meta.base.size = meta.base.size.saturating_add(1);
                let item_k = composer.key_for_idx(meta.head);
                batch.insert(&self.data, item_k, val.as_ref());
            }
        } else {
            for val in values {
                let item_k = composer.key_for_idx(meta.tail);
                batch.insert(&self.data, item_k, val.as_ref());
                meta.tail = meta.tail.wrapping_add(1);
                meta.base.size = meta.base.size.saturating_add(1);
            }
        }

        batch.insert(&self.meta, &meta_k, meta.encode());
        batch.commit()?;
        Ok(meta.base.size)
    }

    /// LPOP key [count]
    pub fn lpop<K: AsRef<[u8]>>(&self, key: K, count: usize) -> Result<Vec<Vec<u8>>> {
        self.lpop_with_kc(&KeyComposer::new("default"), key, count)
    }

    pub fn lpop_with_kc<K: AsRef<[u8]>>(
        &self,
        kc: &KeyComposer<'_>,
        key: K,
        count: usize,
    ) -> Result<Vec<Vec<u8>>> {
        let key_bytes = key.as_ref();
        let meta_k = compose_list_meta_key(kc, key_bytes);
        let now_ms = current_now_ms();

        let mut meta = match self.get_list_meta_checked(kc, key_bytes, &meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(Vec::new()),
        };

        if meta.base.size == 0 || count == 0 {
            return Ok(Vec::new());
        }

        let num_pop = (count as u64).min(meta.base.size) as usize;
        let mut results = Vec::with_capacity(num_pop);
        let mut batch = self.db.batch();
        let mut composer = ListItemKeyComposer::new(kc, key_bytes);

        for _ in 0..num_pop {
            let item_k = composer.key_for_idx(meta.head);
            if let Some(val) = self.data.get(item_k)? {
                results.push(val.to_vec());
            }
            batch.remove(&self.data, item_k);
            meta.head = meta.head.wrapping_add(1);
            meta.base.size = meta.base.size.saturating_sub(1);
        }

        if meta.base.size == 0 {
            batch.remove(&self.meta, &meta_k);
        } else {
            batch.insert(&self.meta, &meta_k, meta.encode());
        }

        batch.commit()?;
        Ok(results)
    }

    /// LPOP 单元素便捷方法（弹出首个元素）
    pub fn lpop_one<K: AsRef<[u8]>>(&self, key: K) -> Result<Option<Vec<u8>>> {
        let mut popped = self.lpop(key, 1)?;
        Ok(popped.pop())
    }

    /// RPOP key [count]
    pub fn rpop<K: AsRef<[u8]>>(&self, key: K, count: usize) -> Result<Vec<Vec<u8>>> {
        self.rpop_with_kc(&KeyComposer::new("default"), key, count)
    }

    pub fn rpop_with_kc<K: AsRef<[u8]>>(
        &self,
        kc: &KeyComposer<'_>,
        key: K,
        count: usize,
    ) -> Result<Vec<Vec<u8>>> {
        let key_bytes = key.as_ref();
        let meta_k = compose_list_meta_key(kc, key_bytes);
        let now_ms = current_now_ms();

        let mut meta = match self.get_list_meta_checked(kc, key_bytes, &meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(Vec::new()),
        };

        if meta.base.size == 0 || count == 0 {
            return Ok(Vec::new());
        }

        let num_pop = (count as u64).min(meta.base.size) as usize;
        let mut results = Vec::with_capacity(num_pop);
        let mut batch = self.db.batch();
        let mut composer = ListItemKeyComposer::new(kc, key_bytes);

        for _ in 0..num_pop {
            meta.tail = meta.tail.wrapping_sub(1);
            meta.base.size = meta.base.size.saturating_sub(1);
            let item_k = composer.key_for_idx(meta.tail);
            if let Some(val) = self.data.get(item_k)? {
                results.push(val.to_vec());
            }
            batch.remove(&self.data, item_k);
        }

        if meta.base.size == 0 {
            batch.remove(&self.meta, &meta_k);
        } else {
            batch.insert(&self.meta, &meta_k, meta.encode());
        }

        batch.commit()?;
        Ok(results)
    }

    /// RPOP 单元素便捷方法（弹出尾部首个元素）
    pub fn rpop_one<K: AsRef<[u8]>>(&self, key: K) -> Result<Option<Vec<u8>>> {
        let mut popped = self.rpop(key, 1)?;
        Ok(popped.pop())
    }

    /// LLEN key
    pub fn llen<K: AsRef<[u8]>>(&self, key: K) -> Result<u64> {
        self.llen_with_kc(&KeyComposer::new("default"), key)
    }

    pub fn llen_with_kc<K: AsRef<[u8]>>(&self, kc: &KeyComposer<'_>, key: K) -> Result<u64> {
        let key_bytes = key.as_ref();
        let meta_k = compose_list_meta_key(kc, key_bytes);
        let now_ms = current_now_ms();

        Ok(self
            .get_list_meta_checked(kc, key_bytes, &meta_k, now_ms)?
            .map_or(0, |m| m.base.size))
    }

    /// LRANGE key start stop (对标 Apache Kvrocks 边界与正负索引规则)
    pub fn lrange<K: AsRef<[u8]>>(&self, key: K, start: i64, stop: i64) -> Result<Vec<Vec<u8>>> {
        self.lrange_with_kc(&KeyComposer::new("default"), key, start, stop)
    }

    pub fn lrange_with_kc<K: AsRef<[u8]>>(
        &self,
        kc: &KeyComposer<'_>,
        key: K,
        start: i64,
        stop: i64,
    ) -> Result<Vec<Vec<u8>>> {
        let key_bytes = key.as_ref();
        let meta_k = compose_list_meta_key(kc, key_bytes);
        let now_ms = current_now_ms();

        let meta = match self.get_list_meta_checked(kc, key_bytes, &meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(Vec::new()),
        };

        if meta.base.size == 0 {
            return Ok(Vec::new());
        }

        let len = meta.base.size as i64;
        let mut s = if start < 0 {
            len.checked_add(start).unwrap_or(i64::MIN)
        } else {
            start
        };
        let mut e = if stop < 0 {
            len.checked_add(stop).unwrap_or(i64::MIN)
        } else {
            stop
        };

        if s >= len || e < 0 || s > e {
            return Ok(Vec::new());
        }

        if s < 0 {
            s = 0;
        }
        if e >= len {
            e = len - 1;
        }

        let num_elems = (e - s + 1) as usize;
        let mut results = Vec::with_capacity(num_elems);
        let mut composer = ListItemKeyComposer::new(kc, key_bytes);
        for idx in s..=e {
            let actual_idx = meta.head.wrapping_add(idx as u64);
            let item_k = composer.key_for_idx(actual_idx);
            if let Some(val) = self.data.get(item_k)? {
                results.push(val.to_vec());
            }
        }
        Ok(results)
    }

    /// LINDEX key index (单键直接寻址，零冗余分配)
    pub fn lindex<K: AsRef<[u8]>>(&self, key: K, index: i64) -> Result<Option<Vec<u8>>> {
        self.lindex_with_kc(&KeyComposer::new("default"), key, index)
    }

    pub fn lindex_with_kc<K: AsRef<[u8]>>(
        &self,
        kc: &KeyComposer<'_>,
        key: K,
        index: i64,
    ) -> Result<Option<Vec<u8>>> {
        let key_bytes = key.as_ref();
        let meta_k = compose_list_meta_key(kc, key_bytes);
        let now_ms = current_now_ms();

        let meta = match self.get_list_meta_checked(kc, key_bytes, &meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(None),
        };

        if meta.base.size == 0 {
            return Ok(None);
        }

        let len = meta.base.size as i64;
        let actual_offset = if index < 0 {
            len.checked_add(index).unwrap_or(i64::MIN)
        } else {
            index
        };

        if actual_offset < 0 || actual_offset >= len {
            return Ok(None);
        }

        let mut composer = ListItemKeyComposer::new(kc, key_bytes);
        let actual_idx = meta.head.wrapping_add(actual_offset as u64);
        let item_k = composer.key_for_idx(actual_idx);
        let val = self.data.get(item_k)?;
        Ok(val.map(|v| v.to_vec()))
    }

    /// LSET key index element (对标 Kvrocks 相同值跳写优化)
    pub fn lset<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, index: i64, value: V) -> Result<()> {
        self.lset_with_kc(&KeyComposer::new("default"), key, index, value)
    }

    pub fn lset_with_kc<K: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        kc: &KeyComposer<'_>,
        key: K,
        index: i64,
        value: V,
    ) -> Result<()> {
        let key_bytes = key.as_ref();
        let meta_k = compose_list_meta_key(kc, key_bytes);
        let now_ms = current_now_ms();

        let meta = match self.get_list_meta_checked(kc, key_bytes, &meta_k, now_ms)? {
            Some(m) => m,
            None => return Err(Error::invalid_data(ERR_NO_SUCH_KEY)),
        };

        let len = meta.base.size as i64;
        let actual_offset = if index < 0 {
            len.checked_add(index).unwrap_or(i64::MIN)
        } else {
            index
        };

        if actual_offset < 0 || actual_offset >= len {
            return Err(Error::invalid_data(ERR_INDEX_OUT_OF_RANGE));
        }

        let mut composer = ListItemKeyComposer::new(kc, key_bytes);
        let actual_idx = meta.head.wrapping_add(actual_offset as u64);
        let item_k = composer.key_for_idx(actual_idx);

        let val_bytes = value.as_ref();
        if let Some(existing) = self.data.get(item_k)?
            && existing.as_ref() == val_bytes
        {
            return Ok(());
        }

        self.data.insert(item_k, val_bytes)?;
        Ok(())
    }

    /// LINSERT key BEFORE | AFTER pivot element (对标 Kvrocks 最小位移量算法)
    pub fn linsert<K: AsRef<[u8]>, P: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        key: K,
        before: bool,
        pivot: P,
        elem: V,
    ) -> Result<i64> {
        let key_bytes = key.as_ref();
        let kc = KeyComposer::new("default");
        let meta_k = compose_list_meta_key(&kc, key_bytes);
        let now_ms = current_now_ms();

        let mut meta = match self.get_list_meta_checked(&kc, key_bytes, &meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(0),
        };

        if meta.base.size == 0 {
            return Ok(0);
        }

        let len = meta.base.size as usize;
        let pivot_bytes = pivot.as_ref();
        let mut composer = ListItemKeyComposer::new(&kc, key_bytes);

        let mut pivot_offset = None;
        for offset in 0..len {
            let idx = meta.head.wrapping_add(offset as u64);
            let item_k = composer.key_for_idx(idx);
            if let Some(v) = self.data.get(item_k)?
                && v.as_ref() == pivot_bytes
            {
                pivot_offset = Some(offset);
                break;
            }
        }

        let p_off = match pivot_offset {
            Some(off) => off,
            None => return Ok(-1),
        };

        let target_insert_offset = if before { p_off } else { p_off + 1 };
        let left_cost = target_insert_offset;
        let right_cost = len - target_insert_offset;

        let mut batch = self.db.batch();

        if left_cost <= right_cost {
            // 左端向左位移 1 个单位
            for offset in 0..target_insert_offset {
                let from_idx = meta.head.wrapping_add(offset as u64);
                let to_idx = from_idx.wrapping_sub(1);
                if let Some(v) = self.data.get(composer.key_for_idx(from_idx))? {
                    batch.insert(&self.data, composer.key_for_idx(to_idx), v.as_ref());
                }
            }
            let insert_idx = meta
                .head
                .wrapping_add(target_insert_offset as u64)
                .wrapping_sub(1);
            batch.insert(&self.data, composer.key_for_idx(insert_idx), elem.as_ref());
            meta.head = meta.head.wrapping_sub(1);
        } else {
            // 右端向右位移 1 个单位
            for offset in (target_insert_offset..len).rev() {
                let from_idx = meta.head.wrapping_add(offset as u64);
                let to_idx = from_idx.wrapping_add(1);
                if let Some(v) = self.data.get(composer.key_for_idx(from_idx))? {
                    batch.insert(&self.data, composer.key_for_idx(to_idx), v.as_ref());
                }
            }
            let insert_idx = meta.head.wrapping_add(target_insert_offset as u64);
            batch.insert(&self.data, composer.key_for_idx(insert_idx), elem.as_ref());
            meta.tail = meta.tail.wrapping_add(1);
        }

        meta.base.size += 1;
        batch.insert(&self.meta, &meta_k, meta.encode());
        batch.commit()?;

        Ok(meta.base.size as i64)
    }

    /// LREM key count element (对标 Kvrocks 最小位移量算法)
    pub fn lrem<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, count: i64, elem: V) -> Result<u64> {
        let key_bytes = key.as_ref();
        let kc = KeyComposer::new("default");
        let meta_k = compose_list_meta_key(&kc, key_bytes);
        let now_ms = current_now_ms();

        let mut meta = match self.get_list_meta_checked(&kc, key_bytes, &meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(0),
        };

        if meta.base.size == 0 {
            return Ok(0);
        }

        let len = meta.base.size as usize;
        let target_del_limit = if count == 0 {
            usize::MAX
        } else {
            count.unsigned_abs() as usize
        };

        let elem_bytes = elem.as_ref();
        let mut to_delete_offsets = Vec::new();
        let mut composer = ListItemKeyComposer::new(&kc, key_bytes);

        if count >= 0 {
            for offset in 0..len {
                let idx = meta.head.wrapping_add(offset as u64);
                let item_k = composer.key_for_idx(idx);
                if let Some(v) = self.data.get(item_k)?
                    && v.as_ref() == elem_bytes
                {
                    to_delete_offsets.push(offset);
                    if to_delete_offsets.len() >= target_del_limit {
                        break;
                    }
                }
            }
        } else {
            for step in 0..len {
                let offset = len - 1 - step;
                let idx = meta.head.wrapping_add(offset as u64);
                let item_k = composer.key_for_idx(idx);
                if let Some(v) = self.data.get(item_k)?
                    && v.as_ref() == elem_bytes
                {
                    to_delete_offsets.push(offset);
                    if to_delete_offsets.len() >= target_del_limit {
                        break;
                    }
                }
            }
            to_delete_offsets.reverse();
        }

        if to_delete_offsets.is_empty() {
            return Ok(0);
        }

        let del_cnt = to_delete_offsets.len();
        let mut batch = self.db.batch();

        if del_cnt == len {
            for offset in 0..len {
                let idx = meta.head.wrapping_add(offset as u64);
                batch.remove(&self.data, composer.key_for_idx(idx));
            }
            batch.remove(&self.meta, &meta_k);
            batch.commit()?;
            return Ok(del_cnt as u64);
        }

        let min_del_offset = to_delete_offsets[0];
        let max_del_offset = to_delete_offsets[del_cnt - 1];
        let left_cost = max_del_offset;
        let right_cost = len - 1 - min_del_offset;

        if left_cost <= right_cost {
            let mut target_offset = max_del_offset;
            let mut del_idx_cursor = del_cnt;

            for offset in (0..=max_del_offset).rev() {
                if del_idx_cursor > 0 && to_delete_offsets[del_idx_cursor - 1] == offset {
                    del_idx_cursor -= 1;
                } else {
                    if target_offset != offset {
                        let from_idx = meta.head.wrapping_add(offset as u64);
                        let to_idx = meta.head.wrapping_add(target_offset as u64);
                        if let Some(v) = self.data.get(composer.key_for_idx(from_idx))? {
                            batch.insert(&self.data, composer.key_for_idx(to_idx), v.as_ref());
                        }
                    }
                    target_offset = target_offset.saturating_sub(1);
                }
            }

            for offset in 0..del_cnt {
                let idx = meta.head.wrapping_add(offset as u64);
                batch.remove(&self.data, composer.key_for_idx(idx));
            }

            meta.head = meta.head.wrapping_add(del_cnt as u64);
        } else {
            let mut target_offset = min_del_offset;
            let mut del_idx_cursor = 0;

            for offset in min_del_offset..len {
                if del_idx_cursor < del_cnt && to_delete_offsets[del_idx_cursor] == offset {
                    del_idx_cursor += 1;
                } else {
                    if target_offset != offset {
                        let from_idx = meta.head.wrapping_add(offset as u64);
                        let to_idx = meta.head.wrapping_add(target_offset as u64);
                        if let Some(v) = self.data.get(composer.key_for_idx(from_idx))? {
                            batch.insert(&self.data, composer.key_for_idx(to_idx), v.as_ref());
                        }
                    }
                    target_offset += 1;
                }
            }

            for offset in (len - del_cnt)..len {
                let idx = meta.head.wrapping_add(offset as u64);
                batch.remove(&self.data, composer.key_for_idx(idx));
            }

            meta.tail = meta.tail.wrapping_sub(del_cnt as u64);
        }

        meta.base.size -= del_cnt as u64;
        batch.insert(&self.meta, &meta_k, meta.encode());
        batch.commit()?;

        Ok(del_cnt as u64)
    }

    /// LTRIM key start stop (对标 Apache Kvrocks 边界与正负索引规则)
    pub fn ltrim<K: AsRef<[u8]>>(&self, key: K, start: i64, stop: i64) -> Result<()> {
        self.ltrim_with_kc(&KeyComposer::new("default"), key, start, stop)
    }

    pub fn ltrim_with_kc<K: AsRef<[u8]>>(
        &self,
        kc: &KeyComposer<'_>,
        key: K,
        start: i64,
        stop: i64,
    ) -> Result<()> {
        let key_bytes = key.as_ref();
        let meta_k = compose_list_meta_key(kc, key_bytes);
        let now_ms = current_now_ms();

        let mut meta = match self.get_list_meta_checked(kc, key_bytes, &meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(()),
        };

        if meta.base.size == 0 {
            return Ok(());
        }

        let len = meta.base.size as i64;
        let mut s = if start < 0 {
            len.checked_add(start).unwrap_or(i64::MIN)
        } else {
            start
        };
        let mut e = if stop < 0 {
            len.checked_add(stop).unwrap_or(i64::MIN)
        } else {
            stop
        };

        let mut batch = self.db.batch();
        let mut composer = ListItemKeyComposer::new(kc, key_bytes);

        if s > e || s >= len || e < 0 {
            for i in 0..meta.base.size {
                let idx = meta.head.wrapping_add(i);
                batch.remove(&self.data, composer.key_for_idx(idx));
            }
            batch.remove(&self.meta, &meta_k);
            batch.commit()?;
            return Ok(());
        }

        if s < 0 {
            s = 0;
        }
        if e >= len {
            e = len - 1;
        }

        for i in 0..s {
            let idx = meta.head.wrapping_add(i as u64);
            batch.remove(&self.data, composer.key_for_idx(idx));
        }
        for i in (e + 1)..len {
            let idx = meta.head.wrapping_add(i as u64);
            batch.remove(&self.data, composer.key_for_idx(idx));
        }

        let new_size = (e - s + 1) as u64;
        meta.head = meta.head.wrapping_add(s as u64);
        meta.tail = meta.head.wrapping_add(new_size);
        meta.base.size = new_size;

        if new_size == 0 {
            batch.remove(&self.meta, &meta_k);
        } else {
            batch.insert(&self.meta, &meta_k, meta.encode());
        }

        batch.commit()?;
        Ok(())
    }

    /// LMOVE source destination LEFT | RIGHT LEFT | RIGHT (对标 Apache Kvrocks LMove)
    pub fn lmove<S: AsRef<[u8]>, D: AsRef<[u8]>>(
        &self,
        src: S,
        dst: D,
        src_left: bool,
        dst_left: bool,
    ) -> Result<Option<Vec<u8>>> {
        let src_bytes = src.as_ref();
        let dst_bytes = dst.as_ref();
        let now_ms = current_now_ms();

        let kc = KeyComposer::new("default");
        let src_meta_k = compose_list_meta_key(&kc, src_bytes);
        let mut src_meta = match self.get_list_meta_checked(&kc, src_bytes, &src_meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(None),
        };

        if src_meta.base.size == 0 {
            return Ok(None);
        }

        if src_bytes == dst_bytes {
            // 同列表轮转
            let mut composer = ListItemKeyComposer::new(&kc, src_bytes);
            let curr_idx = if src_left {
                src_meta.head
            } else {
                src_meta.tail.wrapping_sub(1)
            };
            let elem = match self.data.get(composer.key_for_idx(curr_idx))? {
                Some(v) => v.to_vec(),
                None => return Ok(None),
            };

            if src_left == dst_left || src_meta.base.size == 1 {
                return Ok(Some(elem));
            }

            let mut batch = self.db.batch();
            batch.remove(&self.data, composer.key_for_idx(curr_idx));

            if src_left {
                let new_tail_idx = src_meta.tail;
                batch.insert(&self.data, composer.key_for_idx(new_tail_idx), &elem);
                src_meta.head = src_meta.head.wrapping_add(1);
                src_meta.tail = src_meta.tail.wrapping_add(1);
            } else {
                let new_head_idx = src_meta.head.wrapping_sub(1);
                batch.insert(&self.data, composer.key_for_idx(new_head_idx), &elem);
                src_meta.head = src_meta.head.wrapping_sub(1);
                src_meta.tail = src_meta.tail.wrapping_sub(1);
            }

            batch.insert(&self.meta, &src_meta_k, src_meta.encode());
            batch.commit()?;
            return Ok(Some(elem));
        }

        // 双列表原子迁移
        let dst_meta_k = compose_list_meta_key(&kc, dst_bytes);
        let mut batch = self.db.batch();
        let (mut dst_meta, _) =
            self.prepare_list_meta_for_write(&kc, dst_bytes, &dst_meta_k, now_ms, &mut batch)?;

        let mut src_composer = ListItemKeyComposer::new(&kc, src_bytes);
        let mut dst_composer = ListItemKeyComposer::new(&kc, dst_bytes);

        let src_idx = if src_left {
            src_meta.head
        } else {
            src_meta.tail.wrapping_sub(1)
        };
        let elem = match self.data.get(src_composer.key_for_idx(src_idx))? {
            Some(v) => v.to_vec(),
            None => return Ok(None),
        };

        batch.remove(&self.data, src_composer.key_for_idx(src_idx));

        if src_left {
            src_meta.head = src_meta.head.wrapping_add(1);
        } else {
            src_meta.tail = src_meta.tail.wrapping_sub(1);
        }
        src_meta.base.size -= 1;

        if src_meta.base.size == 0 {
            batch.remove(&self.meta, &src_meta_k);
        } else {
            batch.insert(&self.meta, &src_meta_k, src_meta.encode());
        }

        let dst_idx = if dst_left {
            let idx = dst_meta.head.wrapping_sub(1);
            dst_meta.head = idx;
            idx
        } else {
            let idx = dst_meta.tail;
            dst_meta.tail = idx.wrapping_add(1);
            idx
        };

        batch.insert(&self.data, dst_composer.key_for_idx(dst_idx), &elem);
        dst_meta.base.size += 1;
        batch.insert(&self.meta, &dst_meta_k, dst_meta.encode());

        batch.commit()?;
        Ok(Some(elem))
    }

    /// RPOPLPUSH source destination (原子弹出右端并压入左端)
    #[inline]
    pub fn rpoplpush<S: AsRef<[u8]>, D: AsRef<[u8]>>(
        &self,
        source: S,
        destination: D,
    ) -> Result<Option<Vec<u8>>> {
        self.lmove(source, destination, false, true)
    }

    /// LPOS key element [RANK rank] [COUNT num-matches] [MAXLEN len] (对标 Apache Kvrocks PosSpec)
    pub fn lpos<K: AsRef<[u8]>, V: AsRef<[u8]>>(
        &self,
        key: K,
        elem: V,
        spec: PosSpec,
    ) -> Result<Vec<i64>> {
        if spec.rank == 0 {
            return Err(Error::invalid_data(ERR_RANK_ZERO));
        }

        let key_bytes = key.as_ref();
        let kc = KeyComposer::new("default");
        let meta_k = compose_list_meta_key(&kc, key_bytes);
        let now_ms = current_now_ms();

        let meta = match self.get_list_meta_checked(&kc, key_bytes, &meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(Vec::new()),
        };

        if meta.base.size == 0 {
            return Ok(Vec::new());
        }

        let len = meta.base.size as usize;
        let reversed = spec.rank < 0;
        let target_rank = spec.rank.unsigned_abs() as usize;
        let limit = spec
            .max_len
            .map(|m| if m == 0 { len } else { m.min(len) })
            .unwrap_or(len);

        let elem_bytes = elem.as_ref();
        let mut matches = match spec.count {
            Some(c) if c > 0 => Vec::with_capacity(c.min(limit)),
            None => Vec::with_capacity(1),
            _ => Vec::new(),
        };
        let mut match_count = 0;
        let mut composer = ListItemKeyComposer::new(&kc, key_bytes);

        for i in 0..limit {
            let offset = if !reversed { i } else { len - 1 - i };
            let idx = meta.head.wrapping_add(offset as u64);
            let item_k = composer.key_for_idx(idx);
            if let Some(v) = self.data.get(item_k)?
                && v.as_ref() == elem_bytes
            {
                match_count += 1;
                if match_count >= target_rank {
                    matches.push(offset as i64);
                    if let Some(c) = spec.count {
                        if c > 0 && matches.len() >= c {
                            break;
                        }
                    } else {
                        break;
                    }
                }
            }
        }

        Ok(matches)
    }

    // 通过宏生成 lexpireat/lexpire/lpexpireat/lpexpire/lttl/lpttl/lexpiretime/lpexpiretime/lpersist
    impl_expire_ops! {
        prefix = l,
        meta_type = ListMeta
    }
}