armdb 0.5.4

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)),
        Err(_) => Ok(false),
    }
}

/// 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 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());
    }
}