haematite 0.6.2

//! M1 shard classification and scan (§2): lstat-strict at every model
//! boundary — a symlink is never followed, and only a WHOLLY absent
//! `shard-{id}` entry is lazy (⟨r2, B3⟩). Split from `stats.rs` for the
//! 500-line cap; consumed only by the stats orchestration.

use std::path::{Path, PathBuf};

use crate::shard::router::{SHARD_STORE_DIR, SHARD_WAL_FILE};
use crate::tree::Hash;
use crate::wal::WalRecovery;

use super::super::error::VacuumError;
use super::super::inventory::{StoreInventory, enumerate_store, lstat_kind};
use super::super::report::{EntryKind, MalformedEntry, ShardPresence};

/// One shard's scan product; the inventory dies with it inside
/// [`process_shard`].
pub struct ShardScan {
    pub shard_id: usize,
    pub presence: ShardPresence,
    pub committed_root: Option<Hash>,
    pub store_dir: PathBuf,
    pub inventory: StoreInventory,
    pub shard_malformed: Vec<MalformedEntry>,
}

/// M1 (§2): classify and scan one shard. Every presence decision is
/// lstat-based (Sol r1 fold) — a symlink is NEVER followed at a model
/// boundary, so external data cannot be inventoried as database data and a
/// dangling symlink cannot hide as a lazy shard. Only a WHOLLY absent
/// `shard-{id}` entry is lazy ⟨r2, B3⟩.
pub fn scan_shard(data_dir: &Path, shard_id: usize) -> Result<ShardScan, VacuumError> {
    let shard_dir = data_dir.join(format!("shard-{shard_id}"));
    let store_dir = shard_dir.join(SHARD_STORE_DIR);
    match lstat_kind(&shard_dir)? {
        None => {
            return Ok(ShardScan {
                shard_id,
                presence: ShardPresence::NeverMaterialised,
                committed_root: None,
                store_dir,
                inventory: StoreInventory::default(),
                shard_malformed: Vec::new(),
            });
        }
        Some(EntryKind::Directory) => {}
        Some(kind) => {
            return Err(VacuumError::UnexpectedLayout {
                store: store_dir,
                path: shard_dir,
                reason: format!(
                    "shard entry is a {kind:?}, not a real directory; only a wholly absent \
                     shard-{{id}} is lazy, and symlinks are never followed at model boundaries"
                ),
            });
        }
    }
    let wal_path = shard_dir.join(SHARD_WAL_FILE);
    let store_kind = lstat_kind(&store_dir)?;
    match lstat_kind(&wal_path)? {
        None => {
            // The WAL is created eagerly at boot (`wal/durable.rs`), so
            // dir-present/WAL-missing is damage or a boot cut short — never
            // laziness (⟨r2, B3⟩ / ⟨r15, A3⟩).
            return Err(VacuumError::StoreWithoutWal {
                shard_id,
                shard_dir,
                store_present: store_kind.is_some(),
            });
        }
        Some(EntryKind::File) => {}
        Some(kind) => {
            return Err(VacuumError::UnexpectedLayout {
                store: store_dir,
                path: wal_path,
                reason: format!("shard.wal is a {kind:?}, not a regular file"),
            });
        }
    }
    let recovered = WalRecovery::open(&wal_path)
        .and_then(|mut recovery| recovery.recover_unverified())
        .map_err(|error| VacuumError::CorruptWal {
            shard_id,
            detail: error.to_string(),
        })?;
    if recovered.stopped_at_corruption() {
        return Err(VacuumError::CorruptWal {
            shard_id,
            detail: "replay stopped at a checksum-corrupted frame".to_owned(),
        });
    }
    match store_kind {
        Some(EntryKind::Directory) => {}
        None => {
            return Err(VacuumError::UnexpectedLayout {
                store: store_dir,
                path: shard_dir,
                reason: "shard.wal exists but the store directory is absent; boot creates the \
                         store before the WAL, so this shape is damage"
                    .to_owned(),
            });
        }
        Some(kind) => {
            return Err(VacuumError::UnexpectedLayout {
                store: store_dir.clone(),
                path: store_dir,
                reason: format!("store is a {kind:?}, not a real directory"),
            });
        }
    }
    let shard_malformed = scan_shard_dir(&shard_dir)?;
    let inventory = enumerate_store(&store_dir)?;
    Ok(ShardScan {
        shard_id,
        presence: ShardPresence::Materialised,
        committed_root: recovered.committed_root(),
        store_dir,
        inventory,
        shard_malformed,
    })
}

/// A shard directory holds exactly `store/` and `shard.wal`; anything else
/// is reported and blocks that store's sweep (§4's wrong-model rule, one
/// level up).
fn scan_shard_dir(shard_dir: &Path) -> Result<Vec<MalformedEntry>, VacuumError> {
    let mut malformed = Vec::new();
    let entries = std::fs::read_dir(shard_dir).map_err(|error| VacuumError::Io {
        path: shard_dir.to_path_buf(),
        error,
    })?;
    for entry in entries {
        let entry = entry.map_err(|error| VacuumError::Io {
            path: shard_dir.to_path_buf(),
            error,
        })?;
        let path = entry.path();
        let expected = path
            .file_name()
            .and_then(|name| name.to_str())
            .is_some_and(|name| name == SHARD_STORE_DIR || name == SHARD_WAL_FILE);
        if !expected {
            malformed.push(MalformedEntry {
                path,
                reason: "unexpected entry in shard directory".to_owned(),
            });
        }
    }
    Ok(malformed)
}