storage-engines 0.1.0

//! 高级 KV 操作（独立模块）：批量、范围扫描、模糊搜索分页、元数据、TTL。
//!
//! 全部建立在 [`crate::mvcc::Transaction`] 的快照可见性之上。
//!
//! ## 能力一览
//!
//! | 类别 | API |
//! |------|-----|
//! | 批量 | `batch_set` / `batch_put` / `batch_delete` / `multi_get` |
//! | 计数器 | `incr` / `decr` / `incr_with_ttl` |
//! | 范围 | `scan` / `prefix_scan` / `reverse_scan` / `seek` / `seek_prev` |
//! | 搜索分页 | `search_keys`（`SearchQuery` / `KeyMatchMode` / `SearchPage`） |
//! | 元数据 | `exists` / `get_meta` / `key_count` |
//! | TTL | `set_with_ttl` / `refresh_ttl` / `get_ttl` / `persist` / `purge_expired` |
//!
//! ## TTL 存储约定
//!
//! 逻辑 value 可带信封：
//!
//! ```text
//! 永久： [0x00] user_payload...
//! 带过期：[0x01] expire_unix_secs:u64 LE | user_payload...
//! ```
//!
//! 旧数据无 tag 时按「永久裸 value」兼容。
//! 读路径（`get` / scan / multi_get / search）自动剥信封并丢弃已过期条目。

use std::time::Duration;

use crate::bitcask::bitcask_mvcc::{encode_key, MVCC, Transaction};
use crate::common::codec::parse_i64_user;

/// 公共类型与编解码工具现由 [`crate::common`] 统一定义。
///
/// 这里按原路径重导出，保证 `bitcask::kv_ops::pack_plain` 等既有调用点不变。
pub use crate::common::{
    decode_value_meta, key_matches, logical_to_user, now_unix_secs, pack_plain, pack_ttl,
    prefix_upper_bound, ExportRecord, IncrError, KeyMatchMode, KeyMeta, SearchPage, SearchQuery,
    ValueMeta,
};

// ─── Transaction 扩展 ───────────────────────────────────────────────────────

impl Transaction {
    // ── 批量写 ──

    /// 批量写入多条 K/V。
    ///
    /// 共用**当前事务**版本；全部成功返回 Ok，任一条冲突则回滚本批写入并
    /// `Err(failed_key)`。调用方成功后自行 `commit()`。
    pub fn batch_set(&self, items: &[(Vec<u8>, Vec<u8>)]) -> Result<(), Vec<u8>> {
        let mut written: Vec<Vec<u8>> = Vec::new();
        for (k, v) in items {
            let packed = pack_plain(v);
            if !self.set(k, packed) {
                self.rollback_keys(&written);
                return Err(k.clone());
            }
            written.push(k.clone());
        }
        Ok(())
    }

    /// `batch_set` 别名
    pub fn batch_put(&self, items: &[(Vec<u8>, Vec<u8>)]) -> Result<(), Vec<u8>> {
        self.batch_set(items)
    }

    /// 批量删除；全部成功返回 Ok，否则 Err(首个失败 key) 并回滚本批删除
    pub fn batch_delete(&self, keys: &[Vec<u8>]) -> Result<(), Vec<u8>> {
        let mut written: Vec<Vec<u8>> = Vec::new();
        for k in keys {
            if !self.delete(k) {
                self.rollback_keys(&written);
                return Err(k.clone());
            }
            written.push(k.clone());
        }
        Ok(())
    }

    /// 回滚本事务对给定 keys 的当前 version 写入（不 Abort 整个事务）
    fn rollback_keys(&self, keys: &[Vec<u8>]) {
        if keys.is_empty() {
            return;
        }
        let ver = self.version();
        let mut kv = self.kv.lock().unwrap();
        for k in keys {
            let _ = kv.remove(&encode_key(k, ver));
        }
        if let Ok(mut active) = self.active_txn.lock() {
            if let Some(writes) = active.get_mut(&ver) {
                writes.retain(|w| !keys.iter().any(|k| k == w));
            }
        }
    }

    // ── multi_get ──

    /// 一次查询多个 key；返回与 `keys` 等长，不存在/过期为 None
    pub fn multi_get(&self, keys: &[Vec<u8>]) -> Vec<Option<Vec<u8>>> {
        keys.iter().map(|k| self.get(k)).collect()
    }

    // ── 计数器 incr / decr ──

    /// 原子自增（本事务内：读 → 加 → 写）。
    ///
    /// - key 不存在 / 过期 / tombstone：从 0 起算
    /// - 已有 value：十进制 ASCII `i64`
    pub fn incr(&self, key: &[u8], delta: i64) -> Result<i64, IncrError> {
        self.incr_inner(key, delta, None)
    }

    /// 自减：`incr(key, -delta)`
    pub fn decr(&self, key: &[u8], delta: i64) -> Result<i64, IncrError> {
        self.incr(key, delta.checked_neg().ok_or(IncrError::Overflow)?)
    }

    /// 带 TTL 的自增
    pub fn incr_with_ttl(
        &self,
        key: &[u8],
        delta: i64,
        ttl: Duration,
    ) -> Result<i64, IncrError> {
        let expire = now_unix_secs().saturating_add(ttl.as_secs().max(1));
        self.incr_inner(key, delta, Some(expire))
    }

    fn incr_inner(
        &self,
        key: &[u8],
        delta: i64,
        expire_unix: Option<u64>,
    ) -> Result<i64, IncrError> {
        let now = now_unix_secs();
        let current = match self.latest_visible_raw(key) {
            Some((_ver, Some(raw))) => {
                let m = decode_value_meta(&raw);
                if m.is_expired_at(now) {
                    0i64
                } else {
                    parse_i64_user(m.user())?
                }
            }
            _ => 0i64,
        };
        let new_val = current.checked_add(delta).ok_or(IncrError::Overflow)?;
        let user = new_val.to_string().into_bytes();
        let packed = match expire_unix {
            Some(e) => pack_ttl(&user, e),
            None => pack_plain(&user),
        };
        if !self.set(key, packed) {
            return Err(IncrError::WriteConflict);
        }
        Ok(new_val)
    }

    // ── 范围遍历 ──

    /// 区间扫描 **[start, end)**：`start ≤ key < end`；`end=None` 表示无上界。
    ///
    /// 注意：仍会物化结果 `Vec`。大数据量请用 [`scan_foreach`] 避免 OOM。
    pub fn scan(&self, start: Option<&[u8]>, end: Option<&[u8]>) -> Vec<ExportRecord> {
        let mut out = Vec::new();
        self.scan_foreach(start, end, |rec| {
            out.push(rec);
            true
        });
        out
    }

    /// 流式扫描：对每条存活记录调用 `f`；`f` 返回 `false` 提前结束。
    /// 不收集全库到单一大 Vec（内部按有序 latest 迭代）。
    pub fn scan_foreach(
        &self,
        start: Option<&[u8]>,
        end: Option<&[u8]>,
        mut f: impl FnMut(ExportRecord) -> bool,
    ) {
        let now = now_unix_secs();
        for (key, _ver, raw) in self.collect_latest_raw(false) {
            if let Some(s) = start {
                if key.as_slice() < s {
                    continue;
                }
            }
            if let Some(e) = end {
                if key.as_slice() >= e {
                    break; // key 有序，可提前停
                }
            }
            let Some(bytes) = raw else { continue };
            let m = decode_value_meta(&bytes);
            if m.is_expired_at(now) {
                continue;
            }
            let rec = ExportRecord {
                key,
                value: Some(m.user().to_vec()),
            };
            if !f(rec) {
                break;
            }
        }
    }

    /// 前缀流式扫描
    pub fn prefix_scan_foreach(
        &self,
        prefix: &[u8],
        f: impl FnMut(ExportRecord) -> bool,
    ) {
        let end = prefix_upper_bound(prefix);
        self.scan_foreach(Some(prefix), end.as_deref(), f);
    }

    /// 前缀扫描：所有 `key.starts_with(prefix)`
    pub fn prefix_scan(&self, prefix: &[u8]) -> Vec<ExportRecord> {
        let end = prefix_upper_bound(prefix);
        self.scan(Some(prefix), end.as_deref())
    }

    /// 逆序扫描 [start, end)
    pub fn reverse_scan(&self, start: Option<&[u8]>, end: Option<&[u8]>) -> Vec<ExportRecord> {
        let mut v = self.scan(start, end);
        v.reverse();
        v
    }

    /// 定位 **≥ key** 的第一条存活记录
    pub fn seek(&self, key: &[u8]) -> Option<ExportRecord> {
        self.scan(Some(key), None).into_iter().next()
    }

    /// 定位 **≤ key** 的最后一条存活记录
    pub fn seek_prev(&self, key: &[u8]) -> Option<ExportRecord> {
        let mut upper = key.to_vec();
        upper.push(0);
        let mut v = self.scan(None, Some(&upper));
        v.retain(|r| r.key.as_slice() <= key);
        v.pop()
    }

    // ── 模糊 / 前缀搜索 + 分页 ──

    /// 按 key 模糊/前缀搜索并分页
    pub fn search_keys(&self, query: &SearchQuery) -> SearchPage {
        let page_size = query.page_size.max(1);
        let page = query.page;
        let matched = self.collect_for_search(
            query.include_deleted,
            Some(query.pattern.as_slice()),
            query.mode,
        );
        let total = matched.len();
        let total_pages = if total == 0 {
            0
        } else {
            (total + page_size - 1) / page_size
        };
        let start = page.saturating_mul(page_size);
        let items = if start >= total {
            Vec::new()
        } else {
            let end = (start + page_size).min(total);
            matched[start..end].to_vec()
        };
        SearchPage {
            items,
            total,
            page,
            page_size,
            total_pages,
        }
    }

    fn collect_for_search(
        &self,
        include_deleted: bool,
        pattern: Option<&[u8]>,
        mode: KeyMatchMode,
    ) -> Vec<ExportRecord> {
        self.collect_latest_raw(true)
            .into_iter()
            .filter_map(|(key, _ver, raw)| {
                if let Some(pat) = pattern {
                    if !key_matches(&key, pat, mode) {
                        return None;
                    }
                }
                match raw {
                    None => {
                        if include_deleted {
                            Some(ExportRecord {
                                key,
                                value: None,
                            })
                        } else {
                            None
                        }
                    }
                    Some(bytes) => match logical_to_user(bytes) {
                        Some(user) => Some(ExportRecord {
                            key,
                            value: Some(user),
                        }),
                        None => {
                            if include_deleted {
                                Some(ExportRecord {
                                    key,
                                    value: None,
                                })
                            } else {
                                None
                            }
                        }
                    },
                }
            })
            .collect()
    }

    // ── 元数据 ──

    /// key 是否存在（最新可见且未删除且未过期）
    pub fn exists(&self, key: &[u8]) -> bool {
        match self.latest_visible_raw(key) {
            Some((_ver, Some(raw))) => {
                let m = decode_value_meta(&raw);
                !m.is_expired_at(now_unix_secs())
            }
            _ => false,
        }
    }

    /// 元数据：长度、版本、过期时间、是否过期/删除
    pub fn get_meta(&self, key: &[u8]) -> Option<KeyMeta> {
        let (version, raw) = self.latest_visible_raw(key)?;
        match raw {
            None => Some(KeyMeta {
                key: key.to_vec(),
                value_len: None,
                version,
                expire_unix_secs: None,
                expired: false,
                deleted: true,
            }),
            Some(bytes) => {
                let m = decode_value_meta(&bytes);
                let expired = m.is_expired_at(now_unix_secs());
                Some(KeyMeta {
                    key: key.to_vec(),
                    value_len: if expired {
                        None
                    } else {
                        Some(m.user().len())
                    },
                    version,
                    expire_unix_secs: m.expire_unix_secs(),
                    expired,
                    deleted: false,
                })
            }
        }
    }

    /// 前缀下存活 key 数量
    pub fn key_count(&self, prefix: &[u8]) -> usize {
        self.prefix_scan(prefix).len()
    }

    // ── TTL ──

    /// 带 TTL 写入：`ttl` 为从现在起的存活时长
    pub fn set_with_ttl(&self, key: &[u8], value: Vec<u8>, ttl: Duration) -> bool {
        let expire = now_unix_secs().saturating_add(ttl.as_secs().max(1));
        let packed = pack_ttl(&value, expire);
        self.set(key, packed)
    }

    /// 续期：保留原用户 value，刷新过期时间
    pub fn refresh_ttl(&self, key: &[u8], ttl: Duration) -> bool {
        let Some((_ver, Some(raw))) = self.latest_visible_raw(key) else {
            return false;
        };
        let m = decode_value_meta(&raw);
        if m.is_expired_at(now_unix_secs()) {
            return false;
        }
        let expire = now_unix_secs().saturating_add(ttl.as_secs().max(1));
        let packed = pack_ttl(m.user(), expire);
        self.set(key, packed)
    }

    /// 剩余 TTL：
    /// - `None`：不存在 / tombstone / 已过期
    /// - `Some(None)`：存在且永久
    /// - `Some(Some(d))`：存在且剩余时长 d
    pub fn get_ttl(&self, key: &[u8]) -> Option<Option<Duration>> {
        let (_ver, raw) = self.latest_visible_raw(key)?;
        let raw = raw?;
        let m = decode_value_meta(&raw);
        let now = now_unix_secs();
        if m.is_expired_at(now) {
            return None;
        }
        match m.expire_unix_secs() {
            None => Some(None),
            Some(e) => {
                let left = e.saturating_sub(now);
                Some(Some(Duration::from_secs(left)))
            }
        }
    }

    /// 去掉 TTL，改为永久保存
    pub fn persist(&self, key: &[u8]) -> bool {
        let Some((_ver, Some(raw))) = self.latest_visible_raw(key) else {
            return false;
        };
        let m = decode_value_meta(&raw);
        if m.is_expired_at(now_unix_secs()) {
            return false;
        }
        let packed = pack_plain(m.user());
        self.set(key, packed)
    }

    /// 主动清理：把「最新可见且已过期」的 key 写成 tombstone。返回清理条数。
    pub fn purge_expired(&self) -> usize {
        let now = now_unix_secs();
        let expired_keys: Vec<Vec<u8>> = self
            .collect_latest_raw(false)
            .into_iter()
            .filter_map(|(key, _ver, raw)| {
                let raw = raw?;
                let m = decode_value_meta(&raw);
                if m.is_expired_at(now) {
                    Some(key)
                } else {
                    None
                }
            })
            .collect();
        let mut n = 0usize;
        for k in expired_keys {
            if self.delete(&k) {
                n += 1;
            }
        }
        n
    }
}

// ─── MVCC 便捷包装 ──────────────────────────────────────────────────────────

impl MVCC {
    /// 按 key 模糊/前缀匹配，分页返回最新可见记录
    pub fn search_keys(&self, query: &SearchQuery) -> SearchPage {
        let tx = self.begin_transaction();
        let page = tx.search_keys(query);
        tx.commit();
        page
    }
}

// ─── 测试 ───────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use std::path::PathBuf;
    use std::time::{SystemTime, UNIX_EPOCH};

    fn tmp(tag: &str) -> PathBuf {
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        std::env::temp_dir().join(format!(
            "bitcask_kv_ops_{tag}_{}_{}",
            std::process::id(),
            nanos
        ))
    }

    fn cleanup(path: &PathBuf) {
        let _ = std::fs::remove_dir_all(path);
    }

    #[test]
    fn test_value_meta_roundtrip() {
        let p = pack_plain(b"hi");
        match decode_value_meta(&p) {
            ValueMeta::Plain { user } => assert_eq!(user, b"hi"),
            _ => panic!(),
        }
        let t = pack_ttl(b"x", 1000);
        match decode_value_meta(&t) {
            ValueMeta::Ttl {
                expire_unix_secs,
                user,
            } => {
                assert_eq!(expire_unix_secs, 1000);
                assert_eq!(user, b"x");
            }
            _ => panic!(),
        }
        assert_eq!(
            decode_value_meta(b"raw"),
            ValueMeta::Plain {
                user: b"raw".to_vec()
            }
        );
    }

    #[test]
    fn test_batch_and_multi_get() {
        let path = tmp("batch");
        {
            let mvcc = MVCC::open(&path);
            let tx = mvcc.begin_transaction();
            let items = vec![
                (b"a".to_vec(), b"1".to_vec()),
                (b"b".to_vec(), b"2".to_vec()),
                (b"c".to_vec(), b"3".to_vec()),
            ];
            tx.batch_set(&items).unwrap();
            let got = tx.multi_get(&[b"a".to_vec(), b"b".to_vec(), b"z".to_vec()]);
            assert_eq!(got[0], Some(b"1".to_vec()));
            assert_eq!(got[1], Some(b"2".to_vec()));
            assert_eq!(got[2], None);
            tx.batch_delete(&[b"b".to_vec()]).unwrap();
            assert!(!tx.exists(b"b"));
            tx.commit();
        }
        cleanup(&path);
    }

    #[test]
    fn test_scan_prefix_seek() {
        let path = tmp("scan");
        {
            let mvcc = MVCC::open(&path);
            let tx = mvcc.begin_transaction();
            for (k, v) in [
                (&b"order:01"[..], &b"a"[..]),
                (&b"order:02"[..], &b"b"[..]),
                (&b"order:10"[..], &b"c"[..]),
                (&b"user:1"[..], &b"u"[..]),
            ] {
                assert!(tx.set(k, pack_plain(v)));
            }
            tx.commit();

            let tx = mvcc.begin_transaction();
            let s = tx.prefix_scan(b"order:");
            assert_eq!(s.len(), 3);
            assert_eq!(s[0].key, b"order:01");
            let one = tx.seek(b"order:05").unwrap();
            assert_eq!(one.key, b"order:10");
            let prev = tx.seek_prev(b"order:05").unwrap();
            assert_eq!(prev.key, b"order:02");
            assert_eq!(tx.key_count(b"order:"), 3);
            tx.commit();
        }
        cleanup(&path);
    }

    #[test]
    fn test_search_pagination() {
        let path = tmp("search");
        {
            let mvcc = MVCC::open(&path);
            let tx = mvcc.begin_transaction();
            for k in [&b"apple"[..], &b"apply"[..], &b"banana"[..], &b"user:1"[..]] {
                assert!(tx.set(k, pack_plain(b"v")));
            }
            tx.commit();

            let page = mvcc.search_keys(&SearchQuery::contains(b"pp", 0, 10));
            assert_eq!(page.total, 2);
            assert_eq!(page.items.len(), 2);

            let p0 = mvcc.search_keys(&SearchQuery::prefix(b"user:", 0, 1));
            assert_eq!(p0.total, 1);
            assert!(!p0.has_next());
        }
        cleanup(&path);
    }

    #[test]
    fn test_incr_and_ttl() {
        let path = tmp("incr");
        {
            let mvcc = MVCC::open(&path);
            let tx = mvcc.begin_transaction();
            assert_eq!(tx.incr(b"n", 1).unwrap(), 1);
            assert_eq!(tx.incr(b"n", 9).unwrap(), 10);
            assert_eq!(tx.decr(b"n", 3).unwrap(), 7);
            assert_eq!(tx.get(b"n"), Some(b"7".to_vec()));

            assert_eq!(
                tx.incr_with_ttl(b"hits", 1, Duration::from_secs(60))
                    .unwrap(),
                1
            );
            assert!(matches!(tx.get_ttl(b"hits"), Some(Some(_))));

            assert!(tx.set_with_ttl(b"tmp", b"x".to_vec(), Duration::from_secs(120)));
            assert!(tx.exists(b"tmp"));
            assert!(tx.persist(b"tmp"));
            assert_eq!(tx.get_ttl(b"tmp"), Some(None));

            // 过期清理
            let stale = pack_ttl(b"old", now_unix_secs().saturating_sub(5));
            assert!(tx.set(b"gone", stale));
            assert!(!tx.exists(b"gone"));
            let n = tx.purge_expired();
            assert!(n >= 1);
            tx.commit();
        }
        cleanup(&path);
    }

    #[test]
    fn test_meta_and_exists() {
        let path = tmp("meta");
        {
            let mvcc = MVCC::open(&path);
            let tx = mvcc.begin_transaction();
            assert!(tx.set(b"k", pack_plain(b"hello")));
            assert!(tx.exists(b"k"));
            let m = tx.get_meta(b"k").unwrap();
            assert_eq!(m.value_len, Some(5));
            assert!(!m.deleted);
            assert!(!m.expired);
            assert!(tx.delete(b"k"));
            assert!(!tx.exists(b"k"));
            tx.commit();
        }
        cleanup(&path);
    }

    #[test]
    fn test_prefix_upper_bound() {
        assert_eq!(
            prefix_upper_bound(b"ab"),
            Some(b"ac".to_vec())
        );
        assert_eq!(prefix_upper_bound(b"a\xff"), Some(b"b".to_vec()));
    }
}