haematite 0.7.0

use std::fs;
use std::io::{self, Write};
use std::path::{Path, PathBuf};

use crate::sync::topology::{SyncNodeId, SyncTopology};
use crate::tree::TreePolicy;

use super::{CONFIG_FILE, DatabaseError};

/// Current on-disk format version stamped into `config.json` on create (A5).
///
/// Version history:
/// - `1` — the initial stamped format (count-target v1 chunking). Also the
///   semantics assigned to LEGACY directories written before the stamp existed
///   (their `config.json` has no `format_version` field); those open unchanged,
///   are never rewritten, and run under the v1 detector FOREVER
///   (CHUNKING-POLICY.md §4.1).
/// - `2` — byte-aware chunking (`chunking: "bytes-v2"`). `Database::create`
///   stamps this natively alongside the engine-owned [`ChunkingStamp`] policy
///   block; existing v1/unstamped directories are converted only by an explicit
///   `migrate_chunking` (§4.2), never silently at open.
///
/// Compat discipline: `open` accepts exactly {absent, 1, current}. A stamp NEWER
/// than this constant refuses with [`DatabaseError::FormatVersionTooNew`] (an
/// old binary must fail loudly on a newer directory, never misread it). Any
/// future format-changing work must bump this constant and extend the open-time
/// version check with an explicit arm per older version it still reads.
pub const ON_DISK_FORMAT_VERSION: u32 = 2;

/// The v1 on-disk format version — count-target chunking. Read-only from here
/// on: v1 (and unstamped) directories open under [`TreePolicy::V1_DEFAULT`]
/// forever; only `migrate_chunking` ever rewrites them.
const V1_FORMAT_VERSION: u32 = 1;

/// The policy identity string stamped for byte-aware (format 2) chunking.
const CHUNKING_V2_ID: &str = "bytes-v2";

/// Ratified v2 leaf target (`leaf_target_bytes`, 64 KiB — CHUNKING-POLICY.md
/// §2.3, Tom-signed 2026-07-22). ENGINE-OWNED: stamped at create/migrate,
/// never a [`DatabaseConfig`] field, never defaulted at open.
pub const DEFAULT_LEAF_TARGET_BYTES: u64 = 64 * 1024;

/// Ratified v2 internal target (`internal_target_bytes`, 48 KiB — §2.3).
pub const DEFAULT_INTERNAL_TARGET_BYTES: u64 = 48 * 1024;

/// The `haem` subcommand a fenced (mid-migration) directory names as its resume
/// path in [`DatabaseError::MigrationInProgress`].
const MIGRATE_RESUME_COMMAND: &str = "haem migrate-chunking";

/// The engine-owned v2 chunking policy block written beside the `format_version`
/// stamp (§4.1). It is a property of the directory's on-disk layout — a
/// deployment contract set at create/migrate time — so it deliberately does NOT
/// live on the public [`DatabaseConfig`]; the stamp is the sole source of truth
/// (rule 2), never a value defaulted at open.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
struct ChunkingStamp {
    /// Policy identity; the only accepted value is [`CHUNKING_V2_ID`].
    chunking: String,
    leaf_target_bytes: u64,
    internal_target_bytes: u64,
}

impl ChunkingStamp {
    /// The default v2 stamp `Database::create` writes: the ratified constants.
    fn default_v2() -> Self {
        Self {
            chunking: CHUNKING_V2_ID.to_owned(),
            leaf_target_bytes: DEFAULT_LEAF_TARGET_BYTES,
            internal_target_bytes: DEFAULT_INTERNAL_TARGET_BYTES,
        }
    }
}

/// The structured migration fence (§4.2, step 2): present in `config.json` only
/// while a `migrate_chunking` transaction is in flight (or paused after a crash).
/// Its mere presence fences normal opens with
/// [`DatabaseError::MigrationInProgress`] on both the writer and observer paths;
/// only `migrate_chunking` interprets and clears it. Read here so Phase-2 opens
/// refuse a fenced directory; the write/resume machinery lands with the
/// migration transaction.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
struct MigrationFence {
    source_policy: String,
    target_policy: String,
    phase: String,
}

/// The engine-derived open result: the caller's [`DatabaseConfig`] plus the
/// [`TreePolicy`] the on-disk stamp resolves to. The policy is threaded from
/// here through shard spawn so the mutation path uses the stamped rule, never a
/// literal (CHUNKING-POLICY.md §4.1).
pub(super) struct OpenedConfig {
    pub config: DatabaseConfig,
    pub policy: TreePolicy,
}

/// The [`TreePolicy`] a freshly created (format-2) database mutates under — the
/// ratified byte-aware targets. `create` stamps these and spawns its shards
/// under the SAME policy, so a created database is v2 from birth (§5).
pub(super) const fn default_create_policy() -> TreePolicy {
    TreePolicy::v2(DEFAULT_LEAF_TARGET_BYTES, DEFAULT_INTERNAL_TARGET_BYTES)
}

/// Explicit database configuration; no field has a silent default.
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
pub struct DatabaseConfig {
    pub data_dir: PathBuf,
    pub shard_count: usize,
    /// Distributed sync configuration. `None` keeps the database single-node.
    pub distributed: Option<DistributedDatabaseConfig>,
    /// Worker count for global-commit and sequence-scan fan-out (COMMIT-COLLAPSE
    /// §6). `None` = the signed default `min(shard_count, available_parallelism)`;
    /// `Some(0)` is rejected at validation. `serde(default)` so every existing
    /// `config.json` (which predates this field) parses unchanged to `None`.
    ///
    /// Compatibility direction (§9.5): new-binary-reads-old is clean (the
    /// default); OLD-binary-reads-new is SILENT — a downgraded 0.4.x binary
    /// permissively ignores this field and reverts to thread-per-item fan-out
    /// with no refusal. It is therefore best-effort under downgrade — a resource
    /// tuning, not a safety property.
    #[serde(default)]
    pub executor_threads: Option<usize>,
}

/// Explicit distributed database configuration; no topology is implied.
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
pub struct DistributedDatabaseConfig {
    pub local_node: SyncNodeId,
    pub nodes: Vec<SyncNodeId>,
    /// Sync topology chosen by the caller. `None` is rejected for distributed
    /// creation/opening so there is no silent default topology.
    pub topology: Option<SyncTopology>,
    /// Sync interval in milliseconds.
    pub sync_interval: u64,
}

/// Write-side on-disk envelope: the caller's [`DatabaseConfig`] flattened
/// alongside the engine-owned `format_version` stamp and the v2 [`ChunkingStamp`]
/// policy block. Both stamps are properties of the directory layout the engine
/// writes — never a caller decision — so neither lives on the public
/// [`DatabaseConfig`].
#[derive(serde::Serialize)]
struct StampedConfig<'config> {
    format_version: u32,
    chunking_policy: ChunkingStamp,
    #[serde(flatten)]
    config: &'config DatabaseConfig,
}

/// Read-side on-disk envelope. `format_version` is `Option` because LEGACY
/// directories (written before A5) carry no stamp; `chunking_policy` is `Option`
/// because v1/unstamped directories carry no policy block; `migration_fence` is
/// present only mid-migration. See [`validate_format_version`] /
/// [`derive_policy`] for the accept/refuse policy.
#[derive(serde::Deserialize)]
struct StoredConfig {
    format_version: Option<u32>,
    chunking_policy: Option<ChunkingStamp>,
    migration_fence: Option<MigrationFence>,
    #[serde(flatten)]
    config: DatabaseConfig,
}

pub(super) fn write_config(config: &DatabaseConfig) -> Result<(), DatabaseError> {
    let stamped = StampedConfig {
        format_version: ON_DISK_FORMAT_VERSION,
        chunking_policy: ChunkingStamp::default_v2(),
        config,
    };
    let bytes = serde_json::to_vec_pretty(&stamped).map_err(|error| {
        DatabaseError::ConfigWrite(io::Error::new(io::ErrorKind::InvalidData, error))
    })?;
    install_config_atomic(&config.data_dir.join(CONFIG_FILE), &bytes)
}

/// A5: `create` must never clobber a directory that already holds a database.
///
/// If `config.json` exists, surface the sharpest refusal available: a config
/// stamped by a NEWER binary refuses as
/// [`DatabaseError::FormatVersionTooNew`] (via the [`read_config`] validation)
/// so the operator learns the real problem; a corrupt config surfaces its
/// parse failure; an ordinary same-version config refuses as
/// [`DatabaseError::DataDirAlreadyInitialised`]. Callers hold the exclusive
/// writer lock across this check and the subsequent [`write_config`], so no
/// concurrent writer can initialise the directory between the two.
pub(super) fn ensure_uninitialised(data_dir: &Path) -> Result<(), DatabaseError> {
    let config_path = data_dir.join(CONFIG_FILE);
    if config_path.exists() {
        read_config(data_dir)?;
        return Err(DatabaseError::DataDirAlreadyInitialised { config_path });
    }
    Ok(())
}

pub(super) fn read_config(path: &Path) -> Result<OpenedConfig, DatabaseError> {
    let bytes = fs::read(path.join(CONFIG_FILE)).map_err(DatabaseError::ConfigRead)?;
    let value: serde_json::Value = serde_json::from_slice(&bytes)
        .map_err(|error| DatabaseError::ConfigParse(error.to_string()))?;
    validate_legacy_ttl_cadence(&value)?;
    let stored: StoredConfig = serde_json::from_value(value)
        .map_err(|error| DatabaseError::ConfigParse(error.to_string()))?;
    // §4.2 step 2/8: a fenced directory refuses every NORMAL open (writer AND
    // observer) before any version/policy interpretation — serving a
    // half-migrated shard set as a coherent database is exactly what the fence
    // exists to prevent. Only `migrate_chunking` reads past the fence.
    if let Some(fence) = &stored.migration_fence {
        log::warn!(
            "refusing open of a fenced directory: chunking migration {} -> {} (phase {}); \
             resume with `{MIGRATE_RESUME_COMMAND}`",
            fence.source_policy,
            fence.target_policy,
            fence.phase,
        );
        return Err(DatabaseError::MigrationInProgress {
            resume_command: MIGRATE_RESUME_COMMAND.to_owned(),
        });
    }
    validate_format_version(stored.format_version)?;
    let policy = derive_policy(stored.format_version, stored.chunking_policy.as_ref())?;
    Ok(OpenedConfig {
        config: stored.config,
        policy,
    })
}

/// Resolve the on-disk stamp to the [`TreePolicy`] the mutation path runs under
/// (§4.1). Called AFTER [`validate_format_version`], so only {absent, 1, 2}
/// reach here:
///
/// - absent / `1` — a v1 or legacy directory: [`TreePolicy::V1_DEFAULT`], the
///   byte-for-byte count-target rule, forever. No policy block is expected or
///   consulted.
/// - `2` — byte-aware: the stamped [`ChunkingStamp`] MUST be present, carry the
///   known policy id, and have NONZERO targets. A missing block, unknown id, or
///   zero target is a typed [`DatabaseError::InvalidChunkingStamp`] — the engine
///   refuses to guess a target rather than silently defaulting one (rule 2).
fn derive_policy(
    format_version: Option<u32>,
    stamp: Option<&ChunkingStamp>,
) -> Result<TreePolicy, DatabaseError> {
    match format_version {
        None | Some(V1_FORMAT_VERSION) => Ok(TreePolicy::V1_DEFAULT),
        Some(ON_DISK_FORMAT_VERSION) => {
            let stamp = stamp.ok_or_else(|| {
                DatabaseError::InvalidChunkingStamp(format!(
                    "format_version {ON_DISK_FORMAT_VERSION} requires a chunking_policy block"
                ))
            })?;
            if stamp.chunking != CHUNKING_V2_ID {
                return Err(DatabaseError::InvalidChunkingStamp(format!(
                    "unknown chunking policy id {:?} (expected {CHUNKING_V2_ID:?})",
                    stamp.chunking
                )));
            }
            if stamp.leaf_target_bytes == 0 || stamp.internal_target_bytes == 0 {
                return Err(DatabaseError::InvalidChunkingStamp(format!(
                    "leaf/internal target bytes must be nonzero (got {}/{})",
                    stamp.leaf_target_bytes, stamp.internal_target_bytes
                )));
            }
            Ok(TreePolicy::v2(
                stamp.leaf_target_bytes,
                stamp.internal_target_bytes,
            ))
        }
        // Unreachable after validate_format_version, but never silently default a
        // policy: surface the impossible stamp instead of guessing.
        Some(other) => Err(DatabaseError::ConfigParse(format!(
            "format_version {other} has no chunking-policy mapping"
        ))),
    }
}

/// Reader-only compatibility arm for the retired TTL cadence field. New files
/// never write this key. Legacy null and positive millisecond values are ignored;
/// zero retains the pre-retirement typed refusal rather than blessing malformed
/// stored configuration.
fn validate_legacy_ttl_cadence(value: &serde_json::Value) -> Result<(), DatabaseError> {
    // Plain literal ON PURPOSE: this reader arm is a real (by-design) consumer
    // of the retired key and must stay visible to every future census grep.
    const LEGACY_KEY: &str = "sweep_interval";
    let Some(value) = value.as_object().and_then(|object| object.get(LEGACY_KEY)) else {
        return Ok(());
    };
    match value {
        serde_json::Value::Null => Ok(()),
        serde_json::Value::Number(number) if number.as_u64() == Some(0) => {
            Err(DatabaseError::InvalidSweepInterval)
        }
        serde_json::Value::Number(number) if number.as_u64().is_some() => Ok(()),
        _ => Err(DatabaseError::ConfigParse(
            "legacy TTL cadence must be null or an unsigned integer".to_owned(),
        )),
    }
}

/// Accept exactly {absent, [`V1_FORMAT_VERSION`], [`ON_DISK_FORMAT_VERSION`]}
/// (A5 compat discipline, extended for the v2 bump — §4.1).
///
/// - Absent stamp: a LEGACY pre-stamp directory, accepted with version-1
///   semantics WITHOUT rewriting the file — `open` is a pure read of
///   `config.json`, so the directory stays byte-identical and remains openable
///   by the old binary that wrote it, under the v1 detector forever.
/// - Stamp `1`: an explicit v1 directory — the arm §4.1 promised. Opens and
///   mutates under the v1 detector forever; never force-migrated.
/// - Stamp `2`: the current byte-aware format (its policy block is validated by
///   [`derive_policy`]).
/// - Stamp above the supported version: refuse loudly with the typed
///   [`DatabaseError::FormatVersionTooNew`] naming both versions, so an old
///   binary pointed at a newer directory fails legibly instead of misreading
///   a layout it cannot know.
/// - Stamp `0`: never issued (stamping began at `1`), rejected as a parse
///   failure rather than treated as any real version.
fn validate_format_version(stamp: Option<u32>) -> Result<(), DatabaseError> {
    match stamp {
        None | Some(V1_FORMAT_VERSION | ON_DISK_FORMAT_VERSION) => Ok(()),
        Some(found) if found > ON_DISK_FORMAT_VERSION => Err(DatabaseError::FormatVersionTooNew {
            found,
            supported: ON_DISK_FORMAT_VERSION,
        }),
        Some(found) => Err(DatabaseError::ConfigParse(format!(
            "on-disk format_version {found} has no read arm in this binary (reads \
             {V1_FORMAT_VERSION}..={ON_DISK_FORMAT_VERSION}; unstamped legacy reads as \
             {V1_FORMAT_VERSION}; 0 was never issued — stamping begins at {V1_FORMAT_VERSION})"
        ))),
    }
}

/// Atomically install the serialised config at `path` (house pattern: temp
/// file in the same directory → content fsync → rename → parent-dir fsync).
/// A crash mid-create therefore never leaves a torn or stampless
/// `config.json` that a later open could mistake for a legacy directory.
fn install_config_atomic(path: &Path, bytes: &[u8]) -> Result<(), DatabaseError> {
    let parent = path.parent().ok_or_else(|| {
        DatabaseError::ConfigWrite(io::Error::new(
            io::ErrorKind::InvalidInput,
            "config path has no parent directory",
        ))
    })?;
    let mut temp_file = tempfile::Builder::new()
        .prefix(".config-")
        .suffix(".tmp")
        .tempfile_in(parent)
        .map_err(DatabaseError::ConfigWrite)?;
    temp_file
        .write_all(bytes)
        .map_err(DatabaseError::ConfigWrite)?;
    // tempfile creates its files 0600; installed configs must stay readable by
    // observers/tooling running as other users, matching the pre-A5 behaviour
    // (fs::write yielded umask-derived permissions, typically 0644). The config
    // holds no secrets — paths, shard counts, and node ids only.
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        temp_file
            .as_file()
            .set_permissions(fs::Permissions::from_mode(0o644))
            .map_err(DatabaseError::ConfigWrite)?;
    }
    temp_file
        .as_file_mut()
        .sync_all()
        .map_err(DatabaseError::ConfigWrite)?;
    temp_file
        .persist(path)
        .map(drop)
        .map_err(|error| DatabaseError::ConfigWrite(error.error))?;
    sync_parent_dir(parent)
}

/// Flush the directory entry created by the atomic rename: a synced file with
/// an unsynced directory entry can still vanish on power loss. Unix-only for
/// the same reason as `branch/persist.rs` — `std` cannot open a directory for
/// fsync on Windows, so the directory-entry sync is skipped there.
#[cfg(unix)]
fn sync_parent_dir(parent: &Path) -> Result<(), DatabaseError> {
    fs::File::open(parent)
        .and_then(|dir| dir.sync_all())
        .map_err(DatabaseError::ConfigWrite)
}

#[cfg(not(unix))]
fn sync_parent_dir(_parent: &Path) -> Result<(), DatabaseError> {
    Ok(())
}

pub(super) fn validate_database_config(config: &DatabaseConfig) -> Result<(), DatabaseError> {
    validate_shard_count(config.shard_count)?;
    validate_executor_threads(config.executor_threads)?;
    validate_distributed_config(config)?;
    Ok(())
}

/// `Some(0)` is rejected loudly at validation (COMMIT-COLLAPSE §6): zero workers
/// could never drain a batch. `None` (the signed default) and any positive count
/// are accepted; the default's `available_parallelism()` resolution happens at
/// startup (see [`super::executor::resolve_worker_count`]).
fn validate_executor_threads(executor_threads: Option<usize>) -> Result<(), DatabaseError> {
    if executor_threads == Some(0) {
        return Err(DatabaseError::ExecutorThreadsInvalid(
            "executor_threads = 0 is not a valid worker count; set a positive integer or omit \
             the field for the signed default min(shard_count, available_parallelism)"
                .to_owned(),
        ));
    }
    Ok(())
}

const fn validate_shard_count(shard_count: usize) -> Result<(), DatabaseError> {
    if shard_count == 0 {
        Err(DatabaseError::InvalidShardCount)
    } else {
        Ok(())
    }
}

fn validate_distributed_config(config: &DatabaseConfig) -> Result<(), DatabaseError> {
    let Some(distributed) = &config.distributed else {
        return Ok(());
    };
    let Some(topology) = &distributed.topology else {
        return Err(DatabaseError::MissingSyncTopology);
    };
    if distributed.sync_interval == 0 {
        return Err(DatabaseError::InvalidSyncInterval);
    }
    topology
        .partners_for(&distributed.local_node, &distributed.nodes)
        .map_err(|error| DatabaseError::SyncSchedulerError(error.to_string()))?;
    Ok(())
}

/// On-disk config envelopes for the chunking-v2 migration transaction
/// (§4.2). A PRIVATE child module so the migration's four config states reuse
/// this module's private [`ChunkingStamp`] / [`MigrationFence`] schema and the
/// atomic [`install_config_atomic`] discipline, never a second copy of the
/// layout. Its `pub(in crate::db)` API is re-exported below so `db::migrate` can
/// reach it without the module itself becoming crate- or db-visible (which would
/// trip the pub-crate lints or the source census).
mod migration_io;

pub(in crate::db) use migration_io::{
    FenceView, MigrationState, finalize_v1, finalize_v2, flip_to_abort, install_forward_fence,
    read_state,
};

#[cfg(test)]
#[path = "config_stamp_tests.rs"]
mod stamp_tests;