haematite 0.7.0

//! Rebuild-from-empty primitives for the migration transaction (§4.2 steps
//! 3–4). Reads are POLICY-FREE (the cursor/tree route by ordered separators, so
//! a v1-shaped tree streams identically under a v2 binary); only the REBUILD
//! carries the target [`TreePolicy`], re-chunking the identical logical map
//! under the new rule.
//!
//! The canonicity witness (§4.2 step 3 note, §9): a shard is rebuilt via ONE
//! `batch_mutate_owned` from empty. History-independence across commit batching
//! (`branch/hi_proptest.rs` `prop_commit_batching_never_changes_the_head_roots`)
//! makes that single batch's root EQUAL the fresh-build canonical root of the
//! same map — so migration is canonical by construction, never a mixed-policy
//! tree.
//!
//! Offline discipline (mirrors `db::vacuum`): these open `DiskStore`/`DurableWal`
//! directly on shard paths under the data dir. They are sound ONLY under the A4
//! writer lock with no live `Database` attached — the orchestrator holds that
//! lock for the whole run.

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

use crate::branch::{BranchRefStore, Timestamp, current_timestamp};
use crate::ids::ShardId;
use crate::shard::router::{SHARD_STORE_DIR, SHARD_WAL_FILE};
use crate::store::{DiskStore, NodeStore};
use crate::tree::{Hash, LeafNode, Node, TreePolicy, batch_mutate_owned};
use crate::wal::{DurableWal, FsyncPolicy, WalRecovery};

use super::super::DatabaseError;
use super::super::vacuum::CANONICAL_REFS_DIR;

/// One shard's `store/` dir and `shard.wal` path under `data_dir`.
fn shard_paths(data_dir: &Path, shard_id: usize) -> (PathBuf, PathBuf, PathBuf) {
    let shard_dir = data_dir.join(format!("shard-{shard_id}"));
    let store_dir = shard_dir.join(SHARD_STORE_DIR);
    let wal_path = shard_dir.join(SHARD_WAL_FILE);
    (shard_dir, store_dir, wal_path)
}

fn fail(context: &str, error: impl std::fmt::Display) -> DatabaseError {
    DatabaseError::MigrationFailed(format!("{context}: {error}"))
}

/// The outcome of rebuilding one shard's committed root.
pub(super) enum ShardRebuild {
    /// The shard directory is absent, or materialised but with no committed
    /// root: nothing to rebuild (a lazy/empty shard stays that way).
    Empty,
    /// The rebuilt root equals the current root — the shard is ALREADY
    /// target-shaped (idempotent rerun / already-migrated), so no WAL marker was
    /// written and the on-disk bytes are untouched.
    Unchanged,
    /// The shard was re-chunked and its new root durably committed.
    Rebuilt,
}

/// Stream every `(key, value)` entry reachable from `root` in ascending key
/// order, RAW (TTL/stamp envelope intact — do NOT decode, or the logical
/// content changes). A single depth-first left-to-right leaf walk over the
/// content-addressed nodes.
fn collect_entries<S: NodeStore + ?Sized>(
    store: &S,
    root: Hash,
    out: &mut Vec<(Vec<u8>, Vec<u8>)>,
) -> Result<(), DatabaseError> {
    let node = store
        .get(&root)
        .map_err(|error| fail("stream shard node", error))?
        .ok_or_else(|| {
            DatabaseError::MigrationFailed(format!("node {root:?} missing while streaming a root"))
        })?;
    match &*node {
        Node::Leaf(leaf) => out.extend(leaf.entries().iter().cloned()),
        Node::Internal(internal) => {
            for (_separator, child) in internal.children() {
                collect_entries(store, *child, out)?;
            }
        }
    }
    Ok(())
}

/// Materialise the empty-leaf root in `store` (the baseline `batch_mutate_owned`
/// loads for a from-empty rebuild — it takes a `Hash`, never an `Option`).
fn empty_root<S: NodeStore + ?Sized>(store: &mut S) -> Result<Hash, DatabaseError> {
    let leaf = LeafNode::new(Vec::new()).map_err(|error| fail("empty leaf", error))?;
    store
        .put(&Node::Leaf(leaf))
        .map_err(|error| fail("store empty leaf", error))
}

/// Rebuild `source_root`'s logical map from empty under `policy`, returning the
/// new root. THE canonical single-batch build (§4.2 step 3).
fn rebuild_root<S: NodeStore + ?Sized>(
    store: &mut S,
    source_root: Hash,
    policy: TreePolicy,
) -> Result<Hash, DatabaseError> {
    let mut entries = Vec::new();
    collect_entries(store, source_root, &mut entries)?;
    let baseline = empty_root(store)?;
    if entries.is_empty() {
        return Ok(baseline);
    }
    let batch: Vec<(Vec<u8>, Option<Vec<u8>>)> = entries
        .into_iter()
        .map(|(key, value)| (key, Some(value)))
        .collect();
    batch_mutate_owned(store, baseline, batch, policy).map_err(|error| fail("rebuild root", error))
}

/// Rebuild one shard's committed root from empty under `policy` and durably
/// commit it (§4.2 step 3). Idempotent: a shard already target-shaped rebuilds
/// to the IDENTICAL root and writes no marker.
pub(super) fn rebuild_shard(
    data_dir: &Path,
    shard_id: usize,
    policy: TreePolicy,
) -> Result<ShardRebuild, DatabaseError> {
    let (shard_dir, store_dir, wal_path) = shard_paths(data_dir, shard_id);
    if !shard_dir.exists() {
        return Ok(ShardRebuild::Empty);
    }
    let mut store = DiskStore::new(&store_dir).map_err(|error| fail("open shard store", error))?;
    let recovered = WalRecovery::recover_path(&wal_path, &store)
        .map_err(|error| fail("recover shard wal", error))?;
    let Some(source_root) = recovered.committed_root() else {
        return Ok(ShardRebuild::Empty);
    };
    let new_root = rebuild_root(&mut store, source_root, policy)?;
    if new_root == source_root {
        return Ok(ShardRebuild::Unchanged);
    }
    // Durability barrier BEFORE the WAL marker (mirrors `ShardActor::commit`): a
    // power loss can never leave a committed-root marker referencing nodes whose
    // directory entries are gone.
    store
        .sync_dirty_dirs()
        .map_err(|error| fail("sync shard store dirs", error))?;
    let mut wal = DurableWal::new(&wal_path, FsyncPolicy::CommitOnly)
        .map_err(|error| fail("open shard wal", error))?;
    wal.commit(new_root)
        .map_err(|error| fail("commit shard root", error))?;
    Ok(ShardRebuild::Rebuilt)
}

/// The canonical in-dir branch refstore location, or `None` when the directory
/// holds no branches (the common case — nothing to rebuild).
fn branches_dir(data_dir: &Path) -> Option<PathBuf> {
    let dir = data_dir.join(CANONICAL_REFS_DIR);
    dir.is_dir().then_some(dir)
}

/// Every branch name (HBR1 head) in the canonical refstore, name order. Empty
/// when the directory has no branches.
pub(super) fn branch_names(data_dir: &Path) -> Result<Vec<String>, DatabaseError> {
    let Some(dir) = branches_dir(data_dir) else {
        return Ok(Vec::new());
    };
    let refs = BranchRefStore::open(&dir).map_err(|error| fail("open branch refstore", error))?;
    Ok(refs.list().map(|record| record.name.clone()).collect())
}

/// The outcome of rebuilding one branch HEAD.
pub(super) enum BranchRebuild {
    /// The branch vanished between enumeration and rebuild, or every per-shard
    /// head is already target-shaped: no record advance (idempotent).
    Unchanged,
    /// At least one per-shard head was re-chunked; the record was advanced.
    Rebuilt,
}

/// Rebuild every per-shard head of one branch (§4.2 step 4) from empty under
/// `policy`, then durably advance the record via [`BranchRefStore::advance`] —
/// the atomic all-shard install that preserves each `fork_anchor` verbatim
/// (anchors stay v1-shaped read-only pins). Fork anchors, snapshots, and
/// commit-log roots are NEVER rebuilt.
///
/// CROSS-LANE: §4.2 brackets this rewrite in the doc-1 manifest `Updating`
/// lifecycle "where a manifest exists". The manifest WRITE authority ships with
/// the sweep lane and is not present here (only `read_manifest`), so this
/// installs heads via the record store's own atomic temp→fsync→rename→dir-fsync
/// discipline and NAMES the manifest bracket as the sweep-lane dependency rather
/// than improvising a manifest writer (mechanism-not-policy).
pub(super) fn rebuild_one_branch(
    data_dir: &Path,
    name: &str,
    policy: TreePolicy,
) -> Result<BranchRebuild, DatabaseError> {
    let Some(dir) = branches_dir(data_dir) else {
        return Ok(BranchRebuild::Unchanged);
    };
    let mut refs =
        BranchRefStore::open(&dir).map_err(|error| fail("open branch refstore", error))?;
    let Some(record) = refs.get(name) else {
        return Ok(BranchRebuild::Unchanged);
    };
    let created = record.created;
    let seq = record.seq;
    let parents = record.parents.clone();
    let shards: Vec<(ShardId, Hash)> = record
        .shards
        .iter()
        .map(|shard| (shard.shard_id, shard.head))
        .collect();

    let mut new_heads: Vec<(ShardId, Hash)> = Vec::with_capacity(shards.len());
    let mut changed = false;
    for (shard_id, head) in shards {
        let (_shard_dir, store_dir, _wal_path) = shard_paths(data_dir, shard_id);
        let mut store = DiskStore::new(&store_dir)
            .map_err(|error| fail("open shard store for branch", error))?;
        let new_head = rebuild_root(&mut store, head, policy)?;
        if new_head != head {
            changed = true;
            store
                .sync_dirty_dirs()
                .map_err(|error| fail("sync branch head store dirs", error))?;
        }
        new_heads.push((shard_id, new_head));
    }
    if !changed {
        return Ok(BranchRebuild::Unchanged);
    }
    let timestamp: Timestamp = current_timestamp();
    refs.advance(name, created, seq, &new_heads, parents, timestamp)
        .map_err(|error| fail("advance branch head", error))?;
    Ok(BranchRebuild::Rebuilt)
}