haematite 0.6.2

use super::{RecoveredEntry, WalRecovery, decode_payload};
use crate::tree::Hash;
use crate::wal::buffer::{LookupResult, Mutation, WalError};
use crate::wal::entry::{TAG_DELETE, TAG_PUT, WalEntry};
use crate::wal::{DurableWal, FsyncPolicy};
use std::path::{Path, PathBuf};

#[path = "recovery_persist_003_tests.rs"]
mod persist_003;

#[derive(Debug)]
struct TempWal {
    dir: tempfile::TempDir,
    path: PathBuf,
}

impl TempWal {
    fn path(&self) -> &Path {
        debug_assert!(self.path.starts_with(self.dir.path()));
        &self.path
    }
}

fn temp_path(name: &str) -> Result<TempWal, WalError> {
    let dir = tempfile::tempdir()?;
    let path = dir.path().join(name);
    Ok(TempWal { dir, path })
}

fn entry_frame(entry: &WalEntry) -> Vec<u8> {
    let bytes = entry.serialise();
    let mut frame = Vec::new();
    frame.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
    frame.extend_from_slice(&bytes);
    frame
}

fn payload_frame(payload: &[u8]) -> Vec<u8> {
    let mut entry = payload.to_vec();
    entry.extend_from_slice(&crc32fast::hash(payload).to_le_bytes());
    let mut frame = Vec::new();
    frame.extend_from_slice(&(entry.len() as u32).to_le_bytes());
    frame.extend_from_slice(&entry);
    frame
}

fn commit_frame(root: Hash) -> Vec<u8> {
    let mut payload = vec![0x03];
    payload.extend_from_slice(root.as_bytes());
    payload_frame(&payload)
}

fn put_payload(key: &[u8], value: &[u8]) -> Vec<u8> {
    let mut payload = vec![TAG_PUT];
    payload.extend_from_slice(&(key.len() as u32).to_le_bytes());
    payload.extend_from_slice(key);
    payload.extend_from_slice(&(value.len() as u32).to_le_bytes());
    payload.extend_from_slice(value);
    payload
}

fn delete_payload(key: &[u8]) -> Vec<u8> {
    let mut payload = vec![TAG_DELETE];
    payload.extend_from_slice(&(key.len() as u32).to_le_bytes());
    payload.extend_from_slice(key);
    payload
}

fn recover_file(path: &Path) -> Result<crate::wal::WalBuffer, WalError> {
    let mut recovery = WalRecovery::open(path)?;
    recovery.recover(Hash::from_bytes([0; 32]))
}

#[test]
fn open_existing_file_returns_recovery() -> Result<(), WalError> {
    let temp = temp_path("existing.wal")?;
    std::fs::write(temp.path(), b"wal bytes")?;

    let recovery = WalRecovery::open(temp.path())?;

    assert!(format!("{recovery:?}").contains("WalRecovery"));
    Ok(())
}

#[test]
fn open_missing_file_returns_io_error() -> Result<(), WalError> {
    let temp = temp_path("missing.wal")?;

    assert!(matches!(
        WalRecovery::open(temp.path()),
        Err(WalError::Io(_))
    ));
    Ok(())
}

#[test]
fn open_does_not_truncate_existing_file() -> Result<(), WalError> {
    let temp = temp_path("non-truncate.wal")?;
    std::fs::write(temp.path(), b"durable")?;
    let before = std::fs::metadata(temp.path())?.len();

    let recovery = WalRecovery::open(temp.path())?;
    let after = std::fs::metadata(temp.path())?.len();

    assert_eq!(before, after);
    assert!(format!("{recovery:?}").contains("WalRecovery"));
    Ok(())
}

#[test]
fn read_frame_returns_payload_for_well_formed_frame() -> Result<(), WalError> {
    let temp = temp_path("frame.wal")?;
    let payload = put_payload(b"key", b"value");
    std::fs::write(temp.path(), payload_frame(&payload))?;
    let mut recovery = WalRecovery::open(temp.path())?;

    assert_eq!(recovery.read_frame()?, Some(payload));
    Ok(())
}

#[test]
fn read_frame_rejects_crc_mismatch_with_expected_and_actual() -> Result<(), WalError> {
    let temp = temp_path("bad-crc.wal")?;
    let payload = put_payload(b"k", b"v");
    let mut bytes = payload_frame(&payload);
    if let Some(byte) = bytes.last_mut() {
        *byte ^= 0xff;
    }
    std::fs::write(temp.path(), bytes)?;
    let mut recovery = WalRecovery::open(temp.path())?;

    assert!(matches!(
        recovery.read_frame(),
        Err(WalError::ChecksumMismatch { expected, actual })
            if expected != actual && actual == crc32fast::hash(&payload)
    ));
    Ok(())
}

#[test]
fn read_frame_returns_none_at_clean_eof() -> Result<(), WalError> {
    let temp = temp_path("empty.wal")?;
    std::fs::write(temp.path(), [])?;
    let mut recovery = WalRecovery::open(temp.path())?;

    assert_eq!(recovery.read_frame()?, None);
    Ok(())
}

#[test]
fn read_frame_returns_none_for_short_length_prefix() -> Result<(), WalError> {
    let temp = temp_path("short-prefix.wal")?;
    std::fs::write(temp.path(), [1, 0])?;
    let mut recovery = WalRecovery::open(temp.path())?;

    assert_eq!(recovery.read_frame()?, None);
    Ok(())
}

#[test]
fn read_frame_returns_none_for_short_payload() -> Result<(), WalError> {
    let temp = temp_path("short-payload.wal")?;
    let mut bytes = 10u32.to_le_bytes().to_vec();
    bytes.extend_from_slice(b"abc");
    std::fs::write(temp.path(), bytes)?;
    let mut recovery = WalRecovery::open(temp.path())?;

    assert_eq!(recovery.read_frame()?, None);
    Ok(())
}

#[test]
fn read_frame_returns_none_for_missing_checksum() -> Result<(), WalError> {
    let temp = temp_path("missing-checksum.wal")?;
    let payload = put_payload(b"k", b"v");
    let entry_len = u32::try_from(payload.len() + 4).map_err(|_| WalError::LengthOverflow)?;
    let mut bytes = entry_len.to_le_bytes().to_vec();
    bytes.extend_from_slice(&payload);
    std::fs::write(temp.path(), bytes)?;
    let mut recovery = WalRecovery::open(temp.path())?;

    assert_eq!(recovery.read_frame()?, None);
    Ok(())
}

#[test]
fn decode_put_payload_returns_mutation() -> Result<(), WalError> {
    let entry = decode_payload(&put_payload(b"k", b"v"))?;

    assert_eq!(
        entry,
        RecoveredEntry::Mutation(Mutation::Put {
            key: b"k".to_vec(),
            value: b"v".to_vec(),
        })
    );
    Ok(())
}

#[test]
fn decode_delete_payload_returns_mutation() -> Result<(), WalError> {
    let entry = decode_payload(&delete_payload(b"k"))?;

    assert_eq!(
        entry,
        RecoveredEntry::Mutation(Mutation::Delete { key: b"k".to_vec() })
    );
    Ok(())
}

#[test]
fn decode_commit_payload_returns_root_hash() -> Result<(), WalError> {
    let root = Hash::from_bytes([8; 32]);
    let mut payload = vec![0x03];
    payload.extend_from_slice(root.as_bytes());

    assert_eq!(decode_payload(&payload)?, RecoveredEntry::Commit(root));
    Ok(())
}

#[test]
fn decode_unknown_tag_returns_checksum_mismatch_message() {
    let error = decode_payload(&[0xff]).err();

    assert!(matches!(error, Some(WalError::ChecksumMismatch { .. })));
    assert!(matches!(
        error.as_ref().map(std::string::ToString::to_string),
        Some(message) if message.contains("unknown tag") && message.contains("0xff")
    ));
}

#[test]
fn decode_partial_put_field_is_corruption_error() {
    let mut payload = vec![TAG_PUT];
    payload.extend_from_slice(&4u32.to_le_bytes());
    payload.extend_from_slice(b"xy");

    assert!(matches!(
        decode_payload(&payload),
        Err(WalError::ChecksumMismatch { .. })
    ));
}

#[test]
fn recover_replays_only_mutations_after_last_commit() -> Result<(), WalError> {
    let temp = temp_path("after-commit.wal")?;
    let mut bytes = entry_frame(&WalEntry::put(b"a".to_vec(), b"1".to_vec()));
    bytes.extend_from_slice(&commit_frame(Hash::from_bytes([1; 32])));
    bytes.extend_from_slice(&entry_frame(&WalEntry::put(b"b".to_vec(), b"2".to_vec())));
    bytes.extend_from_slice(&entry_frame(&WalEntry::put(b"c".to_vec(), b"3".to_vec())));
    std::fs::write(temp.path(), bytes)?;

    let buffer = recover_file(temp.path())?;

    assert_eq!(buffer.len(), 2);
    assert_eq!(buffer.get(b"a"), LookupResult::NotBuffered);
    assert_eq!(buffer.get(b"b"), LookupResult::BufferedValue(b"2".to_vec()));
    assert_eq!(buffer.get(b"c"), LookupResult::BufferedValue(b"3".to_vec()));
    Ok(())
}

#[test]
fn recover_accepts_truncated_tail_after_commit() -> Result<(), WalError> {
    let temp = temp_path("truncated-tail.wal")?;
    let mut bytes = entry_frame(&WalEntry::put(b"a".to_vec(), b"1".to_vec()));
    bytes.extend_from_slice(&commit_frame(Hash::from_bytes([2; 32])));
    bytes.extend_from_slice(&entry_frame(&WalEntry::put(b"b".to_vec(), b"2".to_vec())));
    bytes.extend_from_slice(&100u32.to_le_bytes());
    bytes.extend_from_slice(b"short");
    std::fs::write(temp.path(), bytes)?;

    let buffer = recover_file(temp.path())?;

    assert_eq!(buffer.len(), 1);
    assert_eq!(buffer.get(b"b"), LookupResult::BufferedValue(b"2".to_vec()));
    Ok(())
}

#[test]
fn recover_replays_all_entries_when_no_commit_exists() -> Result<(), WalError> {
    let temp = temp_path("no-commit.wal")?;
    let mut bytes = entry_frame(&WalEntry::put(b"a".to_vec(), b"1".to_vec()));
    bytes.extend_from_slice(&entry_frame(&WalEntry::put(b"b".to_vec(), b"2".to_vec())));
    std::fs::write(temp.path(), bytes)?;

    let buffer = recover_file(temp.path())?;

    assert_eq!(buffer.len(), 2);
    assert_eq!(buffer.get(b"a"), LookupResult::BufferedValue(b"1".to_vec()));
    assert_eq!(buffer.get(b"b"), LookupResult::BufferedValue(b"2".to_vec()));
    Ok(())
}

#[test]
fn recover_resets_buffer_after_each_commit() -> Result<(), WalError> {
    let temp = temp_path("two-commits.wal")?;
    let mut bytes = entry_frame(&WalEntry::put(b"a".to_vec(), b"1".to_vec()));
    bytes.extend_from_slice(&commit_frame(Hash::from_bytes([4; 32])));
    bytes.extend_from_slice(&entry_frame(&WalEntry::put(b"b".to_vec(), b"2".to_vec())));
    bytes.extend_from_slice(&commit_frame(Hash::from_bytes([5; 32])));
    bytes.extend_from_slice(&entry_frame(&WalEntry::put(b"c".to_vec(), b"3".to_vec())));
    std::fs::write(temp.path(), bytes)?;

    let buffer = recover_file(temp.path())?;

    assert_eq!(buffer.len(), 1);
    assert_eq!(buffer.get(b"a"), LookupResult::NotBuffered);
    assert_eq!(buffer.get(b"b"), LookupResult::NotBuffered);
    assert_eq!(buffer.get(b"c"), LookupResult::BufferedValue(b"3".to_vec()));
    Ok(())
}

#[test]
fn durable_wal_round_trip_recovers_only_post_commit_mutations() -> Result<(), WalError> {
    let temp = temp_path("durable-round-trip.wal")?;
    let committed_root = Hash::from_bytes([7; 32]);
    let mut wal = DurableWal::new(temp.path(), FsyncPolicy::CommitOnly)?;
    wal.append(&WalEntry::put(b"a".to_vec(), b"1".to_vec()))?;
    wal.commit(committed_root)?;
    wal.append(&WalEntry::put(b"b".to_vec(), b"2".to_vec()))?;
    wal.append(&WalEntry::delete(b"gone".to_vec()))?;
    drop(wal);

    let mut recovery = WalRecovery::open(temp.path())?;
    let buffer = recovery.recover(committed_root)?;

    assert_eq!(buffer.len(), 2);
    assert_eq!(buffer.get(b"a"), LookupResult::NotBuffered);
    assert_eq!(buffer.get(b"b"), LookupResult::BufferedValue(b"2".to_vec()));
    assert_eq!(buffer.get(b"gone"), LookupResult::BufferedDelete);
    Ok(())
}

#[test]
fn recover_does_not_modify_wal_file() -> Result<(), WalError> {
    let temp = temp_path("unchanged.wal")?;
    let mut bytes = commit_frame(Hash::from_bytes([6; 32]));
    bytes.extend_from_slice(&entry_frame(&WalEntry::delete(b"gone".to_vec())));
    std::fs::write(temp.path(), bytes)?;
    let before = std::fs::metadata(temp.path())?.len();

    let buffer = recover_file(temp.path())?;
    let after = std::fs::metadata(temp.path())?.len();

    assert_eq!(before, after);
    assert_eq!(buffer.get(b"gone"), LookupResult::BufferedDelete);
    Ok(())
}