logfs 0.1.1

mod error;
pub use self::error::LogFsError;

mod journal;
mod state;
pub use journal::{Journal2, JournalStore};
use journal::{
    SequenceId, Superblock,
    v2::read::{KeyChunkIter, StdKeyReader},
};

mod crypto;
pub use crypto::CryptoConfig;

use std::{
    path::PathBuf,
    sync::{Arc, Condvar, Mutex, RwLock},
};

type Path = String;

pub struct ConfigBuilder {
    config: LogConfig,
}

const DEFAULT_CHUNK_SIZE: u32 = 4_000_000;

impl ConfigBuilder {
    pub fn new(path: impl Into<PathBuf>) -> Self {
        Self {
            config: LogConfig {
                path: path.into(),
                offset: None,
                allow_create: false,
                raw_mode: false,
                crypto: None,
                default_chunk_size: DEFAULT_CHUNK_SIZE,
                // TODO: determine good defaults for these values!
                partial_index_write_interval: 100,
                full_index_write_interval: 1000,
                readonly: false,
            },
        }
    }

    pub fn raw_mode(mut self) -> Self {
        self.config.raw_mode = true;
        self
    }

    pub fn offset(mut self, offset: Option<u64>) -> Self {
        self.config.offset = offset;
        self
    }

    pub fn default_chunk_size(mut self, size: u32) -> Self {
        self.config.default_chunk_size = size;
        self
    }

    pub fn allow_create(mut self) -> Self {
        self.config.allow_create = true;
        self
    }

    pub fn crypto(mut self, crypto: CryptoConfig) -> Self {
        self.config.crypto = Some(crypto);
        self
    }

    pub fn full_index_write_interval(mut self, interval: u64) -> Self {
        self.config.full_index_write_interval = interval;
        self
    }

    pub fn readonly(mut self, readonly: bool) -> Self {
        self.config.readonly = readonly;
        self
    }

    pub fn build(self) -> LogConfig {
        self.config
    }

    pub fn open(self) -> Result<LogFs, LogFsError> {
        LogFs::open(self.config)
    }
}

#[derive(Clone, Debug)]
pub struct LogConfig {
    pub path: PathBuf,
    pub raw_mode: bool,
    /// Optional file offset where the DB should start.
    pub offset: Option<u64>,
    pub allow_create: bool,
    pub crypto: Option<crypto::CryptoConfig>,
    /// Data is chunked into separate slices, which allows incrementally reading
    /// large keys.
    /// This setting specifies the size of chunks in bytes.
    ///
    /// Note that keys can also be created with a custom chunk size.
    pub default_chunk_size: u32,

    /// Determines after how many journal entries a new partial index snapshot
    /// is written.
    pub partial_index_write_interval: u64,
    /// Determines after how many journal entries a new full index snapshot is
    /// written.
    pub full_index_write_interval: u64,
    pub readonly: bool,
}

pub struct RepairConfig {
    pub dry_run: bool,
    pub start_sequence: Option<u64>,
    /// The path to which a recovered log should be written.
    pub recovery_path: Option<PathBuf>,
    pub skip_bytes: Option<u64>,
}

pub struct LogFs<J = journal::Journal2> {
    inner: Arc<Inner<J>>,
    path: PathBuf,
}

impl Clone for LogFs {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
            path: self.path.clone(),
        }
    }
}

struct Inner<J> {
    config: LogConfig,
    state: Arc<RwLock<state::State>>,
    locks: Arc<Locks>,
    journal: J,
}

struct Locks {
    key_lock: Mutex<bool>,
    key_lock_condvar: Condvar,
}

#[derive(Clone, Debug)]
pub struct KeyMeta {
    pub size: u64,
    pub chunk_size: Option<u32>,
}

pub struct KeyLock(Arc<Locks>);

impl Drop for KeyLock {
    fn drop(&mut self) {
        let mut flag = self.0.key_lock.lock().unwrap();
        *flag = false;
        self.0.key_lock_condvar.notify_all();
    }
}

type DataOffset = u64;

impl<J: JournalStore> LogFs<J> {
    // TODO: add open() without key and open_encrypted() with key.
    pub fn open(mut config: LogConfig) -> Result<Self, LogFsError> {
        tracing::debug!(?config, "opening log");
        let crypto = config
            .crypto
            .take()
            .map(|c| Arc::new(crypto::Crypto::new(c)));
        let state = Arc::new(RwLock::new(state::State::new()));
        let path = config.path.clone();
        let journal = J::open(path.clone(), state.clone(), crypto, &config)?;

        tracing::info!(?config, "log opened");

        Ok(Self {
            path,
            inner: Arc::new(Inner {
                state,
                config,
                journal,
                locks: Arc::new(Locks {
                    key_lock: Mutex::new(false),
                    key_lock_condvar: Condvar::new(),
                }),
            }),
        })
    }

    pub fn superblock(&self) -> Result<Superblock, LogFsError> {
        self.inner.journal.supberlock()
    }

    pub fn repair(mut config: LogConfig, repair_config: RepairConfig) -> Result<(), LogFsError> {
        let crypto = config
            .crypto
            .take()
            .map(|c| Arc::new(crypto::Crypto::new(c)));
        J::repair(
            &config,
            crypto.clone(),
            journal::RepairConfig {
                dry_run: repair_config.dry_run,
                start_sequence: repair_config.start_sequence.map(SequenceId::from_u64),
                recovery_path: repair_config.recovery_path,
                skip_bytes: repair_config.skip_bytes,
            },
        )?;

        Ok(())
    }

    /// Get the file system path.
    pub fn path(&self) -> std::path::PathBuf {
        self.path.clone()
    }

    /// Returns the approximate amount of bytes that could be saved when
    /// re-writing the log.
    ///
    /// Returns [`None`] if no estimate is available.
    /// This is the case if the log was restored from an index without a full
    /// scan.
    // TODO: if estimate is not available, do a full scan to determine estimate.
    pub fn redundant_data_estimate(&self) -> Option<u128> {
        self.inner
            .state
            .read()
            .unwrap()
            .redundant_data_bytes_estimate()
    }

    pub fn get_meta(&self, path: impl AsRef<str>) -> Result<Option<KeyMeta>, LogFsError> {
        match self.inner.state.read().unwrap().get_key(path.as_ref()) {
            Some(pointer) => Ok(Some(KeyMeta {
                size: pointer.size,
                chunk_size: pointer.chunk_size,
            })),
            None => Ok(None),
        }
    }

    /// Get a key.
    pub fn get(&self, path: impl AsRef<str>) -> Result<Option<Vec<u8>>, LogFsError> {
        let pointer = match self
            .inner
            .state
            .read()
            .unwrap()
            .get_key(path.as_ref())
            .cloned()
        {
            Some(pointer) => pointer,
            None => {
                return Ok(None);
            }
        };
        let data = self.inner.journal.read_data(&pointer)?;
        Ok(Some(data))
    }

    pub fn get_reader(&self, path: impl AsRef<str>) -> Result<StdKeyReader, LogFsError> {
        let path = path.as_ref();

        let pointer = match self.inner.state.read().unwrap().get_key(path).cloned() {
            Some(pointer) => pointer,
            None => return Err(LogFsError::NotFound { path: path.into() }),
        };
        let reader = self.inner.journal.reader(&pointer)?;
        Ok(reader)
    }

    pub fn get_chunks(&self, path: impl AsRef<str>) -> Result<KeyChunkIter, LogFsError> {
        let path = path.as_ref();

        let pointer = match self.inner.state.read().unwrap().get_key(path).cloned() {
            Some(pointer) => pointer,
            None => return Err(LogFsError::NotFound { path: path.into() }),
        };
        let reader = self.inner.journal.read_chunks(&pointer)?;
        Ok(reader)
    }

    /// Get all paths in the given range.
    pub fn paths_range<R>(&self, range: R) -> Result<Vec<Path>, LogFsError>
    where
        R: std::ops::RangeBounds<String>,
    {
        Ok(self.inner.state.read().unwrap().paths_range(range))
    }

    /// Get all paths with a given prefix.
    pub fn paths_offset(&self, offset: usize, max: usize) -> Result<Vec<Path>, LogFsError> {
        Ok(self.inner.state.read().unwrap().paths_offset(offset, max))
    }

    /// Get all paths with a given prefix.
    pub fn paths_prefix(&self, prefix: &str) -> Result<Vec<Path>, LogFsError> {
        Ok(self.inner.state.read().unwrap().paths_prefix(prefix))
    }

    fn acquire_key_lock(&self) -> KeyLock {
        let mut flag = self.inner.locks.key_lock.lock().unwrap();
        while *flag {
            flag = self.inner.locks.key_lock_condvar.wait(flag).unwrap();
        }
        *flag = true;
        KeyLock(self.inner.locks.clone())
    }

    fn write_index_if_required(&self, state: &mut state::State) -> Result<(), LogFsError> {
        // TODO: support partial index writes!

        if state.write_counter > self.inner.config.full_index_write_interval {
            self.inner.journal.write_index(&state.tree, true)?;
            state.write_counter = 0;
        }

        Ok(())
    }

    /// Insert a key.
    pub fn insert(&self, path: impl Into<String>, data: Vec<u8>) -> Result<(), LogFsError> {
        if self.inner.config.readonly {
            return Err(LogFsError::ReadOnly);
        }
        let path = path.into();
        let size = data.len();
        tracing::trace!(?path, size, "inserting key");
        let _lock = self.acquire_key_lock();

        let pointer = self.inner.journal.write_insert(path.clone(), data)?;

        let mut state = self.inner.state.write().unwrap();
        state.add_key(path.clone(), pointer);

        self.write_index_if_required(&mut state)?;

        tracing::trace!(?path, size, "key inserted");

        Ok(())
    }

    pub fn insert_writer(
        &self,
        path: impl Into<String>,
    ) -> Result<journal::v2::write::KeyWriter, LogFsError> {
        let lock = self.acquire_key_lock();
        self.inner
            .journal
            .insert_writer(path.into(), self.inner.state.clone(), lock)
    }

    /// Rename a key.
    pub fn rename(
        &self,
        old_key: impl Into<String>,
        new_key: impl Into<String>,
    ) -> Result<(), LogFsError> {
        if self.inner.config.readonly {
            return Err(LogFsError::ReadOnly);
        }
        let old_key = old_key.into();
        let new_key = new_key.into();

        let _lock = self.acquire_key_lock();

        // Ensure key exists.
        if self.inner.state.read().unwrap().get_key(&old_key).is_none() {
            return Err(LogFsError::NotFound {
                path: old_key.to_string(),
            });
        }

        self.inner
            .journal
            .write_rename(old_key.clone(), new_key.clone())?;

        let mut state = self.inner.state.write().unwrap();
        // NOTE: unwrap can't fail, since key existence was checked above.
        state.rename_key(&old_key, new_key).unwrap();
        self.write_index_if_required(&mut state)?;

        Ok(())
    }

    /// Remove a key.
    pub fn remove(&self, path: impl AsRef<str>) -> Result<(), LogFsError> {
        if self.inner.config.readonly {
            return Err(LogFsError::ReadOnly);
        }
        let path = path.as_ref();

        let _lock = self.acquire_key_lock();

        let mut state = self.inner.state.write().unwrap();
        if state.remove_key(path).is_some() {
            self.inner.journal.write_remove(vec![path.to_string()])?;

            self.write_index_if_required(&mut state)?;
        }

        Ok(())
    }

    /// Remove a whole key prefix.
    pub fn remove_prefix(&self, prefix: impl AsRef<str>) -> Result<(), LogFsError> {
        if self.inner.config.readonly {
            return Err(LogFsError::ReadOnly);
        }
        let prefix = prefix.as_ref();
        let _lock = self.acquire_key_lock();
        let paths = {
            let state = self.inner.state.read().unwrap();
            state.paths_prefix(prefix)
        };

        tracing::trace!(%prefix, key_count=%paths.len(), "deleting keys with prefix");

        if paths.is_empty() {
            return Ok(());
        }

        self.inner.journal.write_remove(paths.clone())?;

        let mut state = self.inner.state.write().unwrap();
        for path in &paths {
            state.remove_key(path);
        }

        self.write_index_if_required(&mut state)?;

        Ok(())
    }

    pub fn batch(&self, batch: Batch) -> Result<(), LogFsError> {
        if self.inner.config.readonly {
            return Err(LogFsError::ReadOnly);
        }

        let state = self.inner.state.write().unwrap();

        // Validate.

        for deleted_key in &batch.deleted_keys {
            if state.get_key(deleted_key).is_none() {
                return Err(LogFsError::NotFound {
                    path: deleted_key.clone(),
                });
            }
        }

        for rename in &batch.renames {
            if state.get_key(&rename.old_key).is_none() {
                return Err(LogFsError::NotFound {
                    path: rename.old_key.clone(),
                });
            }
        }

        self.inner.journal.write_batch(batch.clone())?;

        let mut state = state;

        for key in &batch.deleted_keys {
            state.remove_key(key);
        }
        for rename in batch.renames {
            // Unwrap is fine since key existence was validated above.
            state.rename_key(&rename.old_key, rename.new_key).unwrap();
        }

        self.write_index_if_required(&mut state)?;

        Ok(())
    }

    pub fn size_data(&self) -> Result<u64, LogFsError> {
        let size = self
            .inner
            .state
            .read()
            .unwrap()
            .tree
            .values()
            .map(|v| v.size)
            .sum();
        Ok(size)
    }

    pub fn size_log(&self) -> Result<u64, LogFsError> {
        self.inner.journal.size_log()
    }
}

#[derive(Clone, Debug)]
pub struct Rename {
    pub old_key: String,
    pub new_key: String,
}

#[derive(Clone, Debug, Default)]
pub struct Batch {
    pub renames: Vec<Rename>,
    pub deleted_keys: Vec<String>,
}

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

    pub fn and_rename(mut self, old_key: impl Into<String>, new_key: impl Into<String>) -> Self {
        self.renames.push(Rename {
            old_key: old_key.into(),
            new_key: new_key.into(),
        });
        self
    }

    pub fn and_remove(mut self, keys: Vec<String>) -> Self {
        self.deleted_keys.extend(keys);
        self
    }
}

impl LogFs<Journal2> {
    pub fn migrate(self) -> Result<(), LogFsError> {
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use std::{
        io::{Read, Write},
        num::NonZeroU32,
    };

    use crate::journal::Journal2;

    use super::*;

    fn test_config(name: &str) -> LogConfig {
        LogConfig {
            path: temp_test_dir(name),
            offset: None,
            raw_mode: false,
            allow_create: true,
            readonly: false,
            crypto: Some(CryptoConfig {
                key: "logfs".to_string().into(),
                salt: b"salt".to_vec().into(),
                iterations: NonZeroU32::new(1).unwrap(),
            }),
            // Set a very low chunk size to test chunking.
            default_chunk_size: 3,
            partial_index_write_interval: 5,
            full_index_write_interval: 10,
        }
    }

    pub fn temp_test_dir(name: &str) -> PathBuf {
        let tmp_dir = std::env::temp_dir().join("logfs_tests");
        if !tmp_dir.is_dir() {
            std::fs::create_dir_all(&tmp_dir).unwrap();
        }
        let path = tmp_dir.join(name);
        if path.exists() {
            std::fs::remove_file(&path).unwrap();
        }
        path
    }

    fn test_db<J: JournalStore>(name: &str) -> LogFs<J> {
        LogFs::<J>::open(test_config(name)).unwrap()
    }

    #[test]
    fn test_full_flow() {
        let config = test_config("full_flow");
        let log = LogFs::<Journal2>::open(config.clone()).unwrap();

        let key1 = "a/b/c";
        let content1 = b"hello there".to_vec();

        let key2 = "x";
        let content2 = b"xyz".to_vec();

        let key3_a = "rename/first";
        let key3_b = "rename/second";
        let content3 = b"key3!".to_vec();

        // Just insert some keys first.

        log.insert(key1, content1.clone()).unwrap();
        assert_eq!(log.get(key1).unwrap(), Some(content1.clone()));

        log.insert(key2, content2.clone()).unwrap();
        assert_eq!(log.get(key2).unwrap(), Some(content2.clone()));

        // Now drop the DB and re-open to verify that re-loading works.
        std::mem::drop(log);

        let log2 = LogFs::<Journal2>::open(config.clone()).unwrap();

        assert_eq!(log2.get(key1).unwrap(), Some(content1.clone()));
        assert_eq!(log2.get(key2).unwrap(), Some(content2.clone()));

        log2.remove(key1).unwrap();

        log2.insert(key3_a, content3.clone()).unwrap();
        assert_eq!(&log2.get(key3_a).unwrap().unwrap(), &content3);
        log2.rename(key3_a, key3_b).unwrap();
        assert_eq!(log2.get(key3_a).unwrap(), None);
        assert_eq!(&log2.get(key3_b).unwrap().unwrap(), &content3);

        std::mem::drop(log2);

        let log3 = LogFs::<Journal2>::open(config.clone()).unwrap();
        assert_eq!(log3.get(key1).unwrap(), None);
        assert_eq!(log3.get(key2).unwrap(), Some(content2.clone()));

        assert_eq!(log3.get(key3_a).unwrap(), None);
        assert_eq!(&log3.get(key3_b).unwrap().unwrap(), &content3);
    }

    #[test]
    fn test_full_flow_with_offset() {
        let header_content: &[u8] = b"this is a long header in the filer that must not be touched !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! Ok?";

        let mut config = test_config("full_flow_with_offset");
        config.offset = Some(header_content.len() as u64);
        config.allow_create = true;

        // create the file up to the offset.

        {
            let mut f = std::fs::File::create(&config.path).unwrap();
            f.write_all(header_content).unwrap();
        }

        let log = LogFs::<Journal2>::open(config.clone()).unwrap();

        let key1 = "a/b/c";
        let content1 = b"hello there".to_vec();

        let key2 = "x";
        let content2 = b"xyz".to_vec();

        let key3_a = "rename/first";
        let key3_b = "rename/second";
        let content3 = b"key3!".to_vec();

        // Just insert some keys first.

        log.insert(key1, content1.clone()).unwrap();
        assert_eq!(log.get(key1).unwrap(), Some(content1.clone()));

        log.insert(key2, content2.clone()).unwrap();
        assert_eq!(log.get(key2).unwrap(), Some(content2.clone()));

        // Now drop the DB and re-open to verify that re-loading works.
        std::mem::drop(log);

        let log2 = LogFs::<Journal2>::open(config.clone()).unwrap();

        assert_eq!(log2.get(key1).unwrap(), Some(content1.clone()));
        assert_eq!(log2.get(key2).unwrap(), Some(content2.clone()));

        log2.remove(key1).unwrap();

        log2.insert(key3_a, content3.clone()).unwrap();
        assert_eq!(&log2.get(key3_a).unwrap().unwrap(), &content3);
        log2.rename(key3_a, key3_b).unwrap();
        assert_eq!(log2.get(key3_a).unwrap(), None);
        assert_eq!(&log2.get(key3_b).unwrap().unwrap(), &content3);

        std::mem::drop(log2);

        let log3 = LogFs::<Journal2>::open(config.clone()).unwrap();
        assert_eq!(log3.get(key1).unwrap(), None);
        assert_eq!(log3.get(key2).unwrap(), Some(content2.clone()));

        assert_eq!(log3.get(key3_a).unwrap(), None);
        assert_eq!(&log3.get(key3_b).unwrap().unwrap(), &content3);

        std::mem::drop(log3);

        // Now verify that the header content is still there.

        let mut f = std::fs::File::open(&config.path).unwrap();
        let mut buf = vec![0u8; header_content.len()];
        f.read_exact(&mut buf).unwrap();
        assert_eq!(header_content, &buf)
    }

    #[test]
    fn test_iterate_range() -> Result<(), LogFsError> {
        let db = test_db::<Journal2>("iterate_range");
        db.insert("a", vec![0])?;
        db.insert("b", vec![0])?;
        db.insert("c/1", vec![1])?;
        db.insert("c/2", vec![3])?;
        db.insert("d", vec![0])?;
        db.insert("e", vec![0])?;

        // Exclusive range.
        let mut keys = db.paths_range("b".to_string().."d".to_string())?;
        keys.sort();
        assert_eq!(
            keys,
            vec!["b".to_string(), "c/1".to_string(), "c/2".to_string(),]
        );

        // Inclusive range.
        let mut keys = db.paths_range("b".to_string()..="d".to_string())?;
        keys.sort();
        assert_eq!(
            keys,
            vec![
                "b".to_string(),
                "c/1".to_string(),
                "c/2".to_string(),
                "d".to_string(),
            ]
        );

        // All.
        let mut keys = db.paths_range(..)?;
        keys.sort();
        assert_eq!(
            keys,
            vec![
                "a".to_string(),
                "b".to_string(),
                "c/1".to_string(),
                "c/2".to_string(),
                "d".to_string(),
                "e".to_string(),
            ]
        );

        Ok(())
    }

    #[test]
    fn test_iterate_prefix() -> Result<(), LogFsError> {
        let db = test_db::<Journal2>("iterate_prefix");
        db.insert("a", vec![0])?;
        db.insert("b", vec![0])?;
        db.insert("c", vec![1])?;
        db.insert("c/1", vec![1])?;
        db.insert("c/2", vec![3])?;
        db.insert("d", vec![0])?;
        db.insert("e", vec![0])?;

        let mut keys = db.paths_prefix("c")?;
        keys.sort();
        assert_eq!(
            keys,
            vec!["c".to_string(), "c/1".to_string(), "c/2".to_string(),]
        );

        let keys = db.paths_prefix("d")?;
        assert_eq!(keys, vec!["d".to_string(),]);

        // All.
        let mut keys = db.paths_prefix("")?;
        keys.sort();
        assert_eq!(
            keys,
            vec![
                "a".to_string(),
                "b".to_string(),
                "c".to_string(),
                "c/1".to_string(),
                "c/2".to_string(),
                "d".to_string(),
                "e".to_string(),
            ]
        );

        Ok(())
    }

    #[test]
    fn test_remove_multiple_paths() -> Result<(), LogFsError> {
        let db = test_db::<Journal2>("remove_multiple_paths");
        db.insert("other", vec![0])?;
        db.insert("prefix", vec![0])?;
        db.insert("prefix/1", vec![1])?;
        db.insert("prefix/2", vec![2])?;
        db.insert("prefix/3", vec![3])?;
        db.insert("blub", vec![0])?;

        db.remove_prefix("prefix")?;

        let mut keys = db.paths_range(..)?;
        keys.sort();
        assert_eq!(keys, vec!["blub".to_string(), "other".to_string()]);

        Ok(())
    }

    #[test]
    fn test_writer() -> Result<(), LogFsError> {
        let config = test_config("writer");

        let db = LogFs::<Journal2>::open(config.clone())?;

        let path1 = "regular";
        let data1 = b"regular111111111".to_vec();
        db.insert(path1, data1.clone())?;

        let path2 = "writer/1";
        let mut writer = db.insert_writer(path2)?;
        let data2 = b"123456789123456789123456789123456789";
        writer.write_all(data2)?;
        writer.finish()?;

        let path3 = "writer/2";
        let mut writer = db.insert_writer(path3)?;
        let data3 = b"123456789123456789123456789123456789";
        writer.write_all(data3)?;
        writer.finish()?;

        assert_eq!(db.get(path1)?.unwrap(), data1);
        assert_eq!(db.get(path2)?.unwrap(), data2);
        assert_eq!(db.get(path3)?.unwrap(), data3);

        std::mem::drop(db);
        let db = LogFs::<Journal2>::open(config.clone())?;

        assert_eq!(db.get(path1)?.unwrap(), data1);
        assert_eq!(db.get(path2)?.unwrap(), data2);
        assert_eq!(db.get(path3)?.unwrap(), data3);

        Ok(())
    }

    #[test]
    fn test_reader() -> Result<(), LogFsError> {
        let config = test_config("reader");
        let path = "key";
        let data = "aaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbb";

        let db = LogFs::<Journal2>::open(config.clone())?;
        db.insert(path, data.into())?;
        assert_eq!(db.get(path)?.unwrap(), data.as_bytes());

        let mut reader = db.get_reader(path)?;
        let mut buf = String::new();
        reader.read_to_string(&mut buf)?;
        assert_eq!(&buf, data);

        std::mem::drop(db);

        let db = LogFs::<Journal2>::open(config.clone())?;
        assert_eq!(db.get(path)?.unwrap(), data.as_bytes());

        let mut reader = db.get_reader(path)?;
        let mut buf = String::new();
        reader.read_to_string(&mut buf)?;
        assert_eq!(&buf, data);

        let mut all = Vec::new();
        for res in db.get_chunks(path)? {
            all.extend(res?);
        }
        assert_eq!(&all, data.as_bytes());

        Ok(())
    }

    #[test]
    fn test_chunk_iter() {
        let db = test_db::<Journal2>("chunk_iter");

        let data = "000111222333444555666777888999";
        let path = "a";
        db.insert(path, data.as_bytes().to_vec()).unwrap();

        assert_eq!(&db.get(path).unwrap().unwrap(), data.as_bytes());

        let mut chunks = db.get_chunks(path).unwrap();
        assert_eq!(&chunks.next().unwrap().unwrap(), b"000");

        chunks.skip_bytes(6).unwrap();
        assert_eq!(&chunks.next().unwrap().unwrap(), b"333");
        assert_eq!(&chunks.next().unwrap().unwrap(), b"444");

        // Partial chunk seek.
        chunks.skip_bytes(2).unwrap();
        assert_eq!(&chunks.next().unwrap().unwrap(), b"5");
        assert_eq!(&chunks.next().unwrap().unwrap(), b"666");

        chunks.skip_bytes(1).unwrap();
        assert_eq!(&chunks.next().unwrap().unwrap(), b"77");

        // assert_eq!(&chunks.next().unwrap().unwrap(), b"888");
        // assert_eq!(&chunks.next().unwrap().unwrap(), b"999");

        chunks.skip_bytes(5).unwrap();
        assert_eq!(&chunks.next().unwrap().unwrap(), b"9");

        assert!(chunks.next().is_none());

        assert!(chunks.skip_bytes(6).is_err());
    }

    #[test]
    fn test_minimal_index_writes() {
        let mut config = test_config("test_minimal_index_writes");
        config.partial_index_write_interval = 1;
        config.full_index_write_interval = 1;

        {
            let db = LogFs::<Journal2>::open(config.clone()).unwrap();
            db.insert("a", b"a".to_vec()).unwrap();
        }

        let db = LogFs::<Journal2>::open(config.clone()).unwrap();
        assert_eq!(db.get("a").unwrap().unwrap(), b"a");
    }

    #[test]
    fn test_many_index_writes() {
        let mut config = test_config("test_many_index_writes");
        config.partial_index_write_interval = 1;
        config.full_index_write_interval = 2;

        let db = LogFs::<Journal2>::open(config.clone()).unwrap();

        // Insert 100 keys.
        for x in 0..100 {
            eprintln!("writing key {x}");
            db.insert(x.to_string(), x.to_string().into_bytes())
                .unwrap();
        }

        // Rename a third of the keys.

        for x in (0..100).skip(1).step_by(3) {
            eprintln!("renaming key {x}");
            db.rename(x.to_string(), format!("{x}_renamed")).unwrap();
        }

        // delete a third of the keys.
        for x in (0..100).skip(2).step_by(3) {
            eprintln!("deleting key {x}");
            db.remove(x.to_string()).unwrap();
        }

        std::mem::drop(db);

        let db = LogFs::<Journal2>::open(config).unwrap();

        for x in 0..100 {
            if x % 3 == 0 {
                assert_eq!(
                    db.get(x.to_string()).unwrap().unwrap(),
                    x.to_string().into_bytes()
                );
            } else if x % 3 == 1 {
                assert_eq!(
                    db.get(format!("{x}_renamed")).unwrap().unwrap(),
                    x.to_string().into_bytes()
                );
            } else {
                assert_eq!(db.get(x.to_string()).unwrap(), None);
            }
        }
    }

    #[test]
    fn test_batch_writes() {
        let config = test_config("batch_writes");

        {
            let db = LogFs::<Journal2>::open(config.clone()).unwrap();
            for x in 1..20 {
                let val = format!("k{x}");
                db.insert(&val, val.as_bytes().to_vec()).unwrap();
            }

            let batch = Batch::new()
                .and_rename("k1", "n1")
                .and_rename("k2", "n2")
                .and_remove(vec!["k3".to_string(), "k4".to_string(), "k5".to_string()])
                .and_rename("k6", "n6")
                .and_remove(vec!["k7".to_string()]);

            db.batch(batch).unwrap();

            assert_eq!(db.get("n1").unwrap().unwrap(), b"k1");
            assert_eq!(db.get("n2").unwrap().unwrap(), b"k2");
            assert_eq!(db.get("n6").unwrap().unwrap(), b"k6");

            assert_eq!(db.get("k4").unwrap(), None);
            assert_eq!(db.get("k5").unwrap(), None);
            assert_eq!(db.get("k7").unwrap(), None);
        }

        {
            let db = LogFs::<Journal2>::open(config.clone()).unwrap();
            assert_eq!(db.get("n1").unwrap().unwrap(), b"k1");
            assert_eq!(db.get("n2").unwrap().unwrap(), b"k2");
            assert_eq!(db.get("n6").unwrap().unwrap(), b"k6");

            assert_eq!(db.get("k4").unwrap(), None);
            assert_eq!(db.get("k5").unwrap(), None);
            assert_eq!(db.get("k7").unwrap(), None);
        }
    }
}