haematite 0.6.1

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

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

use super::{CONFIG_FILE, DatabaseError};

/// Current on-disk format version stamped into `config.json` on create (A5).
///
/// Version history:
/// - `1` — the initial stamped format. Also the semantics assigned to LEGACY
///   directories written before the stamp existed (their `config.json` has no
///   `format_version` field); those open unchanged and are never rewritten.
///
/// Compat discipline: `open` accepts exactly {absent, 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 (e.g. value-aware chunking, a pack-file node
/// store) must bump this constant and extend the open-time version check with
/// an explicit arm per older version it still reads — either translated in
/// memory (no rewrite) or via a deliberate migration that rewrites
/// `config.json` atomically only after the data migration is durable.
pub const ON_DISK_FORMAT_VERSION: u32 = 1;

/// 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>,
}

/// 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. The stamp is a property
/// of the directory layout the engine writes — never a caller decision — so it
/// deliberately does NOT live on the public [`DatabaseConfig`].
#[derive(serde::Serialize)]
struct StampedConfig<'config> {
    format_version: u32,
    #[serde(flatten)]
    config: &'config DatabaseConfig,
}

/// Read-side on-disk envelope. `format_version` is `Option` because LEGACY
/// directories (written before A5) carry no stamp; see
/// [`validate_format_version`] for the accept/refuse policy.
#[derive(serde::Deserialize)]
struct StoredConfig {
    format_version: Option<u32>,
    #[serde(flatten)]
    config: DatabaseConfig,
}

pub(super) fn write_config(config: &DatabaseConfig) -> Result<(), DatabaseError> {
    let stamped = StampedConfig {
        format_version: ON_DISK_FORMAT_VERSION,
        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<DatabaseConfig, 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()))?;
    validate_format_version(stored.format_version)?;
    Ok(stored.config)
}

/// 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, [`ON_DISK_FORMAT_VERSION`]} (A5 compat discipline).
///
/// - 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.
/// - 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(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 \
             {ON_DISK_FORMAT_VERSION}; unstamped legacy reads as 1; 0 was never issued — \
             stamping begins at 1)"
        ))),
    }
}

/// 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_distributed_config(config)?;
    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(())
}