armdb 0.14.3

use std::fs::File;

use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};

use crate::error::DbResult;

/// GSN bit layout: bit 63 = tombstone, bit 62 = soft-delete (reserved),
/// bits 61-60 = reserved-for-future flags, bits 59-0 = sequence (60 bits).
/// This build always writes flag bits 62-60 as zero; readers strip the whole
/// flag region from the sequence, so future soft-delete data stays
/// sequence-correct (forward-compatible, no migration).
pub const TOMBSTONE_BIT: u64 = 1 << 63;
/// Reserved for the future soft-delete feature (not implemented here).
pub const SOFT_DELETE_BIT: u64 = 1 << 62;
/// Bits 63-60 — the flag region.
pub const FLAGS_MASK: u64 = 0xF000_0000_0000_0000;
/// Bits 59-0 — the monotonic sequence number.
pub const SEQUENCE_MASK: u64 = !FLAGS_MASK;

/// On-disk entry header. 16 bytes, 8-byte aligned, no padding.
#[derive(Debug, Clone, Copy, FromBytes, IntoBytes, KnownLayout, Immutable)]
#[repr(C)]
pub struct EntryHeader {
    /// Global Sequence Number. Bits 63-60 = flags (63 = tombstone,
    /// 62 = soft-delete (reserved), 61-60 = reserved); bits 59-0 = sequence.
    pub gsn: u64,
    /// CRC32 checksum over (gsn || value_len || key || value).
    pub crc32: u32,
    /// Length of the value in bytes. 0 for tombstones.
    pub value_len: u32,
}

const _: () = assert!(size_of::<EntryHeader>() == 16);
const _: () = assert!(align_of::<EntryHeader>() == 8);

impl EntryHeader {
    #[inline]
    pub fn is_tombstone(&self) -> bool {
        self.gsn & TOMBSTONE_BIT != 0
    }

    /// True if the soft-delete flag is set. Always false for data written by
    /// this build; provided for the future soft-delete feature.
    #[inline]
    pub fn is_soft_deleted(&self) -> bool {
        self.gsn & SOFT_DELETE_BIT != 0
    }

    #[inline]
    pub fn sequence(&self) -> u64 {
        self.gsn & SEQUENCE_MASK
    }
}

/// Reject a GSN that has leaked into the reserved flag bits before it is
/// serialized into an on-disk header. The 60-bit sequence space is never
/// realistically exhausted; this protects the disk-format invariant in
/// release builds, where the companion `debug_assert` is compiled out.
#[inline]
pub(crate) fn guard_gsn(gsn: u64) -> DbResult<()> {
    if gsn & FLAGS_MASK != 0 {
        return Err(crate::error::DbError::FormatMismatch(
            "GSN sequence space exhausted (entered reserved flag bits)".into(),
        ));
    }
    Ok(())
}

#[inline]
pub fn make_tombstone_gsn(gsn: u64) -> u64 {
    gsn | TOMBSTONE_BIT
}

/// Compute CRC32 over gsn || value_len || key || value.
pub fn compute_crc32(gsn: u64, value_len: u32, key: &[u8], value: &[u8]) -> u32 {
    let mut hasher = crc32fast::Hasher::new();
    hasher.update(&gsn.to_ne_bytes());
    hasher.update(&value_len.to_ne_bytes());
    hasher.update(key);
    hasher.update(value);
    hasher.finalize()
}

/// True if the bytes from `offset` to the next 4096-byte boundary are all zero.
/// `read_fn` reads `len` bytes at an offset (plain or decrypting), so this works
/// for both plain and encrypted scans. If `offset` is already page-aligned the
/// caller treats it as clean padding (returns true). A short read returns false
/// so callers fall through to their partial-entry-at-EOF handling.
#[allow(clippy::type_complexity)]
pub fn check_page_padding_at(
    file: &File,
    offset: u64,
    read_fn: &dyn Fn(&File, u64, usize) -> DbResult<Vec<u8>>,
) -> DbResult<bool> {
    const PAGE_SIZE: u64 = 4096;
    let page_end = offset.div_ceil(PAGE_SIZE) * PAGE_SIZE;
    if page_end == offset {
        return Ok(true);
    }
    let remaining = (page_end - offset) as usize;
    match read_fn(file, offset, remaining) {
        Ok(b) => Ok(b.iter().all(|&x| x == 0)),
        // F7: a truncated tail is the ordinary partial-entry-at-EOF case — "not
        // clean padding". Any other error means the gap could not be *read*, which
        // is not the same answer: swallowing it here would silently end recovery
        // at this offset and discard every valid entry after it.
        Err(crate::error::DbError::Io(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
            Ok(false)
        }
        Err(e) => Err(e),
    }
}

/// Page granularity of force-flush padding on page-aligned shards.
pub const PADDING_PAGE_SIZE: u64 = 4096;

/// End (exclusive) of the padding page that contains `offset`.
#[inline]
pub const fn padding_page_end(offset: u64) -> u64 {
    (offset / PADDING_PAGE_SIZE + 1) * PADDING_PAGE_SIZE
}

/// True when the header read at `offset` starts zero page padding rather than an
/// entry, as far as the header itself can tell.
///
/// A force flush pads a partial page with zeros and keeps appending at the next
/// page. Entries are 8-byte aligned, so the padding is any multiple of 8 — an
/// 8-byte gap at in-page offset 4088 is shorter than a header, and the header
/// read there takes its `crc32`/`value_len` from the next entry's GSN. Only the
/// bytes inside `offset`'s own page must be zero. `gsn == 0` is never a real
/// entry: sequences start at 1 and a tombstone sets bit 63.
///
/// Callers still confirm the rest of the page with [`padding_rest_is_clean`].
#[inline]
pub fn is_padding_header(header: &EntryHeader, offset: u64) -> bool {
    let header_in_page = padding_page_end(offset) - offset >= size_of::<EntryHeader>() as u64;
    header.gsn == 0 && (!header_in_page || (header.crc32 == 0 && header.value_len == 0))
}

/// For a header at `offset` accepted by [`is_padding_header`]: true if the bytes
/// after the header up to the end of its page are zero too. A header that
/// reaches (or crosses) the page end leaves nothing more to check.
#[allow(clippy::type_complexity)]
pub fn padding_rest_is_clean(
    file: &File,
    offset: u64,
    read_fn: &dyn Fn(&File, u64, usize) -> DbResult<Vec<u8>>,
) -> DbResult<bool> {
    let header_end = offset + size_of::<EntryHeader>() as u64;
    if header_end >= padding_page_end(offset) {
        return Ok(true);
    }
    check_page_padding_at(file, header_end, read_fn)
}

/// Compute the total on-disk size of an entry including padding to 8-byte alignment.
#[inline]
pub const fn entry_size(key_len: usize, value_len: u32) -> u64 {
    let raw = size_of::<EntryHeader>() + key_len + value_len as usize;
    let padded = (raw + 7) & !7;
    padded as u64
}

/// Serialize a complete entry (header + key + value + padding) into a `Vec<u8>`.
pub fn serialize_entry(gsn: u64, key: &[u8], value: &[u8], tombstone: bool) -> Vec<u8> {
    let actual_gsn = if tombstone {
        make_tombstone_gsn(gsn)
    } else {
        gsn
    };
    let value_len = value.len() as u32;
    let crc = compute_crc32(actual_gsn, value_len, key, value);

    let total = entry_size(key.len(), value_len) as usize;
    let mut buf = vec![0u8; total];

    let header = EntryHeader {
        gsn: actual_gsn,
        crc32: crc,
        value_len,
    };

    buf[..16].copy_from_slice(header.as_bytes());
    buf[16..16 + key.len()].copy_from_slice(key);
    buf[16 + key.len()..16 + key.len() + value.len()].copy_from_slice(value);
    // Remaining bytes are already zeroed (padding).

    buf
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::error::DbError;
    use zerocopy::FromBytes;

    #[test]
    fn test_tombstone_encoding() {
        let gsn = 123u64;
        let tombstone_gsn = make_tombstone_gsn(gsn);
        assert_ne!(tombstone_gsn, gsn);
        assert!(tombstone_gsn & TOMBSTONE_BIT != 0);

        let header = EntryHeader {
            gsn: tombstone_gsn,
            crc32: 0,
            value_len: 0,
        };
        assert!(header.is_tombstone());
        assert_eq!(header.sequence(), gsn);
    }

    #[test]
    fn test_sequence_preserves_value() {
        let header = EntryHeader {
            gsn: 42,
            crc32: 0,
            value_len: 0,
        };
        assert_eq!(header.sequence(), 42);
        assert!(!header.is_tombstone());
    }

    #[test]
    fn test_entry_size_alignment() {
        for key_len in 0..64 {
            for val_len in 0..64u32 {
                let size = entry_size(key_len, val_len);
                assert_eq!(size % 8, 0, "key_len={key_len}, val_len={val_len}");
            }
        }
    }

    #[test]
    fn test_entry_size_exact() {
        // header=16, key=8, val=0 => raw=24, padded=24
        assert_eq!(entry_size(8, 0), 24);
        // header=16, key=8, val=8 => raw=32, padded=32
        assert_eq!(entry_size(8, 8), 32);
        // header=16, key=16, val=1 => raw=33, padded=40
        assert_eq!(entry_size(16, 1), 40);
    }

    #[test]
    fn test_serialize_deserialize() {
        let key = b"test_key";
        let value = b"test_value";
        let gsn = 42u64;

        let buf = serialize_entry(gsn, key, value, false);
        let header = EntryHeader::read_from_bytes(&buf[..16]).expect("failed to parse header");

        assert_eq!(header.gsn, 42);
        assert_eq!(header.value_len, value.len() as u32);

        let expected_crc = compute_crc32(gsn, value.len() as u32, key, value);
        assert_eq!(header.crc32, expected_crc);
    }

    #[test]
    fn test_serialize_tombstone() {
        let key = b"tombkey";
        let value = b"";
        let gsn = 99u64;

        let buf = serialize_entry(gsn, key, value, true);
        let header = EntryHeader::read_from_bytes(&buf[..16]).expect("failed to parse header");

        assert!(header.is_tombstone());
        assert_eq!(header.sequence(), gsn);
    }

    #[test]
    fn test_crc_detects_corruption() {
        let key = b"mykey";
        let value = b"myvalue";
        let gsn = 7u64;

        let mut buf = serialize_entry(gsn, key, value, false);
        let header = EntryHeader::read_from_bytes(&buf[..16]).expect("failed to parse header");
        let original_crc = header.crc32;

        // Flip a byte in the value area (starts at offset 16 + key.len())
        let value_offset = 16 + key.len();
        buf[value_offset] ^= 0xFF;

        // Recompute CRC over the corrupted payload
        let corrupted_value = &buf[value_offset..value_offset + value.len()];
        let new_crc = compute_crc32(header.gsn, header.value_len, key, corrupted_value);
        assert_ne!(new_crc, original_crc);
    }

    #[test]
    fn flag_masks_partition_u64() {
        assert_eq!(FLAGS_MASK | SEQUENCE_MASK, u64::MAX);
        assert_eq!(FLAGS_MASK & SEQUENCE_MASK, 0);
        assert_eq!(SEQUENCE_MASK.count_ones(), 60);
        assert_eq!(TOMBSTONE_BIT, 1 << 63);
        assert_eq!(SOFT_DELETE_BIT, 1 << 62);
    }

    #[test]
    fn sequence_strips_all_flag_bits() {
        let gsn = 42u64 | TOMBSTONE_BIT | SOFT_DELETE_BIT;
        let h = EntryHeader {
            gsn,
            crc32: 0,
            value_len: 0,
        };
        assert_eq!(h.sequence(), 42);
        assert!(h.is_tombstone());
        assert!(h.is_soft_deleted());
    }

    #[test]
    fn soft_deleted_future_record_reads_live_sequence() {
        // Simulates data a future soft-delete-aware build would write.
        let gsn = 1000u64 | SOFT_DELETE_BIT;
        let h = EntryHeader {
            gsn,
            crc32: 0,
            value_len: 0,
        };
        assert_eq!(h.sequence(), 1000);
        assert!(!h.is_tombstone());
        assert!(h.is_soft_deleted());
    }

    #[test]
    fn guard_gsn_accepts_max_sequence_rejects_flag_bits() {
        assert!(guard_gsn(SEQUENCE_MASK).is_ok());
        assert!(guard_gsn(0).is_ok());
        assert!(guard_gsn(SEQUENCE_MASK + 1).is_err()); // bit 60 set
        assert!(guard_gsn(5 | SOFT_DELETE_BIT).is_err());
        assert!(guard_gsn(5 | TOMBSTONE_BIT).is_err());
    }

    /// F7: a transient read error while probing a padding gap must surface, not
    /// read as "not clean padding" — that would silently end recovery early and
    /// discard valid later entries.
    #[test]
    fn check_page_padding_propagates_non_eof_io_error() {
        let file = tempfile::tempfile().expect("tempfile");
        let read_fn = |_: &File, _: u64, _: usize| -> DbResult<Vec<u8>> {
            Err(DbError::Io(std::io::Error::from_raw_os_error(libc::EIO)))
        };
        let err = check_page_padding_at(&file, 8, &read_fn)
            .expect_err("EIO while probing padding must propagate");
        match err {
            DbError::Io(e) => assert_eq!(e.raw_os_error(), Some(libc::EIO)),
            other => panic!("expected Io(EIO), got {other:?}"),
        }
    }

    /// F7: a truncated tail still reads as "not clean padding" — that is the
    /// partial-entry-at-EOF case recovery is built around.
    #[test]
    fn check_page_padding_treats_eof_as_not_clean() {
        let file = tempfile::tempfile().expect("tempfile");
        let read_fn = |_: &File, _: u64, _: usize| -> DbResult<Vec<u8>> {
            Err(DbError::Io(std::io::Error::from(
                std::io::ErrorKind::UnexpectedEof,
            )))
        };
        assert!(
            !check_page_padding_at(&file, 8, &read_fn).expect("EOF is not an error"),
            "a truncated tail is not clean padding"
        );
    }

    /// F7 guard: the success paths keep their meaning — all-zero tail is clean,
    /// any non-zero byte is not, and an already page-aligned offset short-circuits.
    #[test]
    fn check_page_padding_reads_zero_tail_as_clean() {
        let file = tempfile::tempfile().expect("tempfile");
        let zeros = |_: &File, _: u64, n: usize| -> DbResult<Vec<u8>> { Ok(vec![0u8; n]) };
        let dirty = |_: &File, _: u64, n: usize| -> DbResult<Vec<u8>> {
            let mut b = vec![0u8; n];
            b[n - 1] = 1;
            Ok(b)
        };
        assert!(check_page_padding_at(&file, 8, &zeros).expect("zero tail"));
        assert!(!check_page_padding_at(&file, 8, &dirty).expect("dirty tail"));
        // Page-aligned offset: no read at all, so even a failing read_fn is fine.
        let boom = |_: &File, _: u64, _: usize| -> DbResult<Vec<u8>> {
            Err(DbError::Io(std::io::Error::from_raw_os_error(libc::EIO)))
        };
        assert!(check_page_padding_at(&file, 4096, &boom).expect("aligned offset short-circuits"));
    }

    /// An 8-byte gap at in-page offset 4088: the header read there takes its
    /// `crc32`/`value_len` from the next page's entry GSN, yet it is padding.
    #[test]
    fn padding_header_crossing_the_page_end_is_padding() {
        let next_gsn: u64 = 0x0000_0001_9AF1_3B58;
        let crossing = EntryHeader {
            gsn: 0,
            crc32: next_gsn as u32,
            value_len: (next_gsn >> 32) as u32,
        };
        assert!(is_padding_header(&crossing, 4088));
        assert!(is_padding_header(&crossing, 2 * 4096 - 8));
        // Fully inside the page the whole header must be zero.
        assert!(!is_padding_header(&crossing, 4080));
        assert!(!is_padding_header(&crossing, 0));
        let zero = EntryHeader {
            gsn: 0,
            crc32: 0,
            value_len: 0,
        };
        assert!(is_padding_header(&zero, 0));
        assert!(is_padding_header(&zero, 4080));
        // A real entry never has gsn == 0, wherever it sits.
        let real = EntryHeader {
            gsn: 1,
            crc32: 0,
            value_len: 0,
        };
        assert!(!is_padding_header(&real, 4088));
        assert!(!is_padding_header(&real, 0));
    }

    #[test]
    fn padding_page_end_is_the_end_of_the_containing_page() {
        assert_eq!(padding_page_end(0), 4096);
        assert_eq!(padding_page_end(4080), 4096);
        assert_eq!(padding_page_end(4088), 4096);
        assert_eq!(padding_page_end(4096), 8192);
    }

    #[test]
    fn padding_rest_checks_only_bytes_after_the_header_in_its_page() {
        let file = tempfile::tempfile().expect("tempfile");
        let boom = |_: &File, _: u64, _: usize| -> DbResult<Vec<u8>> {
            Err(DbError::Io(std::io::Error::from_raw_os_error(libc::EIO)))
        };
        // Header reaches or crosses the page end: nothing to read.
        assert!(padding_rest_is_clean(&file, 4080, &boom).expect("header ends the page"));
        assert!(padding_rest_is_clean(&file, 4088, &boom).expect("header crosses the page"));

        let seen = std::cell::Cell::new((0u64, 0usize));
        let zeros = |_: &File, off: u64, n: usize| -> DbResult<Vec<u8>> {
            seen.set((off, n));
            Ok(vec![0u8; n])
        };
        assert!(padding_rest_is_clean(&file, 4096, &zeros).expect("zero rest"));
        assert_eq!(seen.get(), (4112, 4096 - 16));
        let dirty = |_: &File, _: u64, n: usize| -> DbResult<Vec<u8>> {
            let mut b = vec![0u8; n];
            b[0] = 1;
            Ok(b)
        };
        assert!(!padding_rest_is_clean(&file, 4000, &dirty).expect("dirty rest"));
    }
}