armdb 0.14.1

use crate::{
    FormatMigrationPolicy,
    error::{DbError, DbResult},
};
use std::time::Duration;

/// Default for [`Config::compaction_max_files_per_pass`]: a single compaction
/// pass retires at most this many immutable files per shard. Bounds per-pass
/// work; raise it (or set 0 = unlimited) for continuous overwrite workloads.
pub(crate) const DEFAULT_COMPACTION_MAX_FILES_PER_PASS: usize = 4;

#[cfg(any(feature = "armour", feature = "postcard-codec"))]
fn default_compaction_max_files_per_pass() -> usize {
    DEFAULT_COMPACTION_MAX_FILES_PER_PASS
}

#[cfg(any(feature = "armour", feature = "postcard-codec"))]
fn default_max_projected_rows() -> usize {
    crate::projection::DEFAULT_MAX_PROJECTED_ROWS
}

#[cfg(any(feature = "armour", feature = "postcard-codec"))]
fn default_max_projected_bytes() -> usize {
    crate::projection::DEFAULT_MAX_PROJECTED_BYTES
}

/// Bitcask flush-path I/O strategy. Tunable; in-memory effect only — the
/// on-disk format is identical regardless. Default: `Pwrite`.
///
/// `Uring` requires BOTH the non-default `io-uring` Cargo feature AND Linux; if
/// either is missing it falls back to blocking `pwrite` (logged once at open).
/// io_uring is off by default because its SQPOLL kernel poller busy-polls a CPU
/// per shard — opt in only when benchmarks on your workload justify it. See
/// `docs/research/sqpoll-noise.md`.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
#[cfg_attr(
    any(feature = "armour", feature = "postcard-codec"),
    derive(serde::Deserialize)
)]
pub enum IoBackend {
    /// io_uring writer. `sqpoll_idle_ms = Some(ms)` spawns a dedicated SQPOLL
    /// kernel poller (`iou-sqp-*`) with that idle timeout in milliseconds;
    /// `None` uses io_uring without a poller (each submit makes an
    /// `io_uring_enter` syscall — no kernel thread).
    Uring { sqpoll_idle_ms: Option<u32> },
    /// Blocking `pwrite` via `FileExt::write_all_at` — no io_uring, no poller.
    #[default]
    Pwrite,
}

/// Database configuration.
///
/// # Immutable vs tunable parameters
///
/// **Immutable** (fixed at creation, changing breaks the database):
/// - `shard_count` — stored in `db.meta`, determines key→shard routing
/// - `shard_prefix_bits` — determines key→shard grouping; changing breaks
///   `atomic` guarantees and dead_bytes tracking across shards
/// - `encryption_key` — presence/absence stored in `db.meta`
///
/// **Tunable** (safe to change between restarts):
/// - `max_file_size`, `compaction_threshold`, `enable_fsync`, `write_buffer_size`,
///   `sync_batch_size`, `sync_interval`
/// - `cache` — in-memory only, no on-disk state
/// - `reversed` — only affects in-memory index ordering
/// - `hints` — controls hint file generation/use during recovery
///
/// # Compatibility
///
/// `sync_batch_size` and `sync_interval` are public fields. Existing external
/// exhaustive `Config { ... }` literals must add them (or switch to
/// `ConfigBuilder`). Serialized configurations from before these fields existed
/// remain readable because serde supplies `0` and `None`, respectively.
#[derive(Debug, Clone)]
#[cfg_attr(
    any(feature = "armour", feature = "postcard-codec"),
    derive(serde::Deserialize)
)]
pub struct Config {
    /// Controls whether pending value-format migrations run automatically.
    #[cfg_attr(any(feature = "armour", feature = "postcard-codec"), serde(default))]
    pub format_migration: FormatMigrationPolicy,
    /// Number of shards. **Immutable** — stored in `db.meta`, changing requires
    /// recreating the database. Default: `available_parallelism()`.
    pub shard_count: usize,
    /// Maximum size of a single data file before rotation. Tunable.
    /// Default: 256 MB.
    pub max_file_size: u64,
    /// Ratio of dead_bytes / total_bytes required to trigger compaction. Tunable.
    /// Default: 0.3.
    pub compaction_threshold: f64,
    /// Maximum number of immutable files a single compaction pass retires per
    /// shard. Bounds per-pass work (scan I/O + the temporary disk used by the
    /// rewritten output) so one `compact()` call can't stall on a huge rewrite.
    /// **A low value throttles reclaim under continuous overwrite churn**: if the
    /// writer creates files faster than `cap` files are retired per pass, on-disk
    /// size tracks total bytes written rather than the live set. For such
    /// workloads either raise this, set it to `0` (unlimited — drain the whole
    /// backlog in one pass), or drive `compact()` to convergence each cycle.
    /// Tunable. Default: 4.
    #[cfg_attr(
        any(feature = "armour", feature = "postcard-codec"),
        serde(default = "default_compaction_max_files_per_pass")
    )]
    pub compaction_max_files_per_pass: usize,
    /// Ignored on the Bitcask backend (`Config` configures Bitcask only).
    /// Power-loss durability is the `Db` flush tick, see
    /// [`crate::armour::WorkerIntervals::flush`]; the per-write fsync of
    /// FixedStore is [`crate::fixed::config::FixedConfig::enable_fsync`], a
    /// different struct. Tunable. Default: false.
    pub enable_fsync: bool,
    /// Write buffer size per shard in bytes. Tunable. Writes are buffered in
    /// memory and flushed to disk when the buffer is full or on explicit
    /// flush/close. Default: 1 MB.
    pub write_buffer_size: usize,
    /// Number of writes since the last successful durable barrier that trigger
    /// a write-return sync. `0` disables the batch threshold. Together with
    /// [`Self::sync_interval`], `0` and `None` opt out of write-triggered
    /// barriers. Tunable. Default: 0.
    #[cfg_attr(any(feature = "armour", feature = "postcard-codec"), serde(default))]
    pub sync_batch_size: u32,
    /// Maximum time since the last successful durable barrier before the next
    /// returning write triggers a sync when writes are pending. Therefore the
    /// first write after a long idle period may sync immediately. Time is checked
    /// only as a write returns; it does not start a timer or wake an idle collection.
    /// `None` disables the interval threshold.
    /// [`crate::armour::WorkerIntervals::flush`] is the silence deadline for
    /// pending writes. [`crate::armour::Db::close`] is the explicit final
    /// power-loss barrier for dirty collection data. Tunable. Default: None.
    #[cfg_attr(any(feature = "armour", feature = "postcard-codec"), serde(default))]
    pub sync_interval: Option<Duration>,
    /// Block cache: caches 4 KiB disk blocks for Var values ≤ 8 KiB (dense
    /// small values). Tunable — in-memory only.
    #[cfg(feature = "var-collections")]
    pub block_cache: CacheConfig,
    /// Value cache: caches whole Var values > 8 KiB. Tunable — in-memory only.
    #[cfg(feature = "var-collections")]
    pub value_cache: CacheConfig,
    /// Decoded value cache: caches decoded `Arc<T>` values for `VarTypedTree` /
    /// `VarTypedMap`, keyed by the value's physical location. **Off (0) in every
    /// preset** — opt in per collection. Tunable — in-memory only.
    ///
    /// Read only by the VarTyped wrappers; raw `VarTree` / `VarMap` ignore it.
    /// The budget counts **encoded** bytes: the live heap behind a decoded `T`
    /// is typically 1.5–3× that (`String`/`Vec` capacity, padding, `Arc`
    /// overhead), so size it accordingly. Enabling this together with
    /// `value_cache` holds every value-cache-routed value twice.
    ///
    /// **Encrypted collections:** decryption happens below this cache, so what it
    /// keeps resident is **plaintext** `T` — for as long as the budget holds the
    /// entry, not just for the duration of a read. That is the exposure
    /// `TypedTree` has always had, but it is new for a Var collection with
    /// `encryption_key` set. Leave this at 0 there if plaintext residency is
    /// unacceptable.
    #[cfg(feature = "var-collections")]
    #[cfg_attr(any(feature = "armour", feature = "postcard-codec"), serde(default))]
    pub decoded_cache: CacheConfig,
    /// Number of prefix bits of the key to use for shard routing. **Immutable** —
    /// changing breaks `atomic` guarantees (keys regroup across shards) and
    /// dead_bytes tracking (old entries attributed to wrong shard).
    /// 0 = hash full key (default). Non-zero = hash first N bits for locality.
    pub shard_prefix_bits: usize,
    /// Reverse key ordering in the index. Tunable — only affects in-memory
    /// index ordering. When `true`, forward iteration yields keys in
    /// descending order (newest first for monotonic IDs). Default: true.
    /// Also governs iteration order of iterable Map collections (`iter_view`).
    pub reversed: bool,
    /// Build the in-memory companion key index that enables ordered iteration
    /// (`iter`/`range`/`paginate`/`retain` via `iter_view()`). Tunable,
    /// **in-memory only** — not stored in `db.meta`, so it can be toggled freely
    /// across reopens. Costs extra memory (a per-shard BTreeSet of keys) and
    /// O(log N) per write. Default: false.
    ///
    /// Opt in only for **occasional admin/maintenance** scans (browsing a
    /// collection, bulk-cleaning stale rows) — not for hot-path reads. A Map's
    /// strength is O(1) point lookup; if scans are part of the normal workload,
    /// use a `*Tree` collection instead. Map and Tree scans are both weakly
    /// consistent; the Map implementation releases shard locks between batches, so
    /// concurrent writes may be observed partially.
    #[cfg_attr(any(feature = "armour", feature = "postcard-codec"), serde(default))]
    pub iterable: bool,
    /// Generate and use hint files for faster recovery. Tunable.
    /// `None` ⇒ the collection `open()` chooses by type (Var → true,
    /// inline → false). `Some(x)` is respected as an explicit override.
    /// Default per workload preset: `None`.
    #[cfg_attr(any(feature = "armour", feature = "postcard-codec"), serde(default))]
    pub hints: Option<bool>,
    /// Use `O_DIRECT` on Linux for the hot serving path (active-file write +
    /// live read fds). Tunable — in-memory effect only; the on-disk format is
    /// identical regardless. Falls back to buffered if the filesystem rejects
    /// `O_DIRECT`. No effect off Linux. Default: false.
    pub direct_io: bool,
    /// Bitcask flush-path I/O strategy. Tunable. Default: `Pwrite`.
    /// See [`IoBackend`].
    #[cfg_attr(any(feature = "armour", feature = "postcard-codec"), serde(default))]
    pub io_backend: IoBackend,
    /// Maximum projected index rows per `project` evaluation. Tunable. Default: 4096.
    #[cfg_attr(
        any(feature = "armour", feature = "postcard-codec"),
        serde(default = "default_max_projected_rows")
    )]
    pub max_projected_rows: usize,
    /// Maximum encoded projection payload bytes per `project` evaluation. Tunable. Default: 1 MiB.
    #[cfg_attr(
        any(feature = "armour", feature = "postcard-codec"),
        serde(default = "default_max_projected_bytes")
    )]
    pub max_projected_bytes: usize,
    /// Master AES-256 key. Each shard encrypts under its own subkey derived via
    /// `HKDF-SHA256(master, salt = db.salt, info = [shard_id])`, so multi-shard
    /// encryption and reusing one key across collections are both safe (no
    /// `(key, nonce)` reuse). The per-database salt lives in the `db.salt` sidecar.
    #[cfg(feature = "encryption")]
    pub encryption_key: Option<[u8; 32]>,
}

/// Configuration for the value cache used by `VarTree`.
#[cfg(feature = "var-collections")]
#[derive(Debug, Clone)]
#[cfg_attr(
    any(feature = "armour", feature = "postcard-codec"),
    derive(serde::Deserialize)
)]
pub struct CacheConfig {
    /// Maximum cache size in bytes. 0 = disabled. Default: 0.
    pub max_size: u64,
    /// Estimated number of items (for hash table pre-allocation). Default: 100_000.
    pub estimated_items: usize,
}

impl Config {
    /// Config with small shard count for tests. Keeps fd usage low
    /// so `cargo test` doesn't hit "Too many open files" on default ulimit.
    pub fn test() -> Self {
        Config::balanced().shard_count(2).hints(true).build()
    }

    /// Test config with `hints: None` (lets `open()` resolve by type).
    #[doc(hidden)]
    pub fn test_no_hints() -> Self {
        let mut cfg = Self::test();
        cfg.hints = None;
        cfg
    }

    /// Resolve `hints` against the collection type's default: `None` becomes
    /// `Some(default)`, an explicit `Some(_)` is left untouched. Called by each
    /// leaf collection `open()` (Var → `true`, inline → `false`).
    pub(crate) fn with_resolved_hints(mut self, default: bool) -> Self {
        self.hints = Some(self.hints.unwrap_or(default));
        self
    }

    /// True when a Var collection would open with **no** cache at all
    /// (both tiers disabled) — every `get` then hits disk.
    #[cfg(feature = "var-collections")]
    pub(crate) fn var_cache_warn_needed(&self) -> bool {
        self.block_cache.max_size == 0 && self.value_cache.max_size == 0
    }

    /// True when an app cache is enabled but `direct_io` is off — values are then
    /// cached twice (app cache + OS page cache).
    #[cfg(feature = "var-collections")]
    pub(crate) fn app_cache_without_direct_io(&self) -> bool {
        (self.block_cache.max_size > 0 || self.value_cache.max_size > 0) && !self.direct_io
    }

    /// True when the byte value cache **and** the decoded cache are both
    /// enabled: every value that routes to the value cache is then held twice,
    /// once as bytes and once as a decoded `T`.
    #[cfg(feature = "var-collections")]
    pub(crate) fn value_and_decoded_cache_both_on(&self) -> bool {
        self.value_cache.max_size > 0 && self.decoded_cache.max_size > 0
    }

    pub fn validate(&self) -> DbResult<()> {
        if self.shard_count == 0 || self.shard_count > 255 {
            return Err(DbError::Config("shard_count must be between 1 and 255"));
        }
        if self.max_file_size < 4096 {
            return Err(DbError::Config("max_file_size must be at least 4096"));
        }
        if self.write_buffer_size < 4096 {
            return Err(DbError::Config("write_buffer_size must be at least 4096"));
        }
        if self.max_file_size > u32::MAX as u64 {
            return Err(DbError::Config(
                "max_file_size must not exceed u32::MAX (4 GiB)",
            ));
        }
        if (self.write_buffer_size as u64) > (u32::MAX as u64) - 4096 {
            return Err(DbError::Config(
                "write_buffer_size must not exceed u32::MAX - 4096",
            ));
        }
        if (self.write_buffer_size as u64) > self.max_file_size {
            return Err(DbError::Config(
                "max_file_size must be >= write_buffer_size",
            ));
        }
        if self.shard_prefix_bits > u8::MAX as usize {
            return Err(DbError::Config("shard_prefix_bits must be <= 255"));
        }
        if self.max_projected_rows == 0 {
            return Err(DbError::Config("max_projected_rows must be > 0"));
        }
        if self.max_projected_bytes == 0 {
            return Err(DbError::Config("max_projected_bytes must be > 0"));
        }
        if !self.compaction_threshold.is_finite()
            || !(0.0..=1.0).contains(&self.compaction_threshold)
        {
            return Err(DbError::Config(
                "compaction_threshold must be a finite value in [0.0, 1.0]",
            ));
        }
        let page_aligned = {
            #[cfg(feature = "encryption")]
            {
                self.direct_io || self.encryption_key.is_some()
            }
            #[cfg(not(feature = "encryption"))]
            {
                self.direct_io
            }
        };
        if page_aligned {
            if self.write_buffer_size < 8192 {
                return Err(DbError::Config(
                    "write_buffer_size must be at least 8192 when direct_io or encryption is enabled",
                ));
            }
            if !self.write_buffer_size.is_multiple_of(4096) {
                return Err(DbError::Config(
                    "write_buffer_size must be a multiple of 4096 when direct_io or encryption is enabled",
                ));
            }
        }
        Ok(())
    }
}

#[cfg(feature = "var-collections")]
impl Default for CacheConfig {
    fn default() -> Self {
        Self {
            max_size: 0,
            estimated_items: 100_000,
        }
    }
}

/// Bitcask deployment-intent presets for [`Config`]. Pick one as the builder
/// start point: `Config::for_workload(Workload::…)` or a sugar method.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Workload {
    /// Balanced middle-ground (closest to the former `Default`).
    Balanced,
    /// Embedded / low-latency: few shards, small files & buffers.
    LowLatency,
    /// High-throughput server: many shards, large files & buffers.
    HighThroughput,
    /// Durability-first: `sync_batch_size = 1024`,
    /// `sync_interval = Some(50 ms)`, a small `write_buffer_size`, and the
    /// `Db` flush tick as the deadline while the collection is silent. Use
    /// `.sync_batch_size(0).sync_interval(None)` on its builder to opt out of
    /// both write-triggered thresholds.
    Durable,
    /// Large disk-resident values (Var collections, big blobs): large files,
    /// `O_DIRECT`, and a non-zero value-cache baseline.
    LargeValues,
    /// Var, many small values, working set > RAM (e.g. messages): block cache +
    /// `O_DIRECT`.
    SmallValues,
}

pub(crate) fn default_shard_count() -> usize {
    #[cfg(debug_assertions)]
    {
        std::thread::available_parallelism()
            .map(|p| p.get() / 2)
            .unwrap_or(2)
            .clamp(1, 4)
    }
    #[cfg(not(debug_assertions))]
    {
        std::thread::available_parallelism()
            .map(|p| p.get() / 2)
            .unwrap_or(4)
            .clamp(1, 8)
    }
}

impl Workload {
    fn base(self) -> Config {
        let common = Config {
            format_migration: FormatMigrationPolicy::Auto,
            shard_count: default_shard_count(),
            max_file_size: 256 * 1024 * 1024,
            compaction_threshold: 0.30,
            compaction_max_files_per_pass: DEFAULT_COMPACTION_MAX_FILES_PER_PASS,
            enable_fsync: false,
            write_buffer_size: 1024 * 1024,
            sync_batch_size: 0,
            sync_interval: None,
            #[cfg(feature = "var-collections")]
            block_cache: CacheConfig {
                max_size: 0,
                estimated_items: 100_000,
            },
            #[cfg(feature = "var-collections")]
            value_cache: CacheConfig {
                max_size: 0,
                estimated_items: 100_000,
            },
            #[cfg(feature = "var-collections")]
            decoded_cache: CacheConfig {
                max_size: 0,
                estimated_items: 100_000,
            },
            shard_prefix_bits: 0,
            reversed: true,
            iterable: false,
            hints: None,
            direct_io: false,
            io_backend: IoBackend::Pwrite,
            max_projected_rows: crate::projection::DEFAULT_MAX_PROJECTED_ROWS,
            max_projected_bytes: crate::projection::DEFAULT_MAX_PROJECTED_BYTES,
            #[cfg(feature = "encryption")]
            encryption_key: None,
        };
        match self {
            Workload::Balanced => common,
            Workload::LowLatency => Config {
                shard_count: 4,
                max_file_size: 64 * 1024 * 1024,
                write_buffer_size: 256 * 1024,
                ..common
            },
            Workload::HighThroughput => Config {
                shard_count: 64,
                max_file_size: 512 * 1024 * 1024,
                write_buffer_size: 4 * 1024 * 1024,
                compaction_threshold: 0.25,
                ..common
            },
            Workload::Durable => Config {
                shard_count: 16,
                max_file_size: 128 * 1024 * 1024,
                write_buffer_size: 64 * 1024,
                sync_batch_size: 1024,
                sync_interval: Some(Duration::from_millis(50)),
                ..common
            },
            Workload::SmallValues => Config {
                direct_io: true,
                #[cfg(feature = "var-collections")]
                block_cache: CacheConfig {
                    max_size: 512 << 20,
                    estimated_items: 131_072,
                },
                #[cfg(feature = "var-collections")]
                value_cache: CacheConfig {
                    max_size: 0,
                    estimated_items: 100_000,
                },
                ..common
            },
            Workload::LargeValues => Config {
                max_file_size: 1 << 30, // 1GB
                write_buffer_size: 4 * 1024 * 1024,
                direct_io: true,
                #[cfg(feature = "var-collections")]
                block_cache: CacheConfig {
                    max_size: 0,
                    estimated_items: 100_000,
                },
                #[cfg(feature = "var-collections")]
                value_cache: CacheConfig {
                    max_size: 512 << 20,
                    estimated_items: 8192,
                },
                ..common
            },
        }
    }
}

#[bon::bon]
impl Config {
    /// Build a `Config` from a [`Workload`] preset; any field is overridable via
    /// the chained setters, unset fields keep the preset value.
    #[builder(builder_type = ConfigBuilder, finish_fn = build)]
    pub fn for_workload(
        #[builder(start_fn)] workload: Workload,
        format_migration: Option<FormatMigrationPolicy>,
        shard_count: Option<usize>,
        max_file_size: Option<u64>,
        write_buffer_size: Option<usize>,
        /// Number of writes since the last successful barrier that makes the
        /// returning write issue a barrier. `0` disables this threshold.
        sync_batch_size: Option<u32>,
        /// Time since the last successful barrier that makes a returning write
        /// issue a barrier while writes are pending. The first write after idle
        /// may therefore sync immediately. `None` disables the threshold.
        sync_interval: Option<Option<Duration>>,
        compaction_threshold: Option<f64>,
        compaction_max_files_per_pass: Option<usize>,
        enable_fsync: Option<bool>,
        direct_io: Option<bool>,
        io_backend: Option<IoBackend>,
        reversed: Option<bool>,
        iterable: Option<bool>,
        shard_prefix_bits: Option<usize>,
        /// `None` ⇒ collection `open()` chooses by type.
        hints: Option<bool>,
        #[cfg(feature = "var-collections")] block_cache: Option<CacheConfig>,
        #[cfg(feature = "var-collections")] value_cache: Option<CacheConfig>,
        #[cfg(feature = "var-collections")] decoded_cache: Option<CacheConfig>,
        #[cfg(feature = "encryption")] encryption_key: Option<[u8; 32]>,
        max_projected_rows: Option<usize>,
        max_projected_bytes: Option<usize>,
    ) -> Config {
        let b = workload.base();
        Config {
            format_migration: format_migration.unwrap_or(b.format_migration),
            shard_count: shard_count.unwrap_or(b.shard_count),
            max_file_size: max_file_size.unwrap_or(b.max_file_size),
            write_buffer_size: write_buffer_size.unwrap_or(b.write_buffer_size),
            sync_batch_size: sync_batch_size.unwrap_or(b.sync_batch_size),
            sync_interval: sync_interval.unwrap_or(b.sync_interval),
            compaction_threshold: compaction_threshold.unwrap_or(b.compaction_threshold),
            compaction_max_files_per_pass: compaction_max_files_per_pass
                .unwrap_or(b.compaction_max_files_per_pass),
            enable_fsync: enable_fsync.unwrap_or(b.enable_fsync),
            direct_io: direct_io.unwrap_or(b.direct_io),
            io_backend: io_backend.unwrap_or(b.io_backend),
            reversed: reversed.unwrap_or(b.reversed),
            iterable: iterable.unwrap_or(b.iterable),
            shard_prefix_bits: shard_prefix_bits.unwrap_or(b.shard_prefix_bits),
            hints,
            #[cfg(feature = "var-collections")]
            block_cache: block_cache.unwrap_or(b.block_cache),
            #[cfg(feature = "var-collections")]
            value_cache: value_cache.unwrap_or(b.value_cache),
            #[cfg(feature = "var-collections")]
            decoded_cache: decoded_cache.unwrap_or(b.decoded_cache),
            #[cfg(feature = "encryption")]
            encryption_key: encryption_key.or(b.encryption_key),
            max_projected_rows: max_projected_rows.unwrap_or(b.max_projected_rows),
            max_projected_bytes: max_projected_bytes.unwrap_or(b.max_projected_bytes),
        }
    }
}

impl Config {
    /// Builder starting from the [`Workload::Balanced`] preset.
    pub fn balanced() -> ConfigBuilder {
        Self::for_workload(Workload::Balanced)
    }
    /// Builder starting from the [`Workload::LowLatency`] preset.
    pub fn low_latency() -> ConfigBuilder {
        Self::for_workload(Workload::LowLatency)
    }
    /// Builder starting from the [`Workload::HighThroughput`] preset.
    pub fn high_throughput() -> ConfigBuilder {
        Self::for_workload(Workload::HighThroughput)
    }
    /// Builder starting from the [`Workload::Durable`] preset.
    pub fn durable() -> ConfigBuilder {
        Self::for_workload(Workload::Durable)
    }
    /// Builder starting from the [`Workload::LargeValues`] preset.
    pub fn large_values() -> ConfigBuilder {
        Self::for_workload(Workload::LargeValues)
    }
    /// Builder starting from the [`Workload::SmallValues`] preset.
    pub fn small_values() -> ConfigBuilder {
        Self::for_workload(Workload::SmallValues)
    }
}

#[cfg(test)]
mod io_backend_tests {
    use super::*;

    #[test]
    fn default_io_backend_is_pwrite() {
        let cfg = Config::balanced().build();
        assert_eq!(cfg.io_backend, IoBackend::Pwrite);
    }

    #[test]
    fn io_backend_default_impl_matches_config_default() {
        assert_eq!(IoBackend::default(), IoBackend::Pwrite);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;

    fn default_for_test() -> Config {
        Config {
            format_migration: FormatMigrationPolicy::Auto,
            shard_count: 2,
            max_file_size: 256 * 1024 * 1024,
            compaction_threshold: 0.3,
            compaction_max_files_per_pass: DEFAULT_COMPACTION_MAX_FILES_PER_PASS,
            enable_fsync: false,
            write_buffer_size: 1024 * 1024,
            sync_batch_size: 0,
            sync_interval: None,
            #[cfg(feature = "var-collections")]
            block_cache: CacheConfig::default(),
            #[cfg(feature = "var-collections")]
            value_cache: CacheConfig::default(),
            #[cfg(feature = "var-collections")]
            decoded_cache: CacheConfig::default(),
            shard_prefix_bits: 0,
            reversed: true,
            iterable: false,
            hints: None,
            direct_io: false,
            io_backend: IoBackend::default(),
            max_projected_rows: 4096,
            max_projected_bytes: 1024 * 1024,
            #[cfg(feature = "encryption")]
            encryption_key: None,
        }
    }

    #[test]
    fn test_default_config_is_valid() {
        let cfg = Config::balanced().build();
        assert!(cfg.shard_count >= 1);
        assert!(cfg.validate().is_ok());
    }

    #[test]
    fn validate_rejects_max_file_size_smaller_than_write_buffer() {
        let mut cfg = default_for_test();
        cfg.max_file_size = 4096;
        cfg.write_buffer_size = 8192;
        assert!(cfg.validate().is_err());
    }

    #[test]
    fn validate_rejects_write_buffer_at_u32_limit() {
        let mut cfg = default_for_test();
        cfg.write_buffer_size = (u32::MAX as usize) - 4095;
        cfg.max_file_size = u32::MAX as u64;
        assert!(cfg.validate().is_err());
    }

    #[cfg(feature = "encryption")]
    #[test]
    fn validate_rejects_small_write_buffer_under_encryption() {
        let mut cfg = default_for_test();
        cfg.encryption_key = Some([0u8; 32]);
        cfg.write_buffer_size = 4096;
        assert!(cfg.validate().is_err());
    }

    #[cfg(feature = "encryption")]
    #[test]
    fn validate_rejects_non_page_multiple_write_buffer_under_encryption() {
        let mut cfg = default_for_test();
        cfg.encryption_key = Some([0u8; 32]);
        cfg.write_buffer_size = 8192 + 100;
        cfg.max_file_size = 256 * 1024 * 1024;
        assert!(cfg.validate().is_err());
    }

    #[cfg(feature = "encryption")]
    #[test]
    fn validate_accepts_page_multiple_8k_under_encryption() {
        let mut cfg = default_for_test();
        cfg.encryption_key = Some([0u8; 32]);
        cfg.write_buffer_size = 8192;
        cfg.max_file_size = 256 * 1024 * 1024;
        assert!(cfg.validate().is_ok());
    }

    #[test]
    fn validate_accepts_plain_4096_write_buffer() {
        let mut cfg = default_for_test();
        cfg.write_buffer_size = 4096;
        cfg.max_file_size = 8192;
        assert!(cfg.validate().is_ok());
    }

    #[test]
    fn validate_rejects_shard_prefix_bits_over_255() {
        let mut cfg = default_for_test();
        cfg.shard_prefix_bits = 256;
        let err = cfg.validate().unwrap_err();
        assert!(
            err.to_string().contains("shard_prefix_bits"),
            "expected shard_prefix_bits error, got: {err}"
        );
    }

    #[test]
    fn validate_rejects_compaction_threshold_nan() {
        let mut cfg = default_for_test();
        cfg.compaction_threshold = f64::NAN;
        assert!(cfg.validate().is_err());
    }

    #[test]
    fn validate_rejects_compaction_threshold_negative() {
        let mut cfg = default_for_test();
        cfg.compaction_threshold = -0.1;
        assert!(cfg.validate().is_err());
    }

    #[test]
    fn validate_rejects_compaction_threshold_over_one() {
        let mut cfg = default_for_test();
        cfg.compaction_threshold = 1.1;
        assert!(cfg.validate().is_err());
    }

    #[test]
    fn validate_accepts_compaction_threshold_half() {
        let mut cfg = default_for_test();
        cfg.compaction_threshold = 0.5;
        assert!(cfg.validate().is_ok());
    }

    #[test]
    fn validate_accepts_compaction_threshold_boundaries() {
        let mut cfg = default_for_test();
        cfg.compaction_threshold = 0.0;
        assert!(cfg.validate().is_ok());
        cfg.compaction_threshold = 1.0;
        assert!(cfg.validate().is_ok());
    }

    #[test]
    fn direct_io_requires_page_aligned_write_buffer() {
        let mut c = default_for_test();
        c.direct_io = true;
        c.write_buffer_size = 5000; // not a multiple of 4096
        assert!(c.validate().is_err());

        c.write_buffer_size = 8192;
        assert!(c.validate().is_ok());

        c.write_buffer_size = 4096; // < 8192
        assert!(c.validate().is_err());
    }

    #[test]
    fn direct_io_defaults_false() {
        assert!(!Config::balanced().build().direct_io);
    }

    #[test]
    fn with_resolved_hints_fills_none_with_default() {
        let cfg = default_for_test();
        assert_eq!(cfg.clone().with_resolved_hints(true).hints, Some(true));
        assert_eq!(cfg.with_resolved_hints(false).hints, Some(false));
    }

    #[test]
    fn with_resolved_hints_respects_some() {
        let mut cfg = default_for_test();
        cfg.hints = Some(true);
        assert_eq!(cfg.with_resolved_hints(false).hints, Some(true));
    }

    #[cfg(feature = "var-collections")]
    #[test]
    fn var_cache_warn_needed_true_when_zero() {
        let mut cfg = default_for_test();
        cfg.block_cache.max_size = 0;
        cfg.value_cache.max_size = 0;
        assert!(cfg.var_cache_warn_needed());
        cfg.block_cache.max_size = 1 << 20;
        assert!(!cfg.var_cache_warn_needed());
    }

    #[cfg(feature = "var-collections")]
    #[test]
    fn small_values_preset_targets_block_cache() {
        let c = Config::small_values().build();
        assert!(c.direct_io);
        assert_eq!(c.block_cache.max_size, 512 << 20);
        assert_eq!(c.value_cache.max_size, 0);
    }

    #[cfg(feature = "var-collections")]
    #[test]
    fn large_values_preset_targets_value_cache() {
        let c = Config::large_values().build();
        assert!(c.direct_io);
        assert_eq!(c.value_cache.max_size, 512 << 20);
        assert_eq!(c.block_cache.max_size, 0);
    }

    #[cfg(feature = "var-collections")]
    #[test]
    fn cache_warn_only_when_both_zero() {
        let mut cfg = Config::test();
        cfg.block_cache.max_size = 0;
        cfg.value_cache.max_size = 0;
        assert!(cfg.var_cache_warn_needed());
        cfg.block_cache.max_size = 1 << 20;
        assert!(!cfg.var_cache_warn_needed());
        cfg.block_cache.max_size = 0;
        cfg.value_cache.max_size = 1 << 20;
        assert!(!cfg.var_cache_warn_needed());
    }

    #[cfg(feature = "var-collections")]
    #[test]
    fn decoded_cache_is_off_in_every_preset() {
        for cfg in [
            Config::balanced().build(),
            Config::low_latency().build(),
            Config::high_throughput().build(),
            Config::durable().build(),
            Config::small_values().build(),
            Config::large_values().build(),
        ] {
            assert_eq!(
                cfg.decoded_cache.max_size, 0,
                "decoded_cache must be opt-in: enabling it in a preset would \
                 silently double that preset's memory"
            );
        }
    }

    #[cfg(feature = "var-collections")]
    #[test]
    fn decoded_cache_is_settable_through_the_builder() {
        let cfg = Config::large_values()
            .decoded_cache(CacheConfig {
                max_size: 256 << 20,
                estimated_items: 4096,
            })
            .build();
        assert_eq!(cfg.decoded_cache.max_size, 256 << 20);
        assert_eq!(cfg.decoded_cache.estimated_items, 4096);
        // The preset's own value cache is untouched by the new knob.
        assert_eq!(cfg.value_cache.max_size, 512 << 20);
    }

    #[cfg(feature = "var-collections")]
    #[test]
    fn double_cache_warning_predicate() {
        let mut cfg = default_for_test();
        cfg.value_cache.max_size = 1 << 20;
        cfg.decoded_cache.max_size = 1 << 20;
        assert!(cfg.value_and_decoded_cache_both_on());

        cfg.value_cache.max_size = 0;
        assert!(!cfg.value_and_decoded_cache_both_on());

        cfg.value_cache.max_size = 1 << 20;
        cfg.decoded_cache.max_size = 0;
        assert!(!cfg.value_and_decoded_cache_both_on());
    }

    #[cfg(feature = "var-collections")]
    #[test]
    fn app_cache_without_direct_io_predicate() {
        let mut cfg = Config::test(); // direct_io=false by default
        cfg.block_cache.max_size = 0;
        cfg.value_cache.max_size = 0;
        assert!(!cfg.app_cache_without_direct_io(), "no cache -> no warning");
        cfg.block_cache.max_size = 1 << 20;
        assert!(
            cfg.app_cache_without_direct_io(),
            "cache on + buffered -> warn"
        );
        cfg.direct_io = true;
        assert!(
            !cfg.app_cache_without_direct_io(),
            "direct_io on -> no warning"
        );
        // New presets must not trip it:
        assert!(!Config::small_values().build().app_cache_without_direct_io());
        assert!(!Config::large_values().build().app_cache_without_direct_io());
    }

    #[test]
    fn workload_balanced_matches_table() {
        let c = Config::balanced().build();
        assert_eq!(c.max_file_size, 256 * 1024 * 1024);
        assert_eq!(c.write_buffer_size, 1024 * 1024);
        assert_eq!(c.compaction_threshold, 0.30);
        assert!(!c.enable_fsync);
        assert!(!c.direct_io);
        assert_eq!(c.hints, None);
        assert!(c.shard_count >= 1);
        c.validate().unwrap();
    }

    #[test]
    fn write_sync_threshold_presets_and_builder() {
        for cfg in [
            Config::balanced().build(),
            Config::small_values().build(),
            Config::large_values().build(),
            Config::test(),
        ] {
            assert_eq!(cfg.sync_batch_size, 0);
            assert_eq!(cfg.sync_interval, None);
        }

        let durable = Config::durable().build();
        assert_eq!(durable.sync_batch_size, 1024);
        assert_eq!(durable.sync_interval, Some(Duration::from_millis(50)));

        let every_write = Config::balanced()
            .sync_batch_size(0)
            .sync_interval(Some(Duration::ZERO))
            .build();
        assert_eq!(every_write.sync_interval, Some(Duration::ZERO));

        let disabled_interval = Config::durable().sync_interval(None).build();
        assert_eq!(disabled_interval.sync_interval, None);
    }

    #[test]
    fn workload_high_throughput_matches_table() {
        let c = Config::high_throughput().build();
        assert_eq!(c.shard_count, 64);
        assert_eq!(c.max_file_size, 512 * 1024 * 1024);
        assert_eq!(c.write_buffer_size, 4 * 1024 * 1024);
        assert_eq!(c.compaction_threshold, 0.25);
        c.validate().unwrap();
    }

    #[test]
    fn workload_durable_does_not_set_enable_fsync() {
        let c = Config::durable().build();
        assert_eq!(c.shard_count, 16);
        assert_eq!(c.write_buffer_size, 64 * 1024);
        assert!(!c.enable_fsync);
        c.validate().unwrap();
    }

    #[cfg(feature = "var-collections")]
    #[test]
    fn workload_large_values_has_cache_and_direct_io() {
        let c = Config::large_values().build();
        assert_eq!(c.max_file_size, 1 << 30);
        assert_eq!(c.write_buffer_size, 4 * 1024 * 1024);
        assert!(c.direct_io);
        assert_eq!(c.value_cache.max_size, 512 << 20);
        assert_eq!(c.value_cache.estimated_items, 8192);
        assert_eq!(c.block_cache.max_size, 0);
        c.validate().unwrap();
    }

    #[test]
    fn compaction_max_files_per_pass_default_and_override() {
        // Default preserves the historical per-pass cap (4).
        assert_eq!(Config::balanced().build().compaction_max_files_per_pass, 4);
        assert_eq!(Config::test().compaction_max_files_per_pass, 4);
        // Overridable, including 0 = unlimited.
        let c = Config::for_workload(Workload::Balanced)
            .compaction_max_files_per_pass(0)
            .build();
        assert_eq!(c.compaction_max_files_per_pass, 0);
    }

    #[test]
    fn for_workload_overrides_apply() {
        let c = Config::for_workload(Workload::Balanced)
            .shard_count(7)
            .compaction_threshold(0.5)
            .hints(true)
            .build();
        assert_eq!(c.shard_count, 7);
        assert_eq!(c.compaction_threshold, 0.5);
        assert_eq!(c.hints, Some(true));
        assert_eq!(c.max_file_size, 256 * 1024 * 1024);
    }

    #[cfg(feature = "encryption")]
    #[test]
    fn large_values_with_encryption_validates() {
        let c = Config::large_values().encryption_key([0u8; 32]).build();
        c.validate().unwrap();
    }

    #[test]
    fn iterable_defaults_false_and_overridable() {
        assert!(!Config::balanced().build().iterable);
        assert!(Config::balanced().iterable(true).build().iterable);
        assert!(!Config::test().iterable); // test() does not opt in
    }

    /// A config serialized by a release predating the projection budgets must
    /// still load, taking the documented defaults instead of erroring on a
    /// missing field.
    #[cfg(feature = "armour")]
    #[test]
    fn write_sync_legacy_config_defaults_when_absent_from_serialized_config() {
        let caches = if cfg!(feature = "var-collections") {
            r#""block_cache": {"max_size": 0, "estimated_items": 100000},
               "value_cache": {"max_size": 0, "estimated_items": 100000},"#
        } else {
            ""
        };
        let key = if cfg!(feature = "encryption") {
            r#""encryption_key": null,"#
        } else {
            ""
        };
        let json = format!(
            r#"{{
            "shard_count": 4,
            "max_file_size": 268435456,
            "compaction_threshold": 0.3,
            "enable_fsync": false,
            "write_buffer_size": 1048576,
            {caches}
            {key}
            "shard_prefix_bits": 0,
            "reversed": true,
            "direct_io": false,
            "hints": null
        }}"#
        );
        let cfg: Config = serde_json::from_str(&json).expect("legacy config must deserialize");
        assert_eq!(cfg.max_projected_rows, 4096);
        assert_eq!(cfg.max_projected_bytes, 1024 * 1024);
        assert_eq!(cfg.sync_batch_size, 0);
        assert_eq!(cfg.sync_interval, None);
        cfg.validate().expect("defaults must be valid");
    }
}