rifts 0.3.10

//! # Low-Level Byte Store Engine Abstraction
//!
//! This module defines the [`StorageEngine`] trait -- the fundamental
//! key-value interface that every higher-level store (offset, log, dedupe,
//! snapshot) builds upon. Keys and values are opaque `Vec<u8>` byte slices;
//! all semantic interpretation (topic names, offsets, timestamps) is left to
//! the higher-level stores.
//!
//! ## Implementations
//!
//! - [`MemoryEngine`] -- a `DashMap`-backed engine with no persistence.
//!   Requires zero configuration and is always available. Ideal for
//!   development, testing, and single-process deployments.
//! - [`SledEngine`] -- a `sled::Tree`-backed engine that persists data to
//!   disk. Requires the `sled` Cargo feature. Each higher-level store
//!   opens its own tree from a shared `sled::Db` instance, giving each
//!   store an independent key space without prefix encoding at the engine
//!   level.

use std::sync::Arc;

use dashmap::DashMap;

/// Low-level byte store abstraction.
///
/// All keys and values are opaque `Vec<u8>`.  Higher-level stores
/// (offset, log, dedupe, snapshot) build on top of this trait and
/// perform their own key encoding via the
/// [`encode`](crate::storage::encode) module.
///
/// Implementations must be both `Send` and `Sync` so they can be shared
/// safely across threads (typically via an `Arc`).
pub trait StorageEngine: Send + Sync + 'static {
    /// Retrieve the value associated with `key`, if it exists.
    fn get(&self, key: &[u8]) -> Option<Vec<u8>>;

    /// Insert or overwrite the value for `key`.
    ///
    /// Returns the number of bytes written on success.
    fn put(&self, key: &[u8], value: &[u8]) -> Result<usize, String>;

    /// Remove the entry for `key` from the store.
    fn delete(&self, key: &[u8]) -> Result<(), String>;

    /// Scan all entries whose keys start with `prefix` and return them as
    /// `(key, value)` pairs.
    fn scan_prefix(&self, prefix: &[u8]) -> Vec<(Vec<u8>, Vec<u8>)>;

    /// Flush all pending writes to durable storage.
    ///
    /// Returns the number of bytes flushed on success.
    fn flush(&self) -> Result<usize, String>;
}

/// In-memory storage engine, backed by a concurrent `DashMap`.
#[derive(Debug, Default)]
pub struct MemoryEngine {
    inner: DashMap<Vec<u8>, Vec<u8>>,
}

impl MemoryEngine {
    pub fn new() -> Self {
        Self::default()
    }
}

impl StorageEngine for MemoryEngine {
    fn get(&self, key: &[u8]) -> Option<Vec<u8>> {
        self.inner.get(key).map(|v| v.clone())
    }

    fn put(&self, key: &[u8], value: &[u8]) -> Result<usize, String> {
        self.inner.insert(key.to_vec(), value.to_vec());
        Ok(value.len())
    }

    fn delete(&self, key: &[u8]) -> Result<(), String> {
        self.inner.remove(key);
        Ok(())
    }

    fn scan_prefix(&self, prefix: &[u8]) -> Vec<(Vec<u8>, Vec<u8>)> {
        self.inner
            .iter()
            .filter(|kv| kv.key().starts_with(prefix))
            .map(|kv| (kv.key().clone(), kv.value().clone()))
            .collect()
    }

    fn flush(&self) -> Result<usize, String> {
        Ok(0)
    }
}

pub type SharedEngine = Arc<dyn StorageEngine>;

#[cfg(feature = "sled")]
pub mod sled_engine {
    use super::StorageEngine;

    pub struct SledEngine {
        tree: sled::Tree,
    }

    impl SledEngine {
        pub fn new(tree: sled::Tree) -> Self {
            Self { tree }
        }

        pub fn flush(&self) -> Result<usize, sled::Error> {
            self.tree.flush()
        }

        pub fn cas(
            &self,
            key: Vec<u8>,
            expected: Option<Vec<u8>>,
            new_value: Vec<u8>,
        ) -> Result<Result<(), sled::CompareAndSwapError>, sled::Error> {
            self.tree.compare_and_swap(key, expected, Some(new_value))
        }
    }

    impl StorageEngine for SledEngine {
        fn get(&self, key: &[u8]) -> Option<Vec<u8>> {
            self.tree.get(key).ok().flatten().map(|v| v.to_vec())
        }

        fn put(&self, key: &[u8], value: &[u8]) -> Result<usize, String> {
            let len = value.len();
            self.tree
                .insert(key, value)
                .map(|_| len)
                .map_err(|e| format!("sled put: {e}"))
        }

        fn delete(&self, key: &[u8]) -> Result<(), String> {
            self.tree
                .remove(key)
                .map(|_| ())
                .map_err(|e| format!("sled delete: {e}"))
        }

        fn scan_prefix(&self, prefix: &[u8]) -> Vec<(Vec<u8>, Vec<u8>)> {
            self.tree
                .scan_prefix(prefix)
                .filter_map(|r| match r {
                    Ok(kv) => Some((kv.0.to_vec(), kv.1.to_vec())),
                    Err(e) => {
                        tracing::warn!(error = %e, "sled scan_prefix entry error");
                        None
                    }
                })
                .collect()
        }

        fn flush(&self) -> Result<usize, String> {
            self.tree.flush().map_err(|e| format!("sled flush: {e}"))
        }
    }
}

#[cfg(feature = "sled")]
pub use sled_engine::SledEngine;

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

    #[test]
    fn memory_engine_basic_ops() {
        let e = MemoryEngine::new();
        assert!(e.get(b"k").is_none());
        e.put(b"k", b"v").unwrap();
        assert_eq!(e.get(b"k"), Some(b"v".to_vec()));
        e.delete(b"k").unwrap();
        assert!(e.get(b"k").is_none());
    }

    #[test]
    fn memory_engine_scan_prefix() {
        let e = MemoryEngine::new();
        e.put(b"a:1", b"x").unwrap();
        e.put(b"a:2", b"y").unwrap();
        e.put(b"b:1", b"z").unwrap();
        let results = e.scan_prefix(b"a:");
        assert_eq!(results.len(), 2);
    }
}