yantrikdb-server 0.8.4

use serde::Deserialize;
use std::path::{Path, PathBuf};

pub mod live_reload;
pub mod tenant_overrides;
pub mod versioned;
pub mod watch;

pub use live_reload::{ReloadError, ReloadOutcome, Reloadable};
pub use tenant_overrides::{
    InMemoryTenantConfigStore, OverrideValue, TenantConfigError, TenantConfigOverride,
    TenantConfigStore,
};
pub use versioned::{ConfigDelta, ConfigVersion, VersionedConfig};
pub use watch::{ConfigWatch, ConfigWatchSender};

#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
#[derive(Default)]
pub struct ServerConfig {
    pub server: ServerSection,
    pub tls: TlsSection,
    pub encryption: EncryptionSection,
    pub embedding: EmbeddingSection,
    pub background: BackgroundSection,
    pub limits: LimitsSection,
    pub cluster: ClusterSection,
    /// RFC 014-A: cluster-transport mTLS. Optional in legacy cluster
    /// mode; production gate for RFC 010 PR-4 openraft.
    pub cluster_tls: crate::security::ClusterTlsConfig,
}

#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
pub struct ServerSection {
    pub wire_port: u16,
    pub http_port: u16,
    pub data_dir: PathBuf,
}

#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
pub struct EmbeddingSection {
    pub strategy: EmbeddingStrategy,
    pub dim: usize,
}

#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
#[derive(Default)]
pub struct TlsSection {
    pub cert_path: Option<PathBuf>,
    pub key_path: Option<PathBuf>,
}

impl TlsSection {
    pub fn is_enabled(&self) -> bool {
        self.cert_path.is_some() && self.key_path.is_some()
    }
}

// ── Encryption ─────────────────────────────────────────────────

#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
#[derive(Default)]
pub struct EncryptionSection {
    /// Path to a 32-byte master key file (raw bytes).
    /// If unset and `auto_generate` is true, one is created on first startup.
    pub key_path: Option<PathBuf>,

    /// If true, generate a fresh key file at `key_path` (or `data_dir/master.key`)
    /// when none exists. Default true for ease of setup.
    pub auto_generate: bool,

    /// Master key value as hex string (64 chars). Takes precedence over key_path.
    /// Useful for env-driven config.
    pub key_hex: Option<String>,
}

impl EncryptionSection {
    /// Whether encryption is enabled (any key source configured).
    ///
    /// Not currently called — reserved for future startup banner /
    /// /v1/admin/status surfacing of at-rest encryption state.
    #[allow(dead_code)]
    pub fn is_enabled(&self) -> bool {
        self.key_path.is_some() || self.key_hex.is_some() || self.auto_generate
    }

    /// Resolve the master key from this configuration. Generates one if needed.
    pub fn resolve_key(&self, data_dir: &Path) -> anyhow::Result<Option<[u8; 32]>> {
        // Priority 0: env var override (issue #6 — env-friendly setup).
        // Documented as the env-equivalent of `[encryption] key_hex`. Takes
        // precedence over TOML so an operator can rotate the key without
        // editing the file.
        if let Ok(hex_str) = std::env::var("YANTRIKDB_ENCRYPTION_KEY_HEX") {
            let bytes = hex::decode(hex_str.trim())
                .map_err(|e| anyhow::anyhow!("invalid YANTRIKDB_ENCRYPTION_KEY_HEX: {}", e))?;
            if bytes.len() != 32 {
                anyhow::bail!(
                    "YANTRIKDB_ENCRYPTION_KEY_HEX must decode to exactly 32 bytes (got {})",
                    bytes.len()
                );
            }
            let mut key = [0u8; 32];
            key.copy_from_slice(&bytes);
            tracing::info!("encryption: enabled via YANTRIKDB_ENCRYPTION_KEY_HEX env var");
            return Ok(Some(key));
        }

        // Priority 1: explicit hex value in TOML
        if let Some(ref hex_str) = self.key_hex {
            let bytes = hex::decode(hex_str)
                .map_err(|e| anyhow::anyhow!("invalid encryption.key_hex: {}", e))?;
            if bytes.len() != 32 {
                anyhow::bail!("encryption.key_hex must decode to exactly 32 bytes");
            }
            let mut key = [0u8; 32];
            key.copy_from_slice(&bytes);
            return Ok(Some(key));
        }

        // Priority 2: explicit key file
        let path = match &self.key_path {
            Some(p) => p.clone(),
            None if self.auto_generate => data_dir.join("master.key"),
            None => return Ok(None),
        };

        if path.exists() {
            let bytes = std::fs::read(&path)?;
            if bytes.len() != 32 {
                anyhow::bail!(
                    "key file at {} must be exactly 32 bytes (got {})",
                    path.display(),
                    bytes.len()
                );
            }
            let mut key = [0u8; 32];
            key.copy_from_slice(&bytes);
            return Ok(Some(key));
        }

        // Auto-generate
        if self.auto_generate {
            use rand::RngCore;
            let mut key = [0u8; 32];
            rand::thread_rng().fill_bytes(&mut key);
            if let Some(parent) = path.parent() {
                std::fs::create_dir_all(parent)?;
            }
            std::fs::write(&path, key)?;

            #[cfg(unix)]
            {
                use std::os::unix::fs::PermissionsExt;
                let _ = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600));
            }

            tracing::info!(
                path = %path.display(),
                "auto-generated encryption master key"
            );
            return Ok(Some(key));
        }

        Ok(None)
    }
}

#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EmbeddingStrategy {
    Builtin,
    ClientOnly,
}

#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
pub struct BackgroundSection {
    pub consolidation_interval_minutes: u64,
    pub decay_sweep_interval_minutes: u64,
}

#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
pub struct LimitsSection {
    pub max_databases: usize,
    pub max_connections: usize,
}

// ── Cluster / Replication ──────────────────────────────────────

#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
pub struct ClusterSection {
    /// Unique node identifier (used for HLC and Raft).
    /// 0 means single-node mode (no replication).
    pub node_id: u32,

    /// Role for this node in the cluster.
    pub role: NodeRole,

    /// Port for inter-peer cluster traffic (separate from client wire port).
    /// Defaults to 7440.
    pub cluster_port: u16,

    /// Address other peers should use to reach this node (host:cluster_port).
    /// If unset, derived from cluster_port + hostname.
    pub advertise_addr: Option<String>,

    /// List of peer nodes in the cluster.
    pub peers: Vec<PeerConfig>,

    /// Heartbeat interval in milliseconds (default 1000ms = 1s).
    pub heartbeat_interval_ms: u64,

    /// Election timeout in milliseconds (default 5000ms = 5s).
    /// If a follower doesn't hear from leader for this long, election starts.
    pub election_timeout_ms: u64,

    /// Shared cluster secret for authenticating peer connections.
    /// All nodes in a cluster must share the same secret.
    pub cluster_secret: Option<String>,

    /// Replication mode: async (default) or sync.
    pub replication_mode: ReplicationMode,

    /// RFC 010 PR-4: which Raft engine drives the cluster. Defaults to
    /// `Disabled` (single-node, no Raft). `OpenRaft` activates the
    /// production-grade openraft engine — and triggers the mTLS
    /// production gate at server startup (see
    /// [`crate::raft::build_raft_cluster`]).
    pub raft_mode: crate::raft::RaftClusterMode,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NodeRole {
    /// Standalone node, no replication. Default.
    Single,
    /// Full data node that can become primary or secondary via election.
    Voter,
    /// Read-only replica that consumes oplog but never votes or accepts writes.
    ReadReplica,
    /// Witness — vote-only node, no data storage. Tiebreaker for 2-node clusters.
    Witness,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ReplicationMode {
    /// Writes return immediately, replicas catch up asynchronously.
    Async,
    /// Writes block until quorum of secondaries ack.
    Sync,
}

#[derive(Debug, Clone, Deserialize)]
pub struct PeerConfig {
    /// Peer's wire protocol address (host:port).
    pub addr: String,
    /// Peer's role in the cluster.
    pub role: NodeRole,
}

impl Default for ClusterSection {
    fn default() -> Self {
        Self {
            node_id: 0,
            role: NodeRole::Single,
            cluster_port: 7440,
            advertise_addr: None,
            peers: Vec::new(),
            heartbeat_interval_ms: 1000,
            election_timeout_ms: 5000,
            cluster_secret: None,
            replication_mode: ReplicationMode::Async,
            raft_mode: crate::raft::RaftClusterMode::Disabled,
        }
    }
}

impl ClusterSection {
    /// Whether replication is enabled (i.e. not single-node mode).
    pub fn is_clustered(&self) -> bool {
        self.role != NodeRole::Single
    }

    /// Total voter count (this node + voter peers, excluding witness/read replicas).
    ///
    /// Not currently called — reserved for quorum diagnostics and config
    /// validation on startup.
    #[allow(dead_code)]
    pub fn voter_count(&self) -> usize {
        let self_voter = matches!(self.role, NodeRole::Voter) as usize;
        let peer_voters = self
            .peers
            .iter()
            .filter(|p| p.role == NodeRole::Voter)
            .count();
        self_voter + peer_voters
    }

    /// Total quorum members (voters + witnesses) for elections.
    pub fn quorum_members(&self) -> usize {
        let self_member = matches!(self.role, NodeRole::Voter | NodeRole::Witness) as usize;
        let peer_members = self
            .peers
            .iter()
            .filter(|p| matches!(p.role, NodeRole::Voter | NodeRole::Witness))
            .count();
        self_member + peer_members
    }

    /// Quorum size needed for elections (N/2 + 1).
    pub fn quorum_size(&self) -> usize {
        let total = self.quorum_members();
        total / 2 + 1
    }
}

impl Default for ServerSection {
    fn default() -> Self {
        Self {
            wire_port: 7437,
            http_port: 7438,
            data_dir: PathBuf::from("./data"),
        }
    }
}

impl Default for EmbeddingSection {
    fn default() -> Self {
        Self {
            strategy: EmbeddingStrategy::Builtin,
            dim: 384,
        }
    }
}

impl Default for BackgroundSection {
    fn default() -> Self {
        Self {
            consolidation_interval_minutes: 30,
            decay_sweep_interval_minutes: 60,
        }
    }
}

impl Default for LimitsSection {
    fn default() -> Self {
        Self {
            max_databases: 100,
            max_connections: 1000,
        }
    }
}

impl ServerConfig {
    pub fn load(path: &Path) -> anyhow::Result<Self> {
        let content = std::fs::read_to_string(path)?;
        let config: ServerConfig = toml::from_str(&content)?;
        Ok(config)
    }

    pub fn data_dir(&self) -> &Path {
        &self.server.data_dir
    }

    pub fn control_db_path(&self) -> PathBuf {
        self.server.data_dir.join("control.db")
    }
}

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

    /// Issue #6 regression: YANTRIKDB_ENCRYPTION_KEY_HEX env var must be
    /// honored as the env-equivalent of `[encryption] key_hex` in TOML.
    /// Pre-fix: env var was silently ignored, encryption disabled.
    ///
    /// All env-var assertions consolidated into ONE test because env vars
    /// are process-global and Rust runs tests in parallel by default.
    /// Splitting these into multiple #[test]s race on the shared var.
    #[test]
    fn encryption_env_var_handling() {
        const VAR: &str = "YANTRIKDB_ENCRYPTION_KEY_HEX";
        let cfg_default = EncryptionSection {
            key_path: None,
            auto_generate: false,
            key_hex: None,
        };

        // Case 1: 64 hex chars (32 bytes) — env var produces the right key.
        let key_hex = "f".repeat(64);
        std::env::set_var(VAR, &key_hex);
        let resolved = cfg_default.resolve_key(Path::new("/tmp")).unwrap();
        assert_eq!(resolved, Some([0xffu8; 32]), "env var should produce key");

        // Case 2: invalid hex — error mentions the env var name.
        std::env::set_var(VAR, "not-hex-at-all");
        let err = cfg_default.resolve_key(Path::new("/tmp")).unwrap_err();
        assert!(err.to_string().contains(VAR));

        // Case 3: valid hex but wrong byte length — error mentions length.
        std::env::set_var(VAR, "ab"); // 1 byte
        let err = cfg_default.resolve_key(Path::new("/tmp")).unwrap_err();
        assert!(err.to_string().contains("32 bytes"));

        // Case 4: env var takes precedence over TOML — set both, env wins.
        std::env::set_var(VAR, &key_hex);
        let cfg_with_toml = EncryptionSection {
            key_path: None,
            auto_generate: false,
            key_hex: Some("0".repeat(64)), // would resolve to all-zeros
        };
        let resolved = cfg_with_toml.resolve_key(Path::new("/tmp")).unwrap();
        assert_eq!(
            resolved,
            Some([0xffu8; 32]),
            "env var must beat TOML key_hex"
        );

        std::env::remove_var(VAR);
    }
}