//! The block-store abstraction and its implementations.
//!
//! The bundle is a single file accessed at byte offsets with an explicit
//! *durable flush* at the commit point (Chapter 8 §"Durable Writes": the flush
//! is `fsync`/`FlushFileBuffers`/equivalent — *"durability MUST be requested
//! explicitly at the commit point"*; implementations *"MAY NOT rely on
//! filesystem ordering guarantees or write-back caching alone"*).
//!
//! [`BlockStore`] captures exactly that contract: positioned reads/writes plus a
//! `flush` that makes prior writes durable. Three implementations:
//!
//! * [`MemStore`] — a plain in-memory image (`flush` is a no-op; the bytes *are*
//!   durable). The default for tests and for the cold-open path over an image.
//! * [`FileStore`] — a real file whose `flush` calls `fsync`. Demonstrates the
//!   atomic-commit protocol against a real filesystem.
//! * [`FaultStore`] — a crash simulator that distinguishes *live* (page-cache)
//!   bytes from *durable* (survives-a-crash) bytes and can "crash" after any
//!   syscall, optionally tearing the in-flight write. This is the engine behind
//!   the crash-recovery fuzzer — Agent D's acceptance gate.

use std::io;

/// A positioned byte store with an explicit durability boundary.
///
/// Writes are not guaranteed durable until [`BlockStore::flush`] returns
/// successfully. A crash (process death) loses any write not yet covered by a
/// successful flush. This is the contract the atomic-commit protocol is built
/// on, and the contract [`FaultStore`] adversarially exercises.
pub trait BlockStore {
    /// The current length of the store, in bytes.
    fn len(&self) -> u64;

    /// Whether the store is empty.
    fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Reads exactly `buf.len()` bytes starting at `offset`. Errors (rather than
    /// short-reads or panics) if the range runs past the end of the store.
    fn read_exact_at(&self, offset: u64, buf: &mut [u8]) -> io::Result<()>;

    /// Writes `data` starting at `offset`, extending the store (zero-filling any
    /// gap) as needed. The write is *not* durable until [`Self::flush`].
    fn write_at(&mut self, offset: u64, data: &[u8]) -> io::Result<()>;

    /// Makes all previously-written bytes durable (the platform `fsync`).
    fn flush(&mut self) -> io::Result<()>;
}

/// Reads exactly `len` bytes at `offset` into a fresh `Vec`.
pub(crate) fn read_vec(store: &dyn BlockStore, offset: u64, len: u64) -> io::Result<Vec<u8>> {
    let mut buf = vec![0u8; len as usize];
    store.read_exact_at(offset, &mut buf)?;
    Ok(buf)
}

fn out_of_range() -> io::Error {
    io::Error::new(io::ErrorKind::UnexpectedEof, "read past end of block store")
}

/// Applies a positioned write to an in-memory byte image, extending and
/// zero-filling as needed.
fn apply_write(image: &mut Vec<u8>, offset: u64, data: &[u8]) {
    let end = offset as usize + data.len();
    if image.len() < end {
        image.resize(end, 0);
    }
    image[offset as usize..end].copy_from_slice(data);
}

// --------------------------------------------------------------------------
// MemStore
// --------------------------------------------------------------------------

/// An in-memory block store. `flush` is a no-op: the bytes are the durable
/// truth. The natural backing for tests, for opening a recovered crash image,
/// and (per QUICKSTART decision 3) for v0's in-memory bundle.
#[derive(Clone, Default)]
pub struct MemStore {
    bytes: Vec<u8>,
}

impl MemStore {
    /// A fresh empty store.
    pub fn new() -> Self {
        MemStore { bytes: Vec::new() }
    }

    /// A store over an existing byte image (e.g. a recovered crash image).
    pub fn from_bytes(bytes: Vec<u8>) -> Self {
        MemStore { bytes }
    }

    /// The store's bytes.
    pub fn as_bytes(&self) -> &[u8] {
        &self.bytes
    }

    /// Consumes the store, returning its bytes.
    pub fn into_bytes(self) -> Vec<u8> {
        self.bytes
    }
}

impl BlockStore for MemStore {
    fn len(&self) -> u64 {
        self.bytes.len() as u64
    }

    fn read_exact_at(&self, offset: u64, buf: &mut [u8]) -> io::Result<()> {
        let end = offset as usize + buf.len();
        if end > self.bytes.len() {
            return Err(out_of_range());
        }
        buf.copy_from_slice(&self.bytes[offset as usize..end]);
        Ok(())
    }

    fn write_at(&mut self, offset: u64, data: &[u8]) -> io::Result<()> {
        apply_write(&mut self.bytes, offset, data);
        Ok(())
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

// --------------------------------------------------------------------------
// FileStore (real fsync)
// --------------------------------------------------------------------------

/// A block store backed by a real file, whose [`BlockStore::flush`] issues a
/// `fsync`. This is the production durability path; the crash fuzzer uses
/// [`FaultStore`] instead because a real `fsync` cannot be interrupted
/// deterministically in a test.
///
/// Unix-only: it uses positioned `pread`/`pwrite` (`FileExt`) so reads need no
/// `&mut`. The crash-recovery gate does not depend on this type.
#[cfg(unix)]
pub struct FileStore {
    file: std::fs::File,
    /// The current file length, tracked in memory. Read once (fallibly) at
    /// construction and maintained on every write, so [`BlockStore::len`] is
    /// infallible and never collapses a transient metadata error to `0` — which
    /// would let a later commit's append overwrite the prelude.
    len: u64,
}

#[cfg(unix)]
impl FileStore {
    /// Creates (or truncates) a bundle file at `path`.
    pub fn create(path: impl AsRef<std::path::Path>) -> io::Result<Self> {
        let file = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(true)
            .open(path)?;
        Ok(FileStore { file, len: 0 })
    }

    /// Opens an existing bundle file for read/write.
    pub fn open(path: impl AsRef<std::path::Path>) -> io::Result<Self> {
        let file = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(path)?;
        // Establish the length once, propagating any metadata error here rather
        // than masking it in `len()`.
        let len = file.metadata()?.len();
        Ok(FileStore { file, len })
    }
}

#[cfg(unix)]
impl BlockStore for FileStore {
    fn len(&self) -> u64 {
        self.len
    }

    fn read_exact_at(&self, offset: u64, buf: &mut [u8]) -> io::Result<()> {
        use std::os::unix::fs::FileExt;
        self.file.read_exact_at(buf, offset)
    }

    fn write_at(&mut self, offset: u64, data: &[u8]) -> io::Result<()> {
        use std::os::unix::fs::FileExt;
        self.file.write_all_at(data, offset)?;
        self.len = self.len.max(offset + data.len() as u64);
        Ok(())
    }

    fn flush(&mut self) -> io::Result<()> {
        // sync_all (fsync) makes both data and the file-size metadata durable;
        // the latter matters because commits extend the file.
        self.file.sync_all()
    }
}

// --------------------------------------------------------------------------
// FaultStore (crash simulator)
// --------------------------------------------------------------------------

/// How a crash that lands on a `flush` interacts with the writes that flush was
/// about to make durable.
#[derive(Copy, Clone, PartialEq, Eq, Debug)]
pub enum Tear {
    /// The flush did not happen: nothing written since the previous successful
    /// flush reaches durable storage. Models a crash that interrupts `fsync`
    /// before any dirty page is written back.
    Clean,
    /// The flush partially happened: every write since the previous flush
    /// reaches durable storage *except* the most recent one, which is torn to
    /// its first `prefix` bytes (the tail keeps its prior durable value). Models
    /// a torn write of the in-flight region — the case the superblock CRC must
    /// catch.
    TornLastWrite { prefix: usize },
}

/// The crash configuration for one fuzzer run.
#[derive(Copy, Clone, PartialEq, Eq, Debug)]
pub struct CrashPoint {
    /// Crash after this many successful syscalls (writes + flushes). A value at
    /// or beyond the commit's total syscall count means "no crash".
    pub after_syscalls: u32,
    /// How a crash landing on a flush tears.
    pub tear: Tear,
}

impl CrashPoint {
    /// A configuration that never crashes (the commit runs to completion).
    pub fn never() -> Self {
        CrashPoint {
            after_syscalls: u32::MAX,
            tear: Tear::Clean,
        }
    }
}

/// A crash-simulating block store. It separates *live* bytes (the page cache,
/// where writes land immediately) from *durable* bytes (what survives a crash,
/// updated only by a successful flush). A configured [`CrashPoint`] makes the
/// store "crash" after a chosen number of syscalls: the crashing syscall and
/// every later one returns an error, and on recovery the bundle is reopened from
/// [`FaultStore::durable_image`].
pub struct FaultStore {
    durable: Vec<u8>,
    live: Vec<u8>,
    /// Writes (offset, len) since the last successful flush, in order.
    pending: Vec<(u64, usize)>,
    /// Syscalls performed so far (writes + flushes).
    issued: u32,
    /// Crash configuration.
    crash: CrashPoint,
    /// Set once the store has crashed; all further syscalls fail.
    crashed: bool,
}

impl FaultStore {
    /// A fault store seeded with a durable image, configured to crash at
    /// `crash`. `live` starts equal to `durable` (a freshly opened file's page
    /// cache matches the disk).
    pub fn new(durable_image: Vec<u8>, crash: CrashPoint) -> Self {
        FaultStore {
            live: durable_image.clone(),
            durable: durable_image,
            pending: Vec::new(),
            issued: 0,
            crash,
            crashed: false,
        }
    }

    /// A fault store that never crashes (used to run a commit fully and learn
    /// its total syscall count and post-commit image).
    pub fn no_fault(durable_image: Vec<u8>) -> Self {
        Self::new(durable_image, CrashPoint::never())
    }

    /// The number of syscalls performed (meaningful after a no-fault run: the
    /// commit's total syscall count, i.e. the exhaustive crash-point bound).
    pub fn syscalls_issued(&self) -> u32 {
        self.issued
    }

    /// Whether the store has crashed.
    pub fn crashed(&self) -> bool {
        self.crashed
    }

    /// The durable image: the bytes that survive a crash at the configured
    /// point. Recovery opens a fresh store over this.
    pub fn durable_image(&self) -> Vec<u8> {
        self.durable.clone()
    }

    /// The current durable image as a [`MemStore`], ready to reopen.
    pub fn recover(&self) -> MemStore {
        MemStore::from_bytes(self.durable_image())
    }

    fn simulated_crash() -> io::Error {
        io::Error::other("simulated crash")
    }

    /// Promotes the most recent pending write torn to `prefix` bytes, and all
    /// earlier pending writes fully. `live` already holds every write; we copy
    /// it into `durable`, then revert the torn tail of the last write to its
    /// prior durable value (or zero, if that region did not exist before).
    fn apply_torn_flush(&mut self, prefix: usize) {
        if let Some(&(off, len)) = self.pending.last() {
            // Capture the prior durable tail before we overwrite `durable`.
            let mut prior_tail = vec![0u8; len];
            for (i, slot) in prior_tail.iter_mut().enumerate() {
                if let Some(b) = self.durable.get(off as usize + i) {
                    *slot = *b;
                }
            }
            self.durable = self.live.clone();
            let torn_from = (off as usize) + prefix.min(len);
            let region_end = off as usize + len;
            for (i, idx) in (torn_from..region_end).enumerate() {
                self.durable[idx] = prior_tail[prefix.min(len) + i];
            }
        } else {
            // Nothing pending: a no-op flush.
            self.durable = self.live.clone();
        }
    }
}

impl BlockStore for FaultStore {
    fn len(&self) -> u64 {
        self.live.len() as u64
    }

    fn read_exact_at(&self, offset: u64, buf: &mut [u8]) -> io::Result<()> {
        // Reads see live bytes (the writing process's own view). Reads are not
        // crash points: the commit path issues only writes and flushes.
        let end = offset as usize + buf.len();
        if end > self.live.len() {
            return Err(out_of_range());
        }
        buf.copy_from_slice(&self.live[offset as usize..end]);
        Ok(())
    }

    fn write_at(&mut self, offset: u64, data: &[u8]) -> io::Result<()> {
        if self.crashed {
            return Err(Self::simulated_crash());
        }
        if self.issued >= self.crash.after_syscalls {
            // The crash lands on a write: it does not reach durable storage
            // (it was never flushed), so durable is simply left as-is.
            self.crashed = true;
            return Err(Self::simulated_crash());
        }
        self.issued += 1;
        apply_write(&mut self.live, offset, data);
        self.pending.push((offset, data.len()));
        Ok(())
    }

    fn flush(&mut self) -> io::Result<()> {
        if self.crashed {
            return Err(Self::simulated_crash());
        }
        if self.issued >= self.crash.after_syscalls {
            // The crash lands on this flush; it persists per the tear mode.
            self.crashed = true;
            match self.crash.tear {
                Tear::Clean => { /* durable unchanged */ }
                Tear::TornLastWrite { prefix } => self.apply_torn_flush(prefix),
            }
            return Err(Self::simulated_crash());
        }
        self.issued += 1;
        // A successful flush makes everything durable.
        self.durable = self.live.clone();
        self.pending.clear();
        Ok(())
    }
}

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

    #[test]
    fn memstore_reads_writes_and_extends() {
        let mut s = MemStore::new();
        s.write_at(4, b"abcd").unwrap();
        assert_eq!(s.len(), 8);
        let mut buf = [0u8; 4];
        s.read_exact_at(4, &mut buf).unwrap();
        assert_eq!(&buf, b"abcd");
        // The gap is zero-filled.
        let mut head = [0xFFu8; 4];
        s.read_exact_at(0, &mut head).unwrap();
        assert_eq!(head, [0, 0, 0, 0]);
    }

    #[test]
    fn memstore_read_past_end_errors() {
        let s = MemStore::from_bytes(vec![1, 2, 3]);
        let mut buf = [0u8; 4];
        assert!(s.read_exact_at(0, &mut buf).is_err());
    }

    #[test]
    fn fault_store_unflushed_writes_are_not_durable() {
        let mut s = FaultStore::new(Vec::new(), CrashPoint::never());
        s.write_at(0, b"hello").unwrap();
        // No flush yet: durable is still empty.
        assert!(s.durable_image().is_empty());
        s.flush().unwrap();
        assert_eq!(s.durable_image(), b"hello");
    }

    #[test]
    fn fault_store_crashes_on_the_configured_syscall() {
        // Allow 1 syscall (the write), crash on the 2nd (the flush), clean.
        let mut s = FaultStore::new(
            Vec::new(),
            CrashPoint {
                after_syscalls: 1,
                tear: Tear::Clean,
            },
        );
        s.write_at(0, b"xyz").unwrap();
        assert!(s.flush().is_err(), "flush is the crashing syscall");
        assert!(s.crashed());
        // Clean crash on the flush: the unflushed write is lost.
        assert!(s.durable_image().is_empty());
    }

    #[test]
    fn fault_store_torn_flush_persists_a_prefix() {
        let mut s = FaultStore::new(
            vec![0xAA; 8],
            CrashPoint {
                after_syscalls: 1,
                tear: Tear::TornLastWrite { prefix: 3 },
            },
        );
        // Overwrite all 8 bytes with 0xBB, but the flush tears after 3 bytes.
        s.write_at(0, &[0xBB; 8]).unwrap();
        assert!(s.flush().is_err());
        let durable = s.durable_image();
        assert_eq!(&durable[0..3], &[0xBB, 0xBB, 0xBB], "prefix persisted");
        assert_eq!(&durable[3..8], &[0xAA; 5], "tail kept its prior value");
    }

    #[test]
    fn fault_store_counts_syscalls() {
        let mut s = FaultStore::no_fault(Vec::new());
        s.write_at(0, b"a").unwrap();
        s.write_at(1, b"b").unwrap();
        s.flush().unwrap();
        assert_eq!(s.syscalls_issued(), 3);
    }
}