haematite 0.6.1

//! A4: the blessed read-only observer over a live database's data directory.
//!
//! [`ReadOnlyDatabase`] opens a data dir WITHOUT the exclusive writer lock and
//! without spawning any shard actor, WAL append handle, or sweep — it takes no
//! write path at all. It serves committed state only, re-deriving each shard's
//! committed root per call so it always tracks the live writer's latest commit.
//! The full lockfile/observer design note lives in
//! docs/design/LOCKFILE-SEMANTICS.md.

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

use crate::shard::router::{SHARD_STORE_DIR, SHARD_WAL_FILE, shard_dir, shard_index_for};
use crate::store::{DiskStore, StoreError};
use crate::tree::{Cursor, Hash, TreeError};
use crate::ttl::{TtlDecodeError, Visibility, visible_value};
use crate::wal::{WalError, WalRecovery};

use super::DatabaseError;
use super::config::read_config;

/// One committed `(key, value)` entry returned by [`ReadOnlyDatabase::range`].
pub type ObservedEntry = (Vec<u8>, Vec<u8>);

/// A read-only observer of a haematite data directory, safe to run alongside
/// ONE live writer process (and any number of other observers).
///
/// # Why this is safe without a lock (the exact argument)
///
/// 1. **Node files are immutable and content-addressed.** The writer publishes
///    each node with temp-file → fsync → no-clobber rename and never modifies
///    an existing node file in place (re-putting an existing node is a no-op),
///    so any node file the observer can open is complete and final.
/// 2. **The WAL is replaced, not edited, at commit.** Commit truncation writes
///    a fresh file and atomically renames it over `shard.wal`; a reader sees
///    either the old complete file or the new complete file, never a torn mix.
///    Live appends only land after existing bytes, and recovery is
///    tail-tolerant: an incomplete tail frame stops replay cleanly.
/// 3. **A committed-root marker is written only after its nodes are durable**
///    (node-file fsyncs and the dirty-dir barrier strictly precede the
///    marker), so a root the observer adopts is fully readable at adoption.
/// 4. **The observer takes no write path.** The WAL is opened read-only via
///    [`WalRecovery`] (which never creates, truncates, or appends); a missing
///    shard directory is read as the empty tree, never created.
///
/// It deliberately takes NO shared lock either: observers must never be able
/// to deny startup to the legitimate writer, and no-lock keeps observation
/// working on read-only mounts, snapshots, and restored backups.
///
/// # Semantics, honestly
///
/// - Reads reflect **committed state only** — the writer's uncommitted WAL
///   buffer is invisible (unlike [`super::Database::get`], which overlays it).
/// - Each call re-reads the shard's committed root, so successive calls track
///   the writer's commits; two calls may observe different commits.
/// - A concurrent prune can delete historical nodes mid-traversal; that
///   surfaces as a typed error ([`TreeError::MissingNode`] /
///   [`WalError::MissingCommittedRoot`]) — never silent corruption — and a
///   retry re-adopts the current committed root.
#[derive(Debug)]
pub struct ReadOnlyDatabase {
    data_dir: PathBuf,
    shard_count: usize,
}

impl ReadOnlyDatabase {
    /// Open an existing database directory for read-only observation.
    ///
    /// Reads `config.json` for the shard layout and nothing else: no writer
    /// lock is taken, no shard actor is spawned, nothing is created on disk.
    /// See the type-level docs for the safety argument and read semantics.
    pub fn open(path: impl AsRef<Path>) -> Result<Self, ObserverError> {
        let data_dir = path.as_ref().to_path_buf();
        let config = read_config(&data_dir).map_err(ObserverError::Config)?;
        if config.shard_count == 0 {
            return Err(ObserverError::Config(DatabaseError::InvalidShardCount));
        }
        Ok(Self {
            data_dir,
            shard_count: config.shard_count,
        })
    }

    /// The fixed shard count this database was created with.
    #[must_use]
    pub const fn shard_count(&self) -> usize {
        self.shard_count
    }

    /// The shard index that owns `key` — byte-identical to the live router's
    /// `BLAKE3(key)[..8] % shard_count` convention.
    #[must_use]
    pub fn shard_for(&self, key: &[u8]) -> usize {
        shard_index_for(key, self.shard_count)
    }

    /// Read `key` from the owning shard's COMMITTED tree.
    ///
    /// Applies the same read-time TTL/tombstone visibility filter as the live
    /// read path, so an expired entry or a stamped tombstone reads as `None`.
    /// The writer's uncommitted WAL buffer is invisible here.
    pub fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, ObserverError> {
        let Some((store, root)) = self.committed_shard_state(self.shard_for(key))? else {
            return Ok(None);
        };
        let Some(encoded) = Cursor::new(&store, root).get(key)? else {
            return Ok(None);
        };
        Ok(visible_value(&encoded)?.into_option())
    }

    /// Read a shard-local `[from, to)` key range from the COMMITTED tree, in
    /// ascending key order.
    ///
    /// Routed to the shard owning `from`, mirroring [`super::Database::range`];
    /// expired entries and tombstones are filtered exactly like the live read
    /// path. `from >= to` returns an empty result without routing.
    pub fn range(&self, from: &[u8], to: &[u8]) -> Result<Vec<ObservedEntry>, ObserverError> {
        if from >= to {
            return Ok(Vec::new());
        }
        let Some((store, root)) = self.committed_shard_state(self.shard_for(from))? else {
            return Ok(Vec::new());
        };
        let cursor = Cursor::new(&store, root);
        let mut entries = Vec::new();
        for item in cursor.range(from, to) {
            let (key, encoded) = item?;
            if let Visibility::Live(value) = visible_value(&encoded)? {
                entries.push((key, value));
            }
        }
        Ok(entries)
    }

    /// The committed root of shard `shard_id`, or `None` if the shard has
    /// never committed (including never having been materialised at all).
    pub fn committed_root(&self, shard_id: usize) -> Result<Option<Hash>, ObserverError> {
        if shard_id >= self.shard_count {
            return Err(ObserverError::ShardOutOfRange {
                shard_id,
                shard_count: self.shard_count,
            });
        }
        Ok(self
            .committed_shard_state(shard_id)?
            .map(|(_store, root)| root))
    }

    /// Open shard `shard_id`'s node store and read its current committed root
    /// from the WAL, strictly read-only.
    ///
    /// `None` means the shard holds no committed data: its directory was never
    /// materialised, or its WAL has no committed-root marker yet. The store
    /// directory's existence is checked FIRST so an observer never creates it
    /// (the writer's `DiskStore` constructor would).
    fn committed_shard_state(
        &self,
        shard_id: usize,
    ) -> Result<Option<(DiskStore, Hash)>, ObserverError> {
        let shard_dir = shard_dir(&self.data_dir, shard_id);
        let store_dir = shard_dir.join(SHARD_STORE_DIR);
        match fs::metadata(&store_dir) {
            Ok(metadata) if metadata.is_dir() => {}
            Ok(_metadata) => {
                return Err(ObserverError::Store(StoreError::NotADirectory {
                    path: store_dir,
                }));
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(ObserverError::Store(StoreError::Io(error))),
        }
        let store = DiskStore::new(&store_dir)?;
        // Read-only recovery: never creates, truncates, or appends. A missing
        // WAL is a fresh shard; the committed root (if any) is verified to be
        // present in the store before it is adopted.
        let recovered = WalRecovery::recover_path(shard_dir.join(SHARD_WAL_FILE), &store)?;
        Ok(recovered.committed_root().map(|root| (store, root)))
    }
}

/// Errors raised by the read-only observer.
#[derive(Debug)]
pub enum ObserverError {
    /// The data dir is not an openable database: `config.json` is missing,
    /// unreadable, unparsable, or names an invalid shard layout.
    Config(DatabaseError),
    /// A shard's node store could not be opened or read.
    Store(StoreError),
    /// A shard's WAL could not be scanned for its committed root.
    Wal(WalError),
    /// Traversing a committed tree failed (including a node pruned by the
    /// writer mid-traversal — retry to re-adopt the current committed root).
    Tree(TreeError),
    /// A stored TTL/stamp envelope could not be decoded.
    Ttl(TtlDecodeError),
    /// A shard id at or above the database's fixed `shard_count`.
    ShardOutOfRange { shard_id: usize, shard_count: usize },
}

impl std::fmt::Display for ObserverError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Config(error) => write!(formatter, "observer open failed: {error}"),
            Self::Store(error) => write!(formatter, "observer store error: {error}"),
            Self::Wal(error) => write!(formatter, "observer wal error: {error}"),
            Self::Tree(error) => write!(formatter, "observer tree error: {error}"),
            Self::Ttl(error) => write!(formatter, "observer ttl decode error: {error}"),
            Self::ShardOutOfRange {
                shard_id,
                shard_count,
            } => write!(
                formatter,
                "shard id {shard_id} out of range for shard_count {shard_count}"
            ),
        }
    }
}

impl std::error::Error for ObserverError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Config(error) => Some(error),
            Self::Store(error) => Some(error),
            Self::Wal(error) => Some(error),
            Self::Tree(error) => Some(error),
            Self::Ttl(error) => Some(error),
            Self::ShardOutOfRange { .. } => None,
        }
    }
}

impl From<StoreError> for ObserverError {
    fn from(error: StoreError) -> Self {
        Self::Store(error)
    }
}

impl From<WalError> for ObserverError {
    fn from(error: WalError) -> Self {
        Self::Wal(error)
    }
}

impl From<TreeError> for ObserverError {
    fn from(error: TreeError) -> Self {
        Self::Tree(error)
    }
}

impl From<TtlDecodeError> for ObserverError {
    fn from(error: TtlDecodeError) -> Self {
        Self::Ttl(error)
    }
}