walogs 0.1.0

//! Wal handle — see API.md for the public surface and INVARIANTS.md for rules.
//!
//! v2: Multi-segment WAL with rotation and checkpointing.
//! Segment files are named `wal-NNNNNN.log` inside the directory passed to `Wal::open`.

use std::fs::{File, OpenOptions};
use std::io::{BufReader, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};

use crate::error::WalError;
use crate::frame::{self, DecodeOutcome};
use crate::segment;

/// A Log Sequence Number. Newtype wrapper around `u64` to prevent accidental
/// mixing with raw byte offsets, file positions, or counts.
///
/// - `Lsn(0)` is reserved (see I9). The first entry written to a fresh WAL has `Lsn(1)`.
/// - LSNs are monotonically increasing within a single WAL.
/// - LSNs are NOT timestamps.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct Lsn(pub u64);

/// Whether the WAL's tail was clean or truncated on the most recent `Wal::open`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TailState {
    Clean,
    /// A corrupt or partial frame was found starting at this byte offset
    /// within the last segment. The segment has been physically truncated here.
    TruncatedAt(u64),
}

/// Configuration for the WAL.
#[derive(Debug, Clone)]
pub struct WalConfig {
    /// Maximum segment size in bytes before auto-rotation.
    /// `None` disables auto-rotation (caller must call `rotate()` manually).
    /// Default: `Some(64 * 1024 * 1024)` (64 MiB).
    pub max_segment_size: Option<u64>,
}

impl Default for WalConfig {
    fn default() -> Self {
        Self {
            max_segment_size: Some(64 * 1024 * 1024),
        }
    }
}

/// Metadata for a completed (non-active) segment.
#[derive(Debug, Clone)]
struct SegmentMeta {
    seq: u32,
    last_lsn: u64,
}

/// A handle to a multi-segment WAL directory. Single-writer (see I6).
/// Drop closes the file handles.
#[derive(Debug)]
pub struct Wal {
    /// Write handle to the active (last) segment.
    file: File,
    /// Directory containing all segment files.
    dir: PathBuf,
    /// Sequence number of the active segment.
    active_seq: u32,
    /// Next LSN to assign on the next `append`.
    next_lsn: u64,
    /// Current write offset within the active segment.
    write_offset: u64,
    /// Tail state from the most recent `open` scan.
    tail_state: TailState,
    /// Metadata for all completed (non-active) segments, in ascending order.
    segments: Vec<SegmentMeta>,
    /// First LSN in the active segment, or `None` if active segment is empty.
    active_first_lsn: Option<u64>,
    /// Configuration for rotation policy.
    config: WalConfig,
}

impl Wal {
    /// Open or create a WAL in the given directory.
    ///
    /// `dir` is the directory that will contain segment files (`wal-NNNNNN.log`).
    /// If `dir` does not exist, it will be created.
    ///
    /// On open, all segments are scanned to verify integrity. Any corrupt or
    /// partial tail in the last segment is physically truncated (I2).
    /// Corruption in a non-last (completed) segment is fatal and returns `Err`.
    ///
    /// If a v1 WAL file (`wal.log`) exists, it is automatically migrated
    /// to `wal-000001.log`.
    pub fn open(dir: &Path) -> Result<Self, WalError> {
        Self::open_with_config(dir, WalConfig::default())
    }

    /// Open or create a WAL with explicit configuration.
    pub fn open_with_config(dir: &Path, config: WalConfig) -> Result<Self, WalError> {
        // 1. Ensure directory exists (I3 for directory creation).
        let dir_created = !dir.exists();
        std::fs::create_dir_all(dir)?;
        if dir_created && let Some(parent) = dir.parent() {
            fsync_dir(parent)?;
        }

        // 2. Check for v1 migration.
        let v1_path = dir.join(segment::V1_FILENAME);
        let v1_exists = v1_path.exists();
        let mut seqs = segment::scan_segments(dir)?;

        if v1_exists && !seqs.is_empty() {
            return Err(WalError::Corrupt {
                reason: format!("both wal.log and segment files exist in {}", dir.display()),
            });
        }

        if v1_exists {
            // Migrate v1 → v2: rename wal.log → wal-000001.log.
            let v2_path = segment::segment_path(dir, 1);
            std::fs::rename(&v1_path, &v2_path)?;
            fsync_dir(dir)?;
            seqs = vec![1];
        }

        // 3. Create first segment if none exist.
        if seqs.is_empty() {
            let path = segment::segment_path(dir, 1);
            File::create(&path)?;
            fsync_dir(dir)?;
            seqs = vec![1];
        }

        // 4. Verify segment sequence continuity (I11).
        for window in seqs.windows(2) {
            if window[1] != window[0] + 1 {
                return Err(WalError::Corrupt {
                    reason: format!(
                        "gap in segment sequence: wal-{:06}.log followed by wal-{:06}.log",
                        window[0], window[1]
                    ),
                });
            }
        }

        // 5. Scan all segments to build index, find tail, and verify LSN continuity.
        let mut completed_segments: Vec<SegmentMeta> = Vec::new();
        let mut next_lsn = 1u64;
        let mut tail_state = TailState::Clean;
        let mut write_offset = 0u64;
        let mut active_first_lsn: Option<u64> = None;
        let mut prev_last_lsn: Option<u64> = None;
        let active_seq = *seqs
            .last()
            .expect("seqs is non-empty — step 3 ensures at least one segment exists");

        for (idx, &seq) in seqs.iter().enumerate() {
            let is_last = idx == seqs.len() - 1;
            let seg_path = segment::segment_path(dir, seq);

            let seg_file = File::open(&seg_path)?;
            let mut reader = BufReader::new(seg_file);

            let mut seg_first_lsn: Option<u64> = None;
            let mut seg_last_lsn: Option<u64> = None;
            let mut seg_offset = 0u64;

            loop {
                match frame::decode(&mut reader)? {
                    DecodeOutcome::Entry {
                        lsn,
                        bytes_consumed,
                        ..
                    } => {
                        // I8, I9: derive next_lsn from the log itself.
                        if seg_first_lsn.is_none() {
                            seg_first_lsn = Some(lsn);
                        }
                        seg_last_lsn = Some(lsn);
                        next_lsn = lsn + 1;
                        seg_offset += bytes_consumed;
                    }
                    DecodeOutcome::EndOfLog => {
                        if is_last {
                            tail_state = TailState::Clean;
                            write_offset = seg_offset;
                        }
                        break;
                    }
                    DecodeOutcome::Corrupt => {
                        if is_last {
                            // I5: stop at first corruption in last segment.
                            tail_state = TailState::TruncatedAt(seg_offset);
                            write_offset = seg_offset;
                        } else {
                            // Corruption in a completed segment is unrecoverable.
                            return Err(WalError::Corrupt {
                                reason: format!(
                                    "corruption in completed segment wal-{:06}.log at offset {}",
                                    seq, seg_offset
                                ),
                            });
                        }
                        break;
                    }
                }
            }

            // I12: verify LSN continuity across non-empty segments.
            if let Some(first) = seg_first_lsn {
                if let Some(prev) = prev_last_lsn
                    && first != prev + 1
                {
                    return Err(WalError::Corrupt {
                        reason: format!(
                            "LSN discontinuity: segment wal-{:06}.log starts at LSN {}, \
                             expected {} (previous segment ends at LSN {})",
                            seq,
                            first,
                            prev + 1,
                            prev
                        ),
                    });
                }
                prev_last_lsn = seg_last_lsn;
            }

            if is_last {
                active_first_lsn = seg_first_lsn;
            } else if let Some(last) = seg_last_lsn {
                completed_segments.push(SegmentMeta {
                    seq,
                    last_lsn: last,
                });
            }
        }

        // 6. Truncate last segment if needed (I2).
        let active_path = segment::segment_path(dir, active_seq);
        if let TailState::TruncatedAt(off) = tail_state {
            let trunc_file = OpenOptions::new().write(true).open(&active_path)?;
            trunc_file.set_len(off)?;
            trunc_file.sync_all()?;
            fsync_dir(dir)?;
        }

        // 7. Open active segment for writing, positioned at end.
        let mut file = OpenOptions::new()
            .read(true)
            .write(true)
            .open(&active_path)?;
        file.seek(SeekFrom::Start(write_offset))?;

        Ok(Wal {
            file,
            dir: dir.to_path_buf(),
            active_seq,
            next_lsn,
            write_offset,
            tail_state,
            segments: completed_segments,
            active_first_lsn,
            config,
        })
    }

    /// Append a single entry containing `data` to the WAL. Returns the LSN
    /// assigned to it. Fdatasyncs before returning Ok (I4).
    ///
    /// If auto-rotation is configured and the active segment would exceed
    /// `max_segment_size` with this entry, a rotation is performed first.
    #[must_use = "check Ok/Err and use the returned Lsn"]
    pub fn append(&mut self, data: &[u8]) -> Result<Lsn, WalError> {
        let lsn = self.next_lsn;
        let frame_bytes = frame::encode(lsn, data)?;

        // Auto-rotate if configured and segment would overflow.
        // Only rotate if the segment already has data (write_offset > 0),
        // so a single entry larger than max_segment_size still gets written.
        if let Some(max_size) = self.config.max_segment_size
            && self.write_offset > 0
            && self.write_offset + frame_bytes.len() as u64 > max_size
        {
            self.rotate()?;
        }

        // If write_all fails after a partial write, seek back to write_offset so
        // a subsequent append doesn't write at a wrong position (D3/D4).
        self.file.write_all(&frame_bytes).map_err(|e| {
            let _ = self.file.seek(SeekFrom::Start(self.write_offset));
            WalError::Io(e)
        })?;
        // I4: fdatasync before returning Ok. sync_data == fdatasync on unix.
        self.file.sync_data()?;

        if self.active_first_lsn.is_none() {
            self.active_first_lsn = Some(lsn);
        }

        self.write_offset += frame_bytes.len() as u64;
        self.next_lsn += 1;
        Ok(Lsn(lsn))
    }

    /// Manually rotate to a new segment. The current segment is fsynced
    /// and a new empty segment file is created.
    ///
    /// No-op if the active segment is empty (no entries written to it).
    pub fn rotate(&mut self) -> Result<(), WalError> {
        // No-op if active segment has no entries.
        if self.active_first_lsn.is_none() {
            return Ok(());
        }

        // Fsync current segment to ensure all data is durable.
        self.file.sync_all()?;

        // Record completed segment metadata.
        let last_lsn = self.next_lsn - 1;
        self.segments.push(SegmentMeta {
            seq: self.active_seq,
            last_lsn,
        });

        // Create next segment file (I15: create and fsync dir before first write).
        let new_seq = self.active_seq + 1;
        let new_path = segment::segment_path(&self.dir, new_seq);
        let new_file = OpenOptions::new()
            .read(true)
            .write(true)
            .create_new(true)
            .open(&new_path)?;
        fsync_dir(&self.dir)?;

        // Update state for the new active segment.
        self.file = new_file;
        self.active_seq = new_seq;
        self.write_offset = 0;
        self.active_first_lsn = None;

        Ok(())
    }

    /// Delete all completed segments whose entries have all been applied.
    ///
    /// `applied_up_to` is the highest LSN the caller has durably applied
    /// to its state machine. Segments where every entry has
    /// `lsn <= applied_up_to` are deleted.
    ///
    /// The active segment is never deleted (I13). Segments are deleted in
    /// ascending order with a directory fsync after each deletion (I14).
    ///
    /// Returns the number of segments deleted.
    #[must_use = "check Ok/Err; segment deletion errors leave WAL in a partial state"]
    pub fn checkpoint(&mut self, applied_up_to: Lsn) -> Result<usize, WalError> {
        let mut deleted = 0usize;

        // I14: delete in ascending order, fsync dir after each.
        while let Some(seg) = self.segments.first() {
            if seg.last_lsn <= applied_up_to.0 {
                let path = segment::segment_path(&self.dir, seg.seq);
                std::fs::remove_file(&path)?;
                fsync_dir(&self.dir)?;
                self.segments.remove(0);
                deleted += 1;
            } else {
                break;
            }
        }
        // I13: self.segments never includes the active segment,
        // so the active segment is never deleted.

        Ok(deleted)
    }

    /// Iterate every valid entry currently in the WAL, in LSN order.
    /// Opens separate read-only file handles internally; does not affect
    /// the write handle. Walks across all segments transparently.
    pub fn iter(&self) -> WalIter {
        let mut segment_seqs: Vec<u32> = self.segments.iter().map(|s| s.seq).collect();
        segment_seqs.push(self.active_seq);

        WalIter {
            state: IterState::Init {
                dir: self.dir.clone(),
                segment_seqs,
                seq_idx: 0,
            },
        }
    }

    /// Returns the tail state recorded during the most recent `Wal::open` scan.
    pub fn tail_state(&self) -> TailState {
        self.tail_state
    }

    /// Returns the LSN that the next successful `append` will assign.
    #[must_use]
    pub fn next_lsn(&self) -> Lsn {
        Lsn(self.next_lsn)
    }

    /// Returns the number of segment files currently tracked by this handle
    /// (completed segments + the active segment).
    #[must_use]
    pub fn segment_count(&self) -> usize {
        self.segments.len() + 1
    }
}

/// Iterator over WAL entries across all segments, in LSN order.
///
/// Returned by [`Wal::iter`]. Opens separate read-only file handles internally
/// and does not affect the write handle. Walks across all segments transparently.
///
/// Items are `Result<(Lsn, Vec<u8>), WalError>`. A real I/O error is `Err`.
/// Corruption stops the iterator (returns `None`) without producing an `Err`
/// — query [`Wal::tail_state`] afterwards to distinguish a clean end from a
/// corruption-truncated tail.
pub struct WalIter {
    state: IterState,
}

enum IterState {
    /// Haven't opened the current segment file yet.
    Init {
        dir: PathBuf,
        segment_seqs: Vec<u32>,
        seq_idx: usize,
    },
    /// Currently reading from a segment file.
    Reading {
        dir: PathBuf,
        segment_seqs: Vec<u32>,
        seq_idx: usize,
        reader: BufReader<File>,
    },
    /// Iteration is finished.
    Done,
}

impl Iterator for WalIter {
    type Item = Result<(Lsn, Vec<u8>), WalError>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            match std::mem::replace(&mut self.state, IterState::Done) {
                IterState::Init {
                    dir,
                    segment_seqs,
                    seq_idx,
                } => {
                    if seq_idx >= segment_seqs.len() {
                        return None;
                    }
                    let path = segment::segment_path(&dir, segment_seqs[seq_idx]);
                    match File::open(&path) {
                        Ok(f) => {
                            self.state = IterState::Reading {
                                dir,
                                segment_seqs,
                                seq_idx,
                                reader: BufReader::new(f),
                            };
                            continue;
                        }
                        Err(e) => return Some(Err(WalError::Io(e))),
                    }
                }
                IterState::Reading {
                    dir,
                    segment_seqs,
                    seq_idx,
                    mut reader,
                } => match frame::decode(&mut reader) {
                    Ok(DecodeOutcome::Entry { lsn, data, .. }) => {
                        self.state = IterState::Reading {
                            dir,
                            segment_seqs,
                            seq_idx,
                            reader,
                        };
                        return Some(Ok((Lsn(lsn), data)));
                    }
                    Ok(DecodeOutcome::EndOfLog) => {
                        // End of this segment. Advance to the next one.
                        self.state = IterState::Init {
                            dir,
                            segment_seqs,
                            seq_idx: seq_idx + 1,
                        };
                        continue;
                    }
                    // I5 + I10: corruption ends the iterator without yielding Err.
                    Ok(DecodeOutcome::Corrupt) => return None,
                    Err(e) => return Some(Err(WalError::Io(e))),
                },
                IterState::Done => return None,
            }
        }
    }
}

/// Open a directory and `sync_all` it. On unix this is the standard way to
/// flush a parent directory entry to disk so a freshly-created file's
/// existence survives a crash (I3).
fn fsync_dir(path: &Path) -> std::io::Result<()> {
    File::open(path)?.sync_all()
}