velesdb-core 5.0.0

//! Log-structured payload storage with snapshot support.
//!
//! Stores payloads in an append-only log file with an in-memory index.
//! Supports periodic snapshots for fast cold-start recovery.
//!
//! ## WAL Entry Formats
//!
//! **CRC32-protected (current, markers 0xC3/0xC4):**
//! ```text
//! Store:  [0xC3: 1B] [id: 8B LE] [len: 4B LE] [payload: len B] [crc32: 4B LE]
//! Delete: [0xC4: 1B] [id: 8B LE] [crc32: 4B LE]
//! ```
//!
//! **Legacy (markers 1/2, read-only for backward compatibility):**
//! ```text
//! Store:  [1: 1B] [id: 8B LE] [len: 4B LE] [payload: len B]
//! Delete: [2: 1B] [id: 8B LE]
//! ```
//!
//! CRC32 covers all bytes preceding the CRC field. On CRC mismatch during
//! replay, the corrupted entry is skipped and a warning is logged.
//!
//! Snapshot format and I/O are handled by the [`super::snapshot`] module.

use super::log_payload_io::{compute_delete_crc, write_store_record, CRC_DELETE_MARKER};
use super::snapshot;
use super::traits::PayloadStorage;

// Re-export snapshot items for backward compatibility with existing imports
#[allow(unused_imports)] // SNAPSHOT_MAGIC/VERSION used only in test modules
pub(crate) use snapshot::{crc32_hash, SNAPSHOT_MAGIC, SNAPSHOT_VERSION};

use parking_lot::RwLock;
use rustc_hash::FxHashMap;
use std::fs::{File, OpenOptions};
use std::io::{self, BufReader, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};

/// Controls how payload WAL writes are synced to disk.
///
/// - `Fsync` (default): `flush()` + `sync_all()` — full durability, safe against power loss.
/// - `FlushOnly`: `flush()` only — data reaches OS kernel but may be lost on power failure.
/// - `None`: No sync — maximum throughput for bulk imports where data can be re-derived.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum DurabilityMode {
    /// Full durability: flush buffer + fsync to disk.
    #[default]
    Fsync,
    /// Flush buffer to OS only (no fsync). Faster but not power-loss safe.
    FlushOnly,
    /// No sync at all. Maximum throughput for bulk imports.
    None,
}

/// Log-structured payload storage with snapshot support.
///
/// Stores payloads in an append-only log file with an in-memory index.
/// Supports periodic snapshots for O(1) cold-start recovery instead of O(N) WAL replay.
#[allow(clippy::module_name_repetitions)]
pub struct LogPayloadStorage {
    /// Directory path for storage files
    path: PathBuf,
    /// In-memory index: ID -> Offset of length field in WAL
    index: RwLock<FxHashMap<u64, u64>>,
    /// Write-Ahead Log writer (append-only)
    wal: RwLock<io::BufWriter<File>>,
    /// Independent file handle for reading, protected for seeking
    reader: RwLock<File>,
    /// WAL position at last snapshot (0 = no snapshot)
    last_snapshot_wal_pos: RwLock<u64>,
    /// Durability mode for WAL writes
    durability: DurabilityMode,
    /// Tracked WAL write position (avoids flush+metadata syscall for `DurabilityMode::None`)
    write_offset: RwLock<u64>,
}

use super::wal_entry::WalEntry;

impl LogPayloadStorage {
    /// Creates a new `LogPayloadStorage` with the default durability mode (`Fsync`).
    ///
    /// If a snapshot file exists and is valid, loads from snapshot and replays
    /// only the WAL delta for fast startup. Otherwise, falls back to full WAL replay.
    ///
    /// # Errors
    ///
    /// Returns an error if file operations fail.
    pub fn new<P: AsRef<Path>>(path: P) -> io::Result<Self> {
        Self::new_with_durability(path, DurabilityMode::default())
    }

    /// Creates a new `LogPayloadStorage` with the specified durability mode.
    ///
    /// See [`DurabilityMode`] for available modes and their trade-offs.
    ///
    /// # Errors
    ///
    /// Returns an error if file operations fail.
    pub fn new_with_durability<P: AsRef<Path>>(
        path: P,
        durability: DurabilityMode,
    ) -> io::Result<Self> {
        let path = path.as_ref().to_path_buf();
        std::fs::create_dir_all(&path)?;
        let log_path = path.join("payloads.log");

        let wal = Self::open_wal_writer(&log_path)?;
        let (reader, wal_len) = Self::open_wal_reader(&log_path)?;
        let (index, last_snapshot_wal_pos) = Self::load_or_replay_index(&path, &log_path, wal_len)?;

        Ok(Self {
            path,
            index: RwLock::new(index),
            wal: RwLock::new(wal),
            reader: RwLock::new(reader),
            last_snapshot_wal_pos: RwLock::new(last_snapshot_wal_pos),
            durability,
            write_offset: RwLock::new(wal_len),
        })
    }

    /// Opens the WAL file for append-mode writing.
    fn open_wal_writer(log_path: &Path) -> io::Result<io::BufWriter<File>> {
        let writer_file = OpenOptions::new()
            .create(true)
            .append(true)
            .open(log_path)?;
        Ok(io::BufWriter::new(writer_file))
    }

    /// Opens the WAL file for random-access reading, creating it if absent.
    ///
    /// Returns the reader handle and the current WAL length in bytes.
    fn open_wal_reader(log_path: &Path) -> io::Result<(File, u64)> {
        if !log_path.exists() {
            File::create(log_path)?;
        }
        let reader = File::open(log_path)?;
        let wal_len = reader.metadata()?.len();
        Ok((reader, wal_len))
    }

    /// Loads the payload index, trying a snapshot first, falling back to full WAL replay.
    ///
    /// Returns `(index, last_snapshot_wal_position)`.
    fn load_or_replay_index(
        dir: &Path,
        log_path: &Path,
        wal_len: u64,
    ) -> io::Result<(FxHashMap<u64, u64>, u64)> {
        let snapshot_path = dir.join("payloads.snapshot");
        match snapshot::load_snapshot(&snapshot_path) {
            Ok((snapshot_index, snapshot_wal_pos)) => {
                let index =
                    Self::replay_wal_from(log_path, snapshot_index, snapshot_wal_pos, wal_len)?;
                Ok((index, snapshot_wal_pos))
            }
            Err(e) => {
                // `NotFound` is the expected cold-start case (no snapshot yet).
                // Any other error kind means a snapshot existed but failed to
                // load (corrupt header, truncated file, bad CRC) — surface it
                // so operators can see recovery fell back to a full WAL
                // replay instead of the fast path, even though the fallback
                // itself is correct.
                if e.kind() != io::ErrorKind::NotFound {
                    tracing::warn!(
                        error = %e,
                        path = %snapshot_path.display(),
                        "payload snapshot failed to load, falling back to full WAL replay"
                    );
                }
                let index = Self::replay_wal_from(log_path, FxHashMap::default(), 0, wal_len)?;
                Ok((index, 0))
            }
        }
    }

    /// Applies the configured durability mode to a WAL writer.
    fn sync_wal(wal: &mut io::BufWriter<File>, mode: DurabilityMode) -> io::Result<()> {
        match mode {
            DurabilityMode::Fsync => {
                wal.flush()?;
                wal.get_ref().sync_all()?;
            }
            DurabilityMode::FlushOnly => {
                wal.flush()?;
            }
            DurabilityMode::None => {}
        }
        Ok(())
    }

    /// Syncs the WAL according to durability mode, resyncing `write_offset`
    /// with the actual file length on failure to prevent desync on subsequent
    /// writes.
    ///
    /// RF-2: Shared by `store` and `delete` to eliminate duplicated
    /// sync-and-resync-offset error handling.
    fn sync_wal_or_resync(
        wal: &mut io::BufWriter<File>,
        mode: DurabilityMode,
        offset: &mut u64,
    ) -> io::Result<()> {
        if let Err(e) = Self::sync_wal(wal, mode) {
            if let Ok(meta) = wal.get_ref().metadata() {
                *offset = meta.len();
            }
            return Err(e);
        }
        Ok(())
    }

    /// Replays WAL entries from `start_pos` to `end_pos`, updating the index.
    fn replay_wal_from(
        log_path: &Path,
        mut index: FxHashMap<u64, u64>,
        start_pos: u64,
        end_pos: u64,
    ) -> io::Result<FxHashMap<u64, u64>> {
        if start_pos >= end_pos {
            return Ok(index);
        }

        let file = File::open(log_path)?;
        let mut reader_buf = BufReader::new(file);
        reader_buf.seek(SeekFrom::Start(start_pos))?;

        let mut pos = start_pos;
        while pos < end_pos {
            // `read` returns `None` to stop cleanly on a torn tail (crash
            // mid-append) or on mid-stream corruption (unknown marker), keeping
            // every entry replayed so far; see `wal_entry`'s policy.
            let Some(entry) = WalEntry::read(&mut reader_buf, pos) else {
                break;
            };
            // `apply` returns `Ok(None)` for a torn tail in the payload region
            // (short/oversized final record): stop cleanly, keeping prior entries.
            let Some(next_pos) = entry.apply(&mut index, &mut reader_buf, end_pos)? else {
                break;
            };
            pos = next_pos;
        }

        Ok(index)
    }

    /// Creates a snapshot of the current index state.
    ///
    /// The snapshot captures:
    /// - Current WAL position
    /// - All index entries (ID -> offset mappings)
    /// - CRC32 checksum for integrity
    ///
    /// # Errors
    ///
    /// Returns an error if file operations fail.
    pub fn create_snapshot(&mut self) -> io::Result<()> {
        // Flush WAL before snapshotting to ensure data is on disk for the reader
        {
            let mut wal = self.wal.write();
            wal.flush()?;
            wal.get_ref().sync_all()?;
        }

        let index = self.index.read();
        let wal_pos = *self.write_offset.read();

        snapshot::create_snapshot_file(&self.path, &index, wal_pos)?;

        *self.last_snapshot_wal_pos.write() = wal_pos;

        Ok(())
    }

    /// Returns whether a new snapshot should be created.
    ///
    /// Heuristic: Returns true if WAL has grown by more than the default threshold
    /// bytes since the last snapshot.
    #[must_use]
    pub fn should_create_snapshot(&self) -> bool {
        snapshot::should_create_snapshot(
            *self.last_snapshot_wal_pos.read(),
            *self.write_offset.read(),
        )
    }

    /// Attempts to create a snapshot if the WAL has grown past the threshold.
    ///
    /// Best-effort: on failure the error is logged but not propagated,
    /// because the WAL write that triggered the check already succeeded.
    fn maybe_auto_snapshot(&mut self) {
        if self.should_create_snapshot() {
            if let Err(e) = self.create_snapshot() {
                tracing::warn!(
                    error = %e,
                    "Auto-snapshot after WAL growth failed; will retry on next write"
                );
            }
        }
    }

    /// Stores multiple payloads in a single batch operation.
    ///
    /// Optimized for bulk imports: acquires WAL + index + offset locks once,
    /// writes all records sequentially, and performs a **single** durability
    /// sync at the end instead of per-point fsync.
    ///
    /// # Errors
    ///
    /// Returns an error if serialization or WAL write fails. On partial failure,
    /// entries written before the error are durable (WAL is append-only).
    pub fn store_batch(&mut self, entries: &[(u64, &serde_json::Value)]) -> io::Result<()> {
        self.store_batch_inner(entries, true)
    }

    /// Stores multiple payloads without forcing an fsync at the end.
    ///
    /// Identical to [`store_batch`](Self::store_batch) except the final
    /// `sync_all()` is replaced by a buffer-only `flush()`. WAL entries are
    /// written and the `BufWriter` is flushed to the OS kernel, but not
    /// fsynced to disk.
    ///
    /// Use this for intermediate batches in a streaming bulk import, where
    /// only the final batch needs full durability. Call
    /// [`PayloadStorage::flush()`] after the last batch to force fsync.
    ///
    /// # Errors
    ///
    /// Returns an error if serialization or WAL write fails.
    pub fn store_batch_deferred(
        &mut self,
        entries: &[(u64, &serde_json::Value)],
    ) -> io::Result<()> {
        self.store_batch_inner(entries, false)
    }

    /// Shared implementation for [`store_batch`] and [`store_batch_deferred`].
    ///
    /// When `fsync` is `true`, the configured durability mode is applied.
    /// When `false`, only a buffer flush is performed (no `sync_all`).
    fn store_batch_inner(
        &mut self,
        entries: &[(u64, &serde_json::Value)],
        fsync: bool,
    ) -> io::Result<()> {
        if entries.is_empty() {
            return Ok(());
        }

        {
            let mut wal = self.wal.write();
            let mut index = self.index.write();
            let mut offset = self.write_offset.write();
            let mut record_buf = Vec::with_capacity(256);

            for &(id, payload) in entries {
                write_store_record(
                    &mut wal,
                    id,
                    payload,
                    &mut offset,
                    &mut index,
                    &mut record_buf,
                )?;
            }

            if fsync {
                Self::sync_wal_or_resync(&mut wal, self.durability, &mut offset)?;
            } else {
                // Buffer-only flush: data reaches OS kernel but is not fsynced.
                // Safe for intermediate batches — caller must fsync after the
                // final batch.
                wal.flush()?;
            }
        }

        self.maybe_auto_snapshot();
        Ok(())
    }

    /// Deletes multiple payloads under ONE durability barrier.
    ///
    /// Batched counterpart of [`PayloadStorage::delete`], which syncs the WAL
    /// per id: calling it in a loop cost one fsync PER POINT on this store
    /// (finding C3). Here every CRC-protected tombstone (`Marker(0xC4) |
    /// ID(8) | CRC32(4)`) is built into one buffer, written with a single
    /// `write_all`, and synced once via [`Self::sync_wal_or_resync`] — the
    /// bytes are identical to N sequential `delete` calls, so WAL replay is
    /// unchanged. Ids absent from the index are skipped (no tombstone),
    /// mirroring the single-delete guard.
    ///
    /// # Crash contract
    ///
    /// The batch's tombstones become durable together: a crash before the
    /// sync loses ALL of them (the payloads simply remain live on replay —
    /// nothing is half-applied), a crash after it persists all. A torn tail
    /// inside the record group is skipped by CRC framing on replay. Index
    /// entries are removed only after the sync succeeded, so acknowledged
    /// in-memory state never runs ahead of the WAL.
    ///
    /// # Errors
    ///
    /// Returns an error if the WAL write or sync fails; the index is left
    /// untouched in that case.
    pub fn delete_batch(&mut self, ids: &[u64]) -> io::Result<()> {
        /// Tombstone record size: Marker(1) + ID(8) + CRC32(4).
        const TOMBSTONE_BYTES: u64 = 1 + 8 + 4;

        // SAFETY: `&mut self` serializes all writers, so the read-then-write
        // gap below cannot race a concurrent `store(id)` (same argument as
        // the single-`delete` guard).
        let live: Vec<u64> = {
            let index = self.index.read();
            ids.iter()
                .copied()
                .filter(|id| index.contains_key(id))
                .collect()
        };
        if live.is_empty() {
            return Ok(());
        }

        // Scoped block: all lock guards are released before the auto-snapshot
        // check, which itself acquires locks (see `create_snapshot`).
        {
            let mut wal = self.wal.write();
            let mut index = self.index.write();
            let mut offset = self.write_offset.write();

            let mut records = Vec::with_capacity(live.len() * 13);
            for &id in &live {
                records.push(CRC_DELETE_MARKER);
                records.extend_from_slice(&id.to_le_bytes());
                records.extend_from_slice(&compute_delete_crc(id).to_le_bytes());
            }
            wal.write_all(&records)?;

            // ONE sync for the whole batch (resync offset on failure).
            Self::sync_wal_or_resync(&mut wal, self.durability, &mut offset)?;

            for &id in &live {
                *offset += TOMBSTONE_BYTES;
                index.remove(&id);
            }
        }

        self.maybe_auto_snapshot();
        Ok(())
    }
}

impl PayloadStorage for LogPayloadStorage {
    fn store(&mut self, id: u64, payload: &serde_json::Value) -> io::Result<()> {
        // Scoped block: lock guards released before auto-snapshot (which acquires locks).
        {
            let mut wal = self.wal.write();
            let mut index = self.index.write();
            let mut offset = self.write_offset.write();
            let mut record_buf = Vec::new();

            write_store_record(
                &mut wal,
                id,
                payload,
                &mut offset,
                &mut index,
                &mut record_buf,
            )?;

            Self::sync_wal_or_resync(&mut wal, self.durability, &mut offset)?;
        }

        self.maybe_auto_snapshot();
        Ok(())
    }

    fn retrieve(&self, id: u64) -> io::Result<Option<serde_json::Value>> {
        let index = self.index.read();
        let Some(&offset) = index.get(&id) else {
            return Ok(None);
        };
        drop(index);

        // H-2: Only flush when DurabilityMode::None is configured, because sync_wal()
        // already flushes the BufWriter after every write in Fsync and FlushOnly modes.
        // Skipping this avoids acquiring the WAL write lock on every read, which would
        // serialize all readers behind writers.
        if self.durability == DurabilityMode::None {
            self.wal.write().flush()?;
        }

        // Positional reads (`read_at`/`seek_read`) take `&File` and never touch
        // a shared file cursor, so a *shared* read lock is enough: concurrent
        // hydrations no longer serialize behind an exclusive write lock. The
        // guard is also released before `serde_json::from_slice` so deserialize
        // runs fully outside the lock.
        let payload_bytes = {
            let reader = self.reader.read();
            let file_len = reader.metadata()?.len();
            read_length_prefixed_payload(&reader, offset, file_len)?
        };

        let payload = serde_json::from_slice(&payload_bytes)
            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;

        Ok(Some(payload))
    }

    fn delete(&mut self, id: u64) -> io::Result<()> {
        // If the id is not in the index there is nothing for a tombstone to
        // shadow on WAL replay. Without this guard, callers that issue
        // per-entry deletes (e.g. `write_deduped_payloads` with all-None
        // payloads) pay one fsync per never-stored id.
        //
        // SAFETY: `&mut self` serializes all writers, so the read-then-write
        // gap below cannot race a concurrent `store(id)`. If this signature
        // is ever relaxed to `&self`, replace this with a single write-lock
        // acquisition or fold the check inside the existing scoped block.
        if !self.index.read().contains_key(&id) {
            return Ok(());
        }

        let crc = compute_delete_crc(id);

        // Scoped block: all lock guards are released before the auto-snapshot
        // check, which itself acquires locks (see `create_snapshot`).
        {
            let mut wal = self.wal.write();
            let mut index = self.index.write();
            let mut offset = self.write_offset.write();

            // H-3: Build complete delete record in one buffer to minimize partial-write window.
            // CRC-protected format: Marker(0xC4) | ID(8) | CRC32(4)
            let mut record = [0u8; 1 + 8 + 4];
            record[0] = CRC_DELETE_MARKER;
            record[1..9].copy_from_slice(&id.to_le_bytes());
            record[9..13].copy_from_slice(&crc.to_le_bytes());
            wal.write_all(&record)?;

            // Sync WAL according to durability mode (resync offset on failure).
            Self::sync_wal_or_resync(&mut wal, self.durability, &mut offset)?;

            *offset += 1 + 8 + 4; // Marker(1) + ID(8) + CRC32(4)
            index.remove(&id);
        }

        self.maybe_auto_snapshot();
        Ok(())
    }

    fn flush(&mut self) -> io::Result<()> {
        let mut wal = self.wal.write();
        Self::sync_wal(&mut wal, self.durability)
    }

    fn ids(&self) -> Vec<u64> {
        self.index.read().keys().copied().collect()
    }
}

/// Reads `buf.len()` bytes from `file` starting at absolute `offset` without
/// disturbing any shared file cursor.
///
/// Uses positional I/O (`pread` on Unix via [`std::os::unix::fs::FileExt`],
/// overlapped `ReadFile` on Windows via [`std::os::windows::fs::FileExt`]), so
/// the same `&File` can be read concurrently from many threads under a shared
/// lock — the offset is passed to the syscall rather than seeked on the handle.
#[cfg(unix)]
fn read_exact_at(file: &File, buf: &mut [u8], offset: u64) -> io::Result<()> {
    use std::os::unix::fs::FileExt;
    file.read_exact_at(buf, offset)
}

/// Windows counterpart to [`read_exact_at`]. `seek_read` carries the offset in
/// an `OVERLAPPED` structure (it does not rely on the shared cursor), so it is
/// safe under concurrent shared-locked reads; it may return short, so loop
/// until `buf` is filled.
#[cfg(windows)]
fn read_exact_at(file: &File, buf: &mut [u8], offset: u64) -> io::Result<()> {
    use std::os::windows::fs::FileExt;
    let mut filled = 0usize;
    while filled < buf.len() {
        let read = file.seek_read(&mut buf[filled..], offset + filled as u64)?;
        if read == 0 {
            return Err(io::Error::new(
                io::ErrorKind::UnexpectedEof,
                "failed to fill whole buffer",
            ));
        }
        filled += read;
    }
    Ok(())
}

/// Reads a length-prefixed payload at `offset`: a 4-byte LE length followed by
/// that many payload bytes. The declared length is bounded by the bytes
/// remaining after the prefix (OOM guard #897/#898) before allocating.
///
/// Uses positional reads on a borrowed `&File`, so callers only need a shared
/// read lock — the file cursor is never mutated.
fn read_length_prefixed_payload(file: &File, offset: u64, file_len: u64) -> io::Result<Vec<u8>> {
    let mut len_bytes = [0u8; 4];
    read_exact_at(file, &mut len_bytes, offset)?;
    let declared = u64::from(u32::from_le_bytes(len_bytes));

    // OOM guard (#897/#898): a corrupt length field must not drive an unbounded
    // allocation. The payload cannot exceed the bytes remaining after the prefix.
    let pos_after_len = offset.saturating_add(4);
    let remaining = file_len.saturating_sub(pos_after_len);
    if declared > remaining {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            "payload length exceeds file size",
        ));
    }
    let len = usize::try_from(declared)
        .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "payload length overflow"))?;

    let mut payload_bytes = vec![0u8; len];
    read_exact_at(file, &mut payload_bytes, pos_after_len)?;
    Ok(payload_bytes)
}