config.rs

use super::*;

// ===== Manifest source =====

/// Where to load the manifest on connection open.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
pub enum ManifestSource {
    /// Use local manifest if present, fall back to remote. Correct for single-writer.
    /// Checkpoints keep the local cache fresh; no remote fetch on warm reconnect.
    #[default]
    Auto,
    /// Always fetch from the remote backend on open. For HA followers and
    /// multi-reader setups where another process may have checkpointed.
    Remote,
    /// Legacy spelling for remote manifest loading.
    S3,
}

// ===== Checkpoint mode =====

/// How checkpoints interact with the configured storage backend.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
pub enum CheckpointMode {
    /// A SQLite checkpoint uploads dirty page groups to the backend before
    /// returning. This is the default durability mode.
    #[default]
    Durable,
    /// A SQLite checkpoint commits into the local cache/staging log and returns
    /// without backend I/O. Call `flush_to_storage()` to publish the exact
    /// checkpointed image to the backend later.
    LocalThenFlush,
}

// ===== Legacy sync mode =====

/// Legacy spelling for old S3-owned checkpoint modes.
///
/// Prefer [`CheckpointMode`]. Turbolite receives storage as a
/// `hadb_storage::StorageBackend`; this enum remains only to translate old
/// flat config callers while tests and embedders move to the nested config.
#[deprecated(
    since = "0.2.0",
    note = "use TurboliteConfig.cache.checkpoint_mode instead"
)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
// S3Primary is the deprecated default; the derived Default references it on
// purpose for back-compat with existing serialized configs.
#[allow(deprecated)]
pub enum SyncMode {
    #[default]
    S3Primary,
    LocalThenFlush,
}

// ===== Grouping strategy =====

/// Grouping strategy marker. Only BTreeAware is supported.
/// Kept as an enum for serde backward compatibility with existing manifests.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
pub enum GroupingStrategy {
    /// Legacy positional mapping. No longer used for new databases.
    /// Existing Positional manifests are read-only compatible.
    Positional,
    /// B-tree-aware: explicit page-to-group mapping from btree walking.
    #[default]
    BTreeAware,
}

// ===== Group state for prefetch coordination =====

#[repr(u8)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GroupState {
    None = 0,
    Fetching = 1,
    Present = 2,
}

// ===== Nested configuration groups =====
//
// Phase Cirrus b1: these grouped structs are added alongside the flat
// `TurboliteConfig` fields. `Default::default()` populates both in lockstep
// so the nested view is always consistent with the flat view. b2 will
// remove the flat fields and switch internal reads to the nested form.

/// Cache and durability-eviction knobs.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct CacheConfig {
    /// TTL for cached page groups in seconds (0 = no automatic eviction).
    pub ttl_secs: u64,
    /// Maximum cache size in bytes (sub-chunk granularity). None = unlimited.
    pub max_bytes: Option<u64>,
    /// In-memory page cache budget in bytes. 0 disables the mem cache.
    pub mem_budget: u64,
    /// Evict data tier after successful checkpoint upload.
    pub evict_on_checkpoint: bool,
    /// Compress pages in the local disk cache using zstd.
    pub compression: bool,
    /// Zstd compression level for local cache pages (1-22).
    pub compression_level: i32,
    /// Pages per page group. Each backend object contains this many compressed pages.
    pub pages_per_group: u32,
    /// Pages per sub-chunk frame for seekable page groups. 0 disables seekable encoding.
    pub sub_pages_per_frame: u32,
    /// Automatic garbage collection after each checkpoint.
    pub gc_enabled: bool,
    /// Checkpoint/backend durability behavior.
    pub checkpoint_mode: CheckpointMode,
    /// Override threshold. 0 = auto (frames_per_group / 4).
    pub override_threshold: u32,
    /// Compaction threshold.
    pub compaction_threshold: u32,
}

/// Compression settings for remote page groups (distinct from cache compression).
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct CompressionConfig {
    /// Zstd compression level (1-22).
    pub level: i32,
    /// Zstd compression dictionary (2-5x better on structured data).
    #[cfg(feature = "zstd")]
    pub dictionary: Option<Vec<u8>>,
}

/// At-rest encryption settings.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(default)]
pub struct EncryptionConfig {
    /// AES-256-GCM key. When set, all page data is encrypted at rest.
    /// Requires the `encryption` feature.
    pub key: Option<[u8; 32]>,
}

/// Prefetch and query-plan-driven warmup knobs.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct PrefetchConfig {
    /// Prefetch schedule for SEARCH queries.
    pub search: Vec<f32>,
    /// Prefetch schedule for index lookups / point queries.
    pub lookup: Vec<f32>,
    /// Number of prefetch worker threads.
    pub threads: u32,
    /// Optional prefetch queue capacity. This is intentionally independent
    /// from worker count and cloud I/O permits.
    pub queue_capacity: u32,
    /// Total remote I/O requests Turbolite may run concurrently. Defaults to
    /// worker count + 1 so prefetch can saturate all workers while one permit
    /// stays reserved for foreground range gets and orchestration.
    pub io_permits: u32,
    /// Remote I/O permits reserved for foreground demand reads.
    pub foreground_reserved_permits: u32,
    /// Maximum groups each planned SCAN may keep queued/in-flight.
    pub scan_window_groups: u32,
    /// Maximum uncompressed bytes each planned SCAN may keep queued/in-flight.
    pub scan_window_bytes: u64,
    /// Enable query-plan-aware prefetch.
    pub query_plan: bool,
    /// Advisory index-leaf lookahead for SEARCH -> table rowid chasing. On by
    /// default; only engages for indexed SEARCH that chases into a table, so
    /// scans, point-by-rowid reads, and warm queries are untouched.
    pub lookahead: bool,
    /// Enable predictive cross-tree prefetch.
    pub prediction: bool,
    /// Manifest source — Auto (local fallback) vs Remote (always backend).
    pub manifest_source: ManifestSource,
}

/// WAL replication settings (walrust-backed durability between checkpoints).
#[cfg(feature = "wal")]
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct WalConfig {
    /// Ship WAL frames to the backend for transaction-level durability.
    pub replication: bool,
    /// How often walrust ships WAL frames (milliseconds).
    pub sync_interval_ms: u64,
}

impl Default for CacheConfig {
    fn default() -> Self {
        Self {
            ttl_secs: 0,
            max_bytes: None,
            mem_budget: 64 * 1024 * 1024,
            evict_on_checkpoint: false,
            compression: false,
            compression_level: 3,
            pages_per_group: DEFAULT_PAGES_PER_GROUP,
            sub_pages_per_frame: DEFAULT_SUB_PAGES_PER_FRAME,
            gc_enabled: true,
            checkpoint_mode: CheckpointMode::Durable,
            override_threshold: 0,
            compaction_threshold: 8,
        }
    }
}

impl CacheConfig {
    /// Construct a CacheConfig from the `TURBOLITE_*` environment variables,
    /// falling back to `Default` for any value not overridden.
    pub fn from_env() -> Self {
        let d = Self::default();
        Self {
            max_bytes: std::env::var("TURBOLITE_CACHE_LIMIT")
                .ok()
                .and_then(|v| crate::tiered::settings::parse_byte_size(&v))
                .and_then(|n| if n == 0 { None } else { Some(n) })
                .or(d.max_bytes),
            mem_budget: std::env::var("TURBOLITE_MEM_CACHE_BUDGET")
                .ok()
                .and_then(|v| crate::tiered::settings::parse_byte_size(&v))
                .unwrap_or(d.mem_budget),
            evict_on_checkpoint: std::env::var("TURBOLITE_EVICT_ON_CHECKPOINT")
                .map(|v| matches!(v.as_str(), "true" | "1"))
                .unwrap_or(d.evict_on_checkpoint),
            compression: std::env::var("TURBOLITE_CACHE_COMPRESSION")
                .map(|v| matches!(v.as_str(), "true" | "1"))
                .unwrap_or(d.compression),
            compression_level: std::env::var("TURBOLITE_CACHE_COMPRESSION_LEVEL")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(d.compression_level),
            pages_per_group: std::env::var("TURBOLITE_PAGES_PER_GROUP")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(d.pages_per_group),
            sub_pages_per_frame: std::env::var("TURBOLITE_SUB_PAGES_PER_FRAME")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(d.sub_pages_per_frame),
            override_threshold: std::env::var("TURBOLITE_OVERRIDE_THRESHOLD")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(d.override_threshold),
            compaction_threshold: std::env::var("TURBOLITE_COMPACTION_THRESHOLD")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(d.compaction_threshold),
            ..d
        }
    }
}

impl Default for CompressionConfig {
    fn default() -> Self {
        Self {
            level: 3,
            #[cfg(feature = "zstd")]
            dictionary: None,
        }
    }
}

impl CompressionConfig {
    /// Read `TURBOLITE_COMPRESSION_LEVEL` (the loadable-extension historically
    /// used this name). Falls back to `Default` otherwise.
    pub fn from_env() -> Self {
        let d = Self::default();
        Self {
            level: std::env::var("TURBOLITE_COMPRESSION_LEVEL")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(d.level),
            ..d
        }
    }
}

impl Default for PrefetchConfig {
    fn default() -> Self {
        Self {
            search: vec![0.3, 0.3, 0.4],
            lookup: vec![0.0, 0.0, 0.0],
            threads: Self::default_threads(),
            queue_capacity: 8,
            io_permits: Self::default_threads() + 1,
            foreground_reserved_permits: 1,
            scan_window_groups: 4,
            scan_window_bytes: 32 * 1024 * 1024,
            query_plan: true,
            lookahead: true,
            prediction: false,
            manifest_source: ManifestSource::Auto,
        }
    }
}

impl PrefetchConfig {
    /// Default download pool size: leave one core for SQLite/foreground work.
    pub fn default_threads() -> u32 {
        std::thread::available_parallelism()
            .map(|n| n.get().saturating_sub(1).max(1) as u32)
            .unwrap_or(1)
    }

    pub fn from_env() -> Self {
        let d = Self::default();
        Self {
            search: std::env::var("TURBOLITE_PREFETCH_SEARCH")
                .ok()
                .and_then(|v| crate::tiered::settings::parse_hops(&v))
                .unwrap_or(d.search),
            lookup: std::env::var("TURBOLITE_PREFETCH_LOOKUP")
                .ok()
                .and_then(|v| crate::tiered::settings::parse_hops(&v))
                .unwrap_or(d.lookup),
            threads: std::env::var("TURBOLITE_PREFETCH_THREADS")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(d.threads),
            queue_capacity: std::env::var("TURBOLITE_PREFETCH_QUEUE_CAPACITY")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(d.queue_capacity),
            io_permits: std::env::var("TURBOLITE_CLOUD_IO_PERMITS")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(d.io_permits),
            foreground_reserved_permits: std::env::var("TURBOLITE_FOREGROUND_IO_RESERVED")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(d.foreground_reserved_permits),
            scan_window_groups: std::env::var("TURBOLITE_SCAN_WINDOW_GROUPS")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(d.scan_window_groups),
            scan_window_bytes: std::env::var("TURBOLITE_SCAN_WINDOW_BYTES")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(d.scan_window_bytes),
            query_plan: std::env::var("TURBOLITE_PLAN_AWARE")
                .map(|v| !matches!(v.as_str(), "false" | "0"))
                .unwrap_or(d.query_plan),
            lookahead: std::env::var("TURBOLITE_LOOKAHEAD")
                .map(|v| matches!(v.as_str(), "true" | "1"))
                .unwrap_or(d.lookahead),
            prediction: std::env::var("TURBOLITE_PREDICTION")
                .map(|v| matches!(v.as_str(), "true" | "1"))
                .unwrap_or(d.prediction),
            manifest_source: std::env::var("TURBOLITE_MANIFEST_SOURCE")
                .ok()
                .map(|v| match v.to_lowercase().as_str() {
                    "remote" | "s3" => ManifestSource::Remote,
                    _ => ManifestSource::Auto,
                })
                .unwrap_or(d.manifest_source),
        }
    }
}

#[cfg(feature = "wal")]
impl Default for WalConfig {
    fn default() -> Self {
        Self {
            replication: false,
            sync_interval_ms: 1000,
        }
    }
}

#[cfg(feature = "wal")]
impl WalConfig {
    pub fn from_env() -> Self {
        let d = Self::default();
        Self {
            replication: std::env::var("TURBOLITE_WAL_REPLICATION")
                .map(|v| matches!(v.as_str(), "true" | "1"))
                .unwrap_or(d.replication),
            sync_interval_ms: std::env::var("TURBOLITE_WAL_SYNC_INTERVAL")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(d.sync_interval_ms),
        }
    }
}

// ===== Configuration =====

/// Configuration for turbolite storage.
///
/// turbolite is a byte-level storage consumer; the concrete backend (local
/// filesystem, S3, fenced HTTP, etc.) is an embedder's choice. Use
/// [`TurboliteVfs::new`] for local mode (page groups on `cache_dir`) or
/// [`TurboliteVfs::with_backend`] to inject any
/// `Arc<dyn hadb_storage::StorageBackend>` plus a tokio runtime handle.
#[allow(deprecated)]
#[derive(Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct TurboliteConfig {
    /// Local cache/data directory. Always populated; for non-local backends
    /// this is where the sub-chunk cache, staging logs, and dirty-group
    /// recovery state live.
    pub cache_dir: PathBuf,
    /// Optional local main database image path.
    ///
    /// When unset, Turbolite stores the main image at
    /// `{cache_dir}/data.cache`. Product embedders that want a
    /// SQLite-shaped user artifact can set this to the caller-supplied
    /// database path while keeping metadata/staging under `cache_dir`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub local_data_path: Option<PathBuf>,
    /// Open in read-only mode (no writes, no WAL)
    pub read_only: bool,

    /// Cache and durability-eviction knobs.
    pub cache: CacheConfig,
    /// Compression settings for remote page groups.
    pub compression: CompressionConfig,
    /// At-rest encryption settings.
    pub encryption: EncryptionConfig,
    /// Prefetch and query-plan-driven warmup knobs.
    pub prefetch: PrefetchConfig,
    /// WAL replication settings (walrust-backed durability between checkpoints).
    #[cfg(feature = "wal")]
    pub wal: WalConfig,

    // ---- Legacy flat config fields ----
    //
    // Kept so the pre-hadb S3 integration corpus can be ported onto the new
    // backend-injection implementation without losing its behavioral coverage.
    // New code should use the nested config groups plus TurboliteVfs::with_backend.
    #[serde(default)]
    pub bucket: String,
    #[serde(default)]
    pub prefix: String,
    #[serde(default)]
    pub endpoint_url: Option<String>,
    #[serde(default)]
    pub region: Option<String>,
    #[serde(skip)]
    pub runtime_handle: Option<tokio::runtime::Handle>,
    #[serde(default)]
    pub compression_level: i32,
    #[cfg(feature = "zstd")]
    #[serde(default)]
    pub dictionary: Option<Vec<u8>>,
    #[cfg(feature = "encryption")]
    #[serde(default)]
    pub encryption_key: Option<[u8; 32]>,
    #[serde(default)]
    pub pages_per_group: u32,
    #[serde(default)]
    pub sub_pages_per_frame: u32,
    #[serde(default)]
    pub gc_enabled: bool,
    #[serde(default)]
    pub max_cache_bytes: Option<u64>,
    #[serde(default)]
    pub evict_on_checkpoint: bool,
    #[serde(default)]
    pub prediction_enabled: bool,
    #[serde(default)]
    pub eager_index_load: bool,
    #[serde(default)]
    pub override_threshold: u32,
    #[serde(default)]
    pub compaction_threshold: u32,
    #[serde(default)]
    pub manifest_source: ManifestSource,
    #[allow(deprecated)]
    #[serde(default)]
    pub sync_mode: SyncMode,
    #[cfg(feature = "wal")]
    #[serde(default)]
    pub wal_replication: bool,
    #[cfg(feature = "wal")]
    #[serde(default)]
    pub wal_sync_interval_ms: u64,
}

impl Default for TurboliteConfig {
    fn default() -> Self {
        Self {
            cache_dir: PathBuf::from("/tmp/turbolite-cache"),
            local_data_path: None,
            read_only: false,
            cache: CacheConfig::default(),
            compression: CompressionConfig::default(),
            encryption: EncryptionConfig::default(),
            prefetch: PrefetchConfig::default(),
            #[cfg(feature = "wal")]
            wal: WalConfig::default(),
            bucket: String::new(),
            prefix: String::new(),
            endpoint_url: None,
            region: None,
            runtime_handle: None,
            compression_level: CompressionConfig::default().level,
            #[cfg(feature = "zstd")]
            dictionary: None,
            #[cfg(feature = "encryption")]
            encryption_key: None,
            pages_per_group: CacheConfig::default().pages_per_group,
            sub_pages_per_frame: CacheConfig::default().sub_pages_per_frame,
            gc_enabled: CacheConfig::default().gc_enabled,
            max_cache_bytes: CacheConfig::default().max_bytes,
            evict_on_checkpoint: CacheConfig::default().evict_on_checkpoint,
            prediction_enabled: PrefetchConfig::default().prediction,
            eager_index_load: true,
            override_threshold: CacheConfig::default().override_threshold,
            compaction_threshold: CacheConfig::default().compaction_threshold,
            manifest_source: ManifestSource::default(),
            #[allow(deprecated)]
            sync_mode: SyncMode::default(),
            #[cfg(feature = "wal")]
            wal_replication: WalConfig::default().replication,
            #[cfg(feature = "wal")]
            wal_sync_interval_ms: WalConfig::default().sync_interval_ms,
        }
    }
}

impl TurboliteConfig {
    /// Return a path-derived hidden state directory for a database file.
    ///
    /// For example, `state_dir_for_database_path("app.db", "-turbolite")`
    /// returns `app.db-turbolite` beside `app.db`.
    pub fn state_dir_for_database_path(
        db_path: impl AsRef<std::path::Path>,
        suffix: &str,
    ) -> PathBuf {
        let db_path = db_path.as_ref();
        let file_name = db_path
            .file_name()
            .map(|name| name.to_string_lossy())
            .unwrap_or_else(|| "turbolite.db".into());
        db_path.with_file_name(format!("{file_name}{suffix}"))
    }

    /// Build a file-first local configuration.
    ///
    /// The caller's `db_path` becomes the local database image and Turbolite's
    /// metadata/staging state lives under `<db_path>-turbolite/`.
    pub fn for_database_path(db_path: impl Into<PathBuf>) -> Self {
        let db_path = db_path.into();
        Self {
            cache_dir: Self::state_dir_for_database_path(&db_path, "-turbolite"),
            local_data_path: Some(db_path),
            ..Default::default()
        }
    }

    /// Convert an existing config to the file-first local layout.
    pub fn with_database_path(mut self, db_path: impl Into<PathBuf>) -> Self {
        let db_path = db_path.into();
        self.cache_dir = Self::state_dir_for_database_path(&db_path, "-turbolite");
        self.local_data_path = Some(db_path);
        self
    }

    pub(crate) fn has_legacy_backend_config(&self) -> bool {
        !self.bucket.is_empty() || !self.prefix.is_empty() || self.endpoint_url.is_some()
    }

    pub fn apply_legacy_flat_fields(&mut self) {
        if !self.has_legacy_backend_config() {
            return;
        }
        self.compression.level = self.compression_level;
        #[cfg(feature = "zstd")]
        {
            self.compression.dictionary = self.dictionary.clone();
        }
        #[cfg(feature = "encryption")]
        {
            self.encryption.key = self.encryption_key;
        }
        self.cache.pages_per_group = self.pages_per_group;
        self.cache.sub_pages_per_frame = self.sub_pages_per_frame;
        self.cache.gc_enabled = self.gc_enabled;
        #[allow(deprecated)]
        {
            if self.sync_mode == SyncMode::LocalThenFlush {
                self.cache.checkpoint_mode = CheckpointMode::LocalThenFlush;
            }
        }
        self.cache.max_bytes = self.max_cache_bytes;
        self.cache.evict_on_checkpoint = self.evict_on_checkpoint;
        self.cache.override_threshold = self.override_threshold;
        self.cache.compaction_threshold = self.compaction_threshold;
        self.prefetch.prediction = self.prediction_enabled;
        self.prefetch.query_plan = self.eager_index_load;
        self.prefetch.manifest_source = match self.manifest_source {
            ManifestSource::S3 => ManifestSource::Remote,
            other => other,
        };
        #[cfg(feature = "wal")]
        {
            self.wal.replication = self.wal_replication;
            self.wal.sync_interval_ms = self.wal_sync_interval_ms;
        }
    }

    /// Construct a TurboliteConfig by reading `TURBOLITE_*` environment variables.
    /// Fields not overridden by env vars come from `Default`. `cache_dir` reads
    /// `TURBOLITE_CACHE_DIR` (default `/tmp/turbolite-cache`).
    pub fn from_env() -> Self {
        let cache_dir = std::env::var("TURBOLITE_CACHE_DIR")
            .map(PathBuf::from)
            .unwrap_or_else(|_| PathBuf::from("/tmp/turbolite-cache"));
        Self {
            cache_dir,
            read_only: std::env::var("TURBOLITE_READ_ONLY")
                .map(|v| matches!(v.as_str(), "true" | "1"))
                .unwrap_or(false),
            cache: CacheConfig::from_env(),
            compression: CompressionConfig::from_env(),
            encryption: EncryptionConfig::default(),
            prefetch: PrefetchConfig::from_env(),
            #[cfg(feature = "wal")]
            wal: WalConfig::from_env(),
            ..Default::default()
        }
    }
}

#[cfg(test)]
#[path = "test_config.rs"]
mod tests;

// ===== Manifest =====

/// Location of a page within the explicit group mapping.
#[derive(Debug, Clone, Copy)]
pub struct PageLocation {
    pub group_id: u64,
    /// Position within the group's page list (index into group_pages[group_id])
    pub index: u32,
}

/// B-tree metadata stored in the manifest.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct BTreeManifestEntry {
    pub name: String,
    /// "table" or "index"
    pub obj_type: String,
    /// Group IDs containing this B-tree's pages
    pub group_ids: Vec<u64>,
    /// Exact 0-based pages belonging to this B-tree. Older manifests omit this
    /// and fall back to group-level attribution.
    #[serde(default)]
    pub pages: Vec<u64>,
}