ethrex-storage 30.0.0

use crate::api::{
    PrefixResult, StorageBackend, StorageLockedView, StorageReadView, StorageWriteBatch,
};
use crate::error::StoreError;
use rustc_hash::FxHashMap;
use std::path::Path;
use std::sync::{Arc, RwLock};

type Table = FxHashMap<Vec<u8>, Vec<u8>>;
type Database = FxHashMap<&'static str, Table>;

#[derive(Debug)]
pub struct InMemoryBackend {
    // RCU-style snapshot store: readers clone the inner Arc and then read lock-free.
    // Writes run under the outer write lock and use Arc::make_mut for copy-on-write.
    // If read snapshots are still alive, writes may clone the full Database.
    inner: Arc<RwLock<Arc<Database>>>,
}

impl InMemoryBackend {
    pub fn open() -> Result<Self, StoreError> {
        Ok(Self {
            inner: Arc::new(RwLock::new(Arc::new(Database::default()))),
        })
    }
}

impl StorageBackend for InMemoryBackend {
    fn clear_table(&self, table: &str) -> Result<(), StoreError> {
        let mut db = self
            .inner
            .write()
            .map_err(|_| StoreError::Custom("Failed to acquire write lock".to_string()))?;

        let db_mut = Arc::make_mut(&mut *db);
        if let Some(table_ref) = db_mut.get_mut(table) {
            table_ref.clear();
        }
        Ok(())
    }

    fn begin_read(&self) -> Result<Arc<dyn StorageReadView>, StoreError> {
        let snapshot = self
            .inner
            .read()
            .map_err(|_| StoreError::Custom("Failed to acquire read lock".to_string()))?
            .clone();
        Ok(Arc::new(InMemoryReadTx { snapshot }))
    }

    fn begin_write(&self) -> Result<Box<dyn StorageWriteBatch + 'static>, StoreError> {
        Ok(Box::new(InMemoryWriteTx {
            backend: self.inner.clone(),
        }))
    }

    fn begin_locked(
        &self,
        table_name: &'static str,
    ) -> Result<Box<dyn StorageLockedView>, StoreError> {
        let snapshot = self
            .inner
            .read()
            .map_err(|_| StoreError::Custom("Failed to acquire read lock".to_string()))?
            .clone();
        Ok(Box::new(InMemoryLocked {
            snapshot,
            table_name,
        }))
    }

    fn create_checkpoint(&self, _path: &Path) -> Result<(), StoreError> {
        // Checkpoints are not supported for the InMemory DB
        // Silently ignoring the request to create a checkpoint is harmless
        Ok(())
    }
}

pub struct InMemoryLocked {
    snapshot: Arc<Database>,
    table_name: &'static str,
}

impl StorageLockedView for InMemoryLocked {
    fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, StoreError> {
        Ok(self
            .snapshot
            .get(&self.table_name)
            .and_then(|table_ref| table_ref.get(key))
            .cloned())
    }
}

pub struct InMemoryReadTx {
    snapshot: Arc<Database>,
}

impl StorageReadView for InMemoryReadTx {
    fn get(&self, table: &str, key: &[u8]) -> Result<Option<Vec<u8>>, StoreError> {
        Ok(self
            .snapshot
            .get(table)
            .and_then(|table_ref| table_ref.get(key))
            .cloned())
    }

    fn prefix_iterator(
        &self,
        table: &str,
        prefix: &[u8],
    ) -> Result<Box<dyn Iterator<Item = PrefixResult> + '_>, StoreError> {
        let table_data = self.snapshot.get(table).cloned().unwrap_or_default();

        // Seek and walk forward, as RocksDB does: callers seek past the start of a prefix
        // (`block_hash || start_index` for receipts) or across several prefixes (witness
        // pruning) and stop on their own.
        let mut entries: Vec<(Vec<u8>, Vec<u8>)> = table_data
            .into_iter()
            .filter(|(key, _)| key.as_slice() >= prefix)
            .collect();
        entries.sort_unstable_by(|(left, _), (right, _)| left.cmp(right));

        let results: Vec<PrefixResult> = entries
            .into_iter()
            .map(|(k, v)| Ok((k.into_boxed_slice(), v.into_boxed_slice())))
            .collect();

        Ok(Box::new(results.into_iter()))
    }

    fn first_key(&self, table: &'static str) -> Result<Option<Vec<u8>>, StoreError> {
        let Some(table_data) = self.snapshot.get(table) else {
            return Ok(None);
        };
        Ok(table_data.keys().min().cloned())
    }

    fn last_key(&self, table: &'static str) -> Result<Option<Vec<u8>>, StoreError> {
        let Some(table_data) = self.snapshot.get(table) else {
            return Ok(None);
        };
        Ok(table_data.keys().max().cloned())
    }
}

pub struct InMemoryWriteTx {
    backend: Arc<RwLock<Arc<Database>>>,
}

impl StorageWriteBatch for InMemoryWriteTx {
    fn put_batch(
        &mut self,
        table: &'static str,
        batch: Vec<(Vec<u8>, Vec<u8>)>,
    ) -> Result<(), StoreError> {
        let mut db = self
            .backend
            .write()
            .map_err(|_| StoreError::Custom("Failed to acquire write lock".to_string()))?;

        // Copy-on-write update of the current snapshot.
        let db_mut = Arc::make_mut(&mut *db);
        let table_ref = db_mut.entry(table).or_default();

        for (key, value) in batch {
            table_ref.insert(key, value);
        }

        Ok(())
    }

    fn delete(&mut self, table: &str, key: &[u8]) -> Result<(), StoreError> {
        let mut db = self
            .backend
            .write()
            .map_err(|_| StoreError::Custom("Failed to acquire write lock".to_string()))?;

        let db_mut = Arc::make_mut(&mut *db);
        if let Some(table_ref) = db_mut.get_mut(table) {
            table_ref.remove(key);
        }
        Ok(())
    }

    fn delete_range(
        &mut self,
        table: &'static str,
        start: &[u8],
        end: &[u8],
    ) -> Result<(), StoreError> {
        let mut db = self
            .backend
            .write()
            .map_err(|_| StoreError::Custom("Failed to acquire write lock".to_string()))?;

        let db_mut = Arc::make_mut(&mut *db);
        if let Some(table_ref) = db_mut.get_mut(table) {
            table_ref.retain(|k, _| !(k.as_slice() >= start && k.as_slice() < end));
        }
        Ok(())
    }

    fn merge(&mut self, table: &'static str, key: &[u8], operand: &[u8]) -> Result<(), StoreError> {
        // InMemory has no native merge operator, so apply the merge inline.
        // Only TRANSACTION_LOCATIONS uses merge today; dispatch by table.
        if table != crate::api::tables::TRANSACTION_LOCATIONS {
            return Err(StoreError::Custom(format!(
                "merge not supported for table {table}"
            )));
        }
        let mut db = self
            .backend
            .write()
            .map_err(|_| StoreError::Custom("Failed to acquire write lock".to_string()))?;
        let db_mut = Arc::make_mut(&mut *db);
        let table_ref = db_mut.entry(table).or_default();
        let existing = table_ref.get(key).map(|v| v.as_slice());
        let merged = crate::store::tx_locations_merge(existing, std::iter::once(operand))
            .ok_or_else(|| StoreError::Custom("tx_locations_merge returned None".to_string()))?;
        table_ref.insert(key.to_vec(), merged);
        Ok(())
    }

    fn commit(&mut self) -> Result<(), StoreError> {
        // NOTE: every `put`, `delete`, and `delete_range` above mutates the live
        // `Arc<Database>` immediately under the write lock, so `commit` is a no-op
        // and multi-op sequences (e.g. the journal entry + trie writes in
        // `commit_to_disk`, or the `delete_range` + finalized-number update in
        // `forkchoice_update_inner`) are not atomic. That's acceptable here: this
        // backend is RAM-backed (dev/test only), and atomicity only guards crash
        // recovery — a process death loses all in-memory state anyway, so a
        // half-applied batch is never observable.
        Ok(())
    }
}

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

    /// Like the RocksDB backend, which configures no prefix extractor, the iterator starts
    /// at the first key not less than the one given and runs through the rest of the
    /// table in key order.
    #[test]
    fn prefix_iterator_seeks_and_walks_forward() {
        let backend = InMemoryBackend::open().unwrap();
        let mut write = backend.begin_write().unwrap();
        let keys = ["b1", "a2", "c", "a1"].map(|key| (key.as_bytes().to_vec(), vec![]));
        write.put_batch("table", keys.to_vec()).unwrap();
        write.commit().unwrap();

        let read = backend.begin_read().unwrap();
        let keys_from = |start: &[u8]| {
            read.prefix_iterator("table", start)
                .unwrap()
                .map(|item| String::from_utf8(item.unwrap().0.to_vec()).unwrap())
                .collect::<Vec<_>>()
        };
        assert_eq!(keys_from(b""), ["a1", "a2", "b1", "c"]);
        assert_eq!(keys_from(b"a2"), ["a2", "b1", "c"]);
        assert_eq!(keys_from(b"a3"), ["b1", "c"]);
        assert_eq!(keys_from(b"d"), Vec::<String>::new());
    }
}