pub mod toml_writer;

use std::time::SystemTime;

use anyhow::{Result, ensure};
use online_config::OnlineConfig;
use serde::{Deserialize, Serialize};
use tici_common::{
    config::{LogConfig, S3Config, ServerConfig},
    security::SecurityConfig,
};
use tici_shard::config::{FragWriterConfig, WorkerNodeConfig};
use tici_system::readable::{ReadableDuration, ReadableSize};

// MetaServiceConfig represents the ** top-level ** configuration for a
// tici-meta node.
#[derive(Debug, Serialize, Deserialize, Clone, Default, PartialEq, OnlineConfig)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct MetaServiceConfig {
    #[online_config(skip)]
    pub tidb_server: TiDBServerConfig,
    #[online_config(skip)]
    pub server: ServerConfig,
    #[online_config(skip)]
    pub election: ElectionConfig,
    #[online_config(skip)]
    pub reader_pool: ReaderPoolConfig,
    #[online_config(skip)]
    pub cache_scheduling: CacheSchedulingConfig,
    #[online_config(skip)]
    pub s3: S3Config,
    #[online_config(submodule)]
    pub shard: ShardConfig,
    #[online_config(skip)]
    pub worker_node: WorkerNodeMgrConfig,
    #[online_config(skip)]
    pub metrics: MetricsConfig,
    #[online_config(skip)]
    pub security: SecurityConfig,
    #[online_config(skip)]
    pub logger: LogConfig,
    #[online_config(skip)]
    pub import: ImportConfig,
}

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

    pub fn validate(&self) -> Result<()> {
        self.election.validate()?;
        Ok(())
    }
}

#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct ElectionConfig {
    /// Lease TTL used for leader election in etcd.
    pub campaign_ttl: ReadableDuration,
    /// Interval for retrying the local campaign loop.
    pub campaign_interval: ReadableDuration,
    /// Default config sets this to campaign_ttl / 3.
    pub lease_keepalive_interval: ReadableDuration,
    /// Lease TTL for service registration in etcd.
    // TODO(P1) There is a gap here, because etcd can only handles seconds in integer.
    pub registration_lease_ttl: ReadableDuration,
    /// Default config sets this to registration_lease_ttl / 3.
    pub registration_keepalive_interval: ReadableDuration,
}

impl Default for ElectionConfig {
    fn default() -> Self {
        let ttl = ReadableDuration::secs(10);
        let keepalive_interval =
            ReadableDuration::micros(ttl.as_millis().saturating_mul(1_000) / 3);
        Self {
            campaign_ttl: ttl,
            campaign_interval: ReadableDuration::secs(1),
            lease_keepalive_interval: keepalive_interval,
            registration_lease_ttl: ttl,
            registration_keepalive_interval: keepalive_interval,
        }
    }
}

impl ElectionConfig {
    pub fn effective_lease_keepalive_interval(&self) -> ReadableDuration {
        self.lease_keepalive_interval
    }

    pub fn effective_registration_keepalive_interval(&self) -> ReadableDuration {
        self.registration_keepalive_interval
    }

    pub fn validate(&self) -> Result<()> {
        ensure!(
            !self.campaign_ttl.is_zero(),
            "election.campaign_ttl must be greater than 0"
        );
        ensure!(
            !self.campaign_interval.is_zero(),
            "election.campaign_interval must be greater than 0"
        );
        ensure!(
            !self.effective_lease_keepalive_interval().is_zero(),
            "election.lease_keepalive_interval must be greater than 0"
        );
        ensure!(
            !self.registration_lease_ttl.is_zero(),
            "election.registration_lease_ttl must be greater than 0"
        );
        ensure!(
            !self.effective_registration_keepalive_interval().is_zero(),
            "election.registration_keepalive_interval must be greater than 0"
        );
        Ok(())
    }
}

/// Universal Compaction configuration for tiered compaction strategy
#[derive(Debug, Serialize, PartialEq, Deserialize, Clone, OnlineConfig)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct UniCompactionConfig {
    /// Maximum ratio of largest fragment in compaction window.
    /// Range: (0, 1]. Default: 0.5
    /// Controls write amplification: lower value = less write amplification but
    /// more fragments
    pub max_largest_ratio: f64,
    /// Minimum number of fragments required for a compaction.
    /// Default: 2
    pub min_compaction_frags: usize,
    /// Maximum compaction iterations in a chain, including the current one.
    /// Default: 2
    pub max_compaction_iterations: usize,
}

impl Default for UniCompactionConfig {
    fn default() -> Self {
        Self {
            max_largest_ratio: 0.5,
            min_compaction_frags: 5,
            max_compaction_iterations: 2,
        }
    }
}

/// Compactor configuration consolidating split/compaction parameters
#[derive(Debug, Serialize, PartialEq, Deserialize, Clone, OnlineConfig)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct CompactorConfig {
    /// Generate a potential split point for every `min_split_row_count` rows
    pub min_split_row_count: u64,
    /// Generate no more than `max_split_point_count` potential split points
    pub max_split_point_count: u64,
    /// Maximum concurrent compaction tasks per shard
    pub compaction_concurrency_shard: u64,
    /// Compaction speed factor for throttling
    pub compaction_speed_factor: f64,
}

impl Default for CompactorConfig {
    fn default() -> Self {
        Self {
            min_split_row_count: 40000,
            max_split_point_count: 4,
            compaction_concurrency_shard: 2,
            compaction_speed_factor: 1.0,
        }
    }
}

#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, OnlineConfig)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct MergeStrategyConfig {
    pub merge_check_tick: u64,
    pub merge_threshold: f64,
    pub write_qps: f64,
}

impl Default for MergeStrategyConfig {
    fn default() -> Self {
        Self {
            merge_check_tick: 3,
            merge_threshold: 0.2,
            write_qps: 0.01,
        }
    }
}

#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, OnlineConfig)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct SplitConfig {
    pub split_threshold: f64,
    pub split_by_append_factor: f64,
    /// Ratio of delta fragments (by size) to scan for split point computation.
    pub split_point_scan_size_ratio: f64,
}

impl Default for SplitConfig {
    fn default() -> Self {
        Self {
            split_threshold: 0.9,
            split_by_append_factor: 0.75,
            split_point_scan_size_ratio: 0.8,
        }
    }
}

#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, OnlineConfig)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct GcConfig {
    pub safe_gc_ts_diff: ReadableDuration,
    pub gc_worker_check_interval: ReadableDuration,
}

impl Default for GcConfig {
    fn default() -> Self {
        Self {
            safe_gc_ts_diff: ReadableDuration::minutes(60),
            gc_worker_check_interval: ReadableDuration::minutes(60),
        }
    }
}

#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, OnlineConfig)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct AppendConfig {
    pub append_empty_persist_freq: u64,
}

impl Default for AppendConfig {
    fn default() -> Self {
        Self {
            append_empty_persist_freq: 1,
        }
    }
}

#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, OnlineConfig)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct ShardConfig {
    pub max_size: ReadableSize,
    pub max_fragments: u64,
    pub auto_compaction: bool,
    /// Compaction policy selector. Supported values: "uni".
    pub compaction_policy: String,
    pub enable_split: bool,
    #[online_config(submodule)]
    pub split: SplitConfig,
    #[online_config(submodule)]
    pub uni_compaction: UniCompactionConfig,
    #[online_config(submodule)]
    pub compactor: CompactorConfig,
    pub enable_merge: bool,
    #[online_config(submodule)]
    pub merge_strategy: MergeStrategyConfig,
    /// Worker node scheduling strategy for shard assignment.
    /// Supported values: "round-robin", "balanced-count".
    pub writer_scheduling_strategy: String,
    pub patrol_interval: ReadableDuration,
    #[online_config(submodule)]
    pub append: AppendConfig,
    // Should be grater than `check_dead_shard_tick``
    pub shard_no_heartbeat_tick: u64,
    // Check all dead shards is expensive, so we don't do this in every tick.
    pub check_dead_shard_tick: u64,
    pub shard_scheduler_runner_cnt: u64,
    #[online_config(submodule)]
    pub gc_config: GcConfig,
}

impl Default for ShardConfig {
    fn default() -> Self {
        Self {
            max_size: ReadableSize::gb(4),
            max_fragments: 2000,
            auto_compaction: true,
            compaction_policy: "uni".to_string(),
            enable_split: true,
            split: SplitConfig::default(),
            uni_compaction: UniCompactionConfig::default(),
            compactor: CompactorConfig::default(),
            enable_merge: true,
            merge_strategy: MergeStrategyConfig::default(),
            writer_scheduling_strategy: "balanced-count".to_string(),
            patrol_interval: ReadableDuration::secs(10),
            append: AppendConfig::default(),
            shard_no_heartbeat_tick: 4,
            check_dead_shard_tick: 1,
            shard_scheduler_runner_cnt: 5,
            gc_config: GcConfig::default(),
        }
    }
}

impl ShardConfig {
    pub fn get_split_size(&self) -> u64 {
        (self.max_size.0 as f64 * self.split.split_threshold).ceil() as u64
    }

    pub fn get_split_size_with_propotion(&self, p: f64) -> u64 {
        (self.max_size.0 as f64 * self.split.split_threshold * p).ceil() as u64
    }

    pub fn get_merge_size(&self) -> u64 {
        (self.max_size.0 as f64 * self.merge_strategy.merge_threshold).ceil() as u64
    }

    pub fn is_abnormal_shard_no_heartbeat(&self, last_ts: u128) -> bool {
        let max_gap = self.shard_no_heartbeat_tick * self.patrol_interval.as_millis();
        SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .unwrap()
            .as_millis()
            > last_ts + max_gap as u128
    }

    #[cfg(any(test, feature = "testexport"))]
    pub fn enable_patrol(&mut self) {
        self.patrol_interval = ReadableDuration::millis(500);
    }

    #[cfg(any(test, feature = "testexport"))]
    pub fn disable_patrol(&mut self) {
        self.patrol_interval = ReadableDuration::days(99999);
    }
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct WorkerNodeMgrConfig {
    pub health_check_interval: ReadableDuration,
    // If a worker node fails to send heartbeat in `worker_node_down_timeout`,
    // the worker node will be unregisted from Worker Node Manager.
    // And all shards will be rescheduled.
    pub worker_node_down_timeout: ReadableDuration,
}

impl Default for WorkerNodeMgrConfig {
    fn default() -> Self {
        Self {
            health_check_interval: ReadableDuration::secs(5),
            worker_node_down_timeout: ReadableDuration::secs(30),
        }
    }
}

#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, OnlineConfig)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct TiDBServerConfig {
    // TODO(P2) TiDB could be dynamic later. Need discovery from PD.
    #[online_config(skip)]
    pub dsns: Vec<String>,
    // If set to 0, the blacklist mechanism is inactivated.
    pub blacklist_tidb_server_threshold: usize,
    pub blacklist_duration: ReadableDuration,
    #[online_config(skip)]
    pub cert_path: String,
    // Connection retry count. -1 means retry forever.
    pub tidb_connection_retry_count: i32,
    pub tidb_connection_retry_interval: ReadableDuration,
}

impl Default for TiDBServerConfig {
    fn default() -> Self {
        Self {
            dsns: vec!["mysql://root@localhost:4000".to_string()],
            blacklist_tidb_server_threshold: 10,
            blacklist_duration: ReadableDuration::secs(15),
            cert_path: "".to_string(),
            tidb_connection_retry_count: -1,
            tidb_connection_retry_interval: ReadableDuration::secs(2),
        }
    }
}

impl TiDBServerConfig {
    pub fn new_from_urls(v: &[&str]) -> Self {
        Self {
            dsns: v.iter().map(|x| (*x).to_owned()).collect::<Vec<_>>(),
            ..Default::default()
        }
    }

    pub fn overwrite_by_cmdline(&mut self, v: &Option<String>) {
        if let Some(val) = v {
            if !val.is_empty() {
                // split val by ',' and fill self.dsns
                self.dsns = val
                    .split(',')
                    .map(|s| s.trim().to_string())
                    .filter(|s| !s.is_empty())
                    .collect();
            }
        }
    }

    pub fn with_retry_count(mut self, count: i32) -> Self {
        self.tidb_connection_retry_count = count;
        self
    }

    pub fn with_retry_interval(mut self, interval: ReadableDuration) -> Self {
        self.tidb_connection_retry_interval = interval;
        self
    }
}

#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, OnlineConfig)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct ImportConfig {
    pub max_concurrency: usize,
}

impl Default for ImportConfig {
    fn default() -> Self {
        Self {
            max_concurrency: 128,
        }
    }
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Clone, Default)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct MetricsConfig {}

/// Configuration for reader pool management
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct ReaderPoolConfig {
    /// Time-to-live for reader nodes
    pub ttl_duration: ReadableDuration,
    /// Cleanup interval for expired readers
    pub cleanup_interval_duration: ReadableDuration,
    /// Scheduling strategy for reader selection
    pub scheduling_strategy: String,
}

impl Default for ReaderPoolConfig {
    fn default() -> Self {
        Self {
            ttl_duration: ReadableDuration::secs(30),
            cleanup_interval_duration: ReadableDuration::secs(1),
            scheduling_strategy: "round-robin".to_string(),
        }
    }
}

/// Configuration for cache scheduling and allocation
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct CacheSchedulingConfig {
    /// Scheduling strategy for cache allocation ("round-robin", "least-loaded",
    /// etc.)
    pub strategy: String,
    /// Maximum retry attempts for cache allocation
    pub max_retry_attempts: usize,
    /// Warmup timeout for cache allocation
    pub warmup_timeout_ms: ReadableDuration,
    /// Warm up shard caches eagerly when a shard is added.
    pub eager_schedule: bool,
    // If a shard cache is not included in the heartbeat for `dead_timeout`,
    // it is considered dead, and removed out of the ready list.
    pub dead_timeout: ReadableDuration,
    /// Maximum time to wait for manual replica deletion to be acknowledged by
    /// reader heartbeat.
    pub manual_remove_wait_timeout: ReadableDuration,
    pub allocate_parallelism: usize,
    // If not set, return more fragmented shards, but they can have newer data.
    // If set, the shard can only be replaced as a whole, so the scan performance could be better.
    pub optimize_for_fast_scan: bool,
}

impl Default for CacheSchedulingConfig {
    fn default() -> Self {
        Self {
            strategy: "balanced-count".to_string(),
            max_retry_attempts: 3,
            warmup_timeout_ms: ReadableDuration::millis(5000),
            eager_schedule: false,
            dead_timeout: ReadableDuration::secs(1),
            manual_remove_wait_timeout: ReadableDuration::secs(30),
            // For each reader node
            allocate_parallelism: 12,
            optimize_for_fast_scan: false,
        }
    }
}

pub fn make_default_test_meta_config(enable_status_server: bool) -> MetaServiceConfig {
    let mut config = MetaServiceConfig {
        tidb_server: TiDBServerConfig {
            tidb_connection_retry_count: 30,
            tidb_connection_retry_interval: ReadableDuration::secs(2),
            ..Default::default()
        },
        server: ServerConfig::default_for_meta_test(enable_status_server),
        s3: S3Config::default_for_test(),
        reader_pool: ReaderPoolConfig {
            ttl_duration: ReadableDuration::secs(4),
            cleanup_interval_duration: ReadableDuration::secs(2),
            ..Default::default()
        },
        ..Default::default()
    };
    config.shard.gc_config = GcConfig {
        safe_gc_ts_diff: ReadableDuration::secs(60),
        gc_worker_check_interval: ReadableDuration::secs(15),
    };
    config
}

pub fn make_default_test_worker_config(enable_status_server: bool) -> WorkerNodeConfig {
    let server_config = ServerConfig::default_for_worker_node_test(enable_status_server);
    WorkerNodeConfig {
        server: server_config,
        s3: S3Config::default_for_test(),
        frag_writer: FragWriterConfig {
            // smaller value to more frequently flush data during tests
            index_num_threads: 4,
            index_mem_budget: ReadableSize::mb(255),
            index_flush_interval: ReadableDuration::secs(5),
            index_flush_size_limit: ReadableSize::mb(5),
            ..Default::default()
        },
        heartbeat_interval: tici_system::readable::ReadableDuration::millis(500),
        meta_service_wait_timeout: 60,
        ..Default::default()
    }
}

#[cfg(test)]
mod tests {
    use std::{collections::HashMap, io::Write};

    use online_config::ConfigValue;
    use tempfile::NamedTempFile;

    use super::*;

    #[test]
    fn test_load_valid_config() {
        let mut config = String::from(
            r#"
        [tidb-server]
        dsns = ["mysql://root@localhost:4000", "mysql://root@localhost:4001"]
        cert-path = "/path/to/cert"

        [server]
        addr = "127.0.0.1:8500"
        advertise-addr = "127.0.0.1:8500"
        status-addr = "127.0.0.1:8501"
        advertise-status-addr = "127.0.0.1:8501"
        pd-addr = "127.0.0.1:2379"

        [reader-pool]
        ttl-duration = "10s"
        cleanup-interval-duration = "2s"
        scheduling-strategy = "least-loaded"

        [election]
        campaign-ttl = "12s"
        campaign-interval = "2500ms"
        lease-keepalive-interval = "1500ms"
        registration-lease-ttl = "8s"
        registration-keepalive-interval = "1200ms"

        [cache-scheduling]
        strategy = "least-loaded"
        max-retry-attempts = 5
        warmup-timeout-ms = "10000ms"
        eager-schedule = true
        dead-timeout = "2s"
        allocate-parallelism = 32

        [s3]
        endpoint = "localhost:9000"
        region = "us-west-2"
        access-key = "testkey"
        secret-key = "testsecret"
        use-path-style = true
        bucket = "test-bucket"
        prefix = "test-prefix"

        [shard]
        max-size = "2GB"
        auto-compaction = false
        enable-split = false
        enable-merge = false
        patrol-interval = "15s"
        shard-no-heartbeat-tick = 6
        check-dead-shard-tick = 2
        shard-scheduler-runner-cnt = 2

        [shard.split]
        split-threshold = 0.8
        split-by-append-factor = 0.6
        split-point-scan-size-ratio = 0.4

        [shard.compactor]
        compaction-concurrency-shard = 4
    "#,
        );

        config.push_str(
            r#"

        [shard.merge-strategy]
        merge-check-tick = 5
        merge-threshold = 0.3
        write-qps = 0.02

        [shard.gc-config]
        safe-gc-ts-diff = "20m"
        gc-worker-check-interval = "90m"

        [worker-node]
        health-check-interval = "10s"
        worker-node-down-timeout = "60s"

        [security]
        cert-path = "/path/to/cert"
        key-path = "/path/to/key"
        ca-path = "/path/to/ca-cert"

        [logger]
        filename = "tici_meta.log"
        level = "warn"
        size = "64MB"
        count = 8

        [import]
        max-concurrency = 256
    "#,
        );

        let mut file = NamedTempFile::new().unwrap();
        file.write_all(config.as_bytes()).unwrap();

        let config = MetaServiceConfig::load(file.path().to_str().unwrap())
            .expect("Failed to load valid config");

        assert_eq!(
            config.tidb_server.dsns,
            vec!["mysql://root@localhost:4000", "mysql://root@localhost:4001"]
        );
        assert_eq!(config.tidb_server.cert_path, "/path/to/cert");

        assert_eq!(config.server.addr, "127.0.0.1:8500");
        assert_eq!(config.server.get_advertise_addr(), "127.0.0.1:8500");
        assert_eq!(config.server.status_addr, "127.0.0.1:8501");
        assert_eq!(config.server.get_advertise_status_addr(), "127.0.0.1:8501");
        assert_eq!(config.server.pd_addr, "127.0.0.1:2379");

        assert_eq!(config.reader_pool.ttl_duration, ReadableDuration::secs(10));
        assert_eq!(
            config.reader_pool.cleanup_interval_duration,
            ReadableDuration::secs(2)
        );
        assert_eq!(config.reader_pool.scheduling_strategy, "least-loaded");
        assert_eq!(config.election.campaign_ttl, ReadableDuration::secs(12));
        assert_eq!(
            config.election.campaign_interval,
            ReadableDuration::millis(2500)
        );
        assert_eq!(
            config.election.lease_keepalive_interval,
            ReadableDuration::millis(1500)
        );
        assert_eq!(
            config.election.registration_lease_ttl,
            ReadableDuration::secs(8)
        );
        assert_eq!(
            config.election.registration_keepalive_interval,
            ReadableDuration::millis(1200)
        );

        assert_eq!(config.cache_scheduling.strategy, "least-loaded");
        assert_eq!(config.cache_scheduling.max_retry_attempts, 5);
        assert_eq!(
            config.cache_scheduling.warmup_timeout_ms,
            ReadableDuration::millis(10000)
        );
        assert!(config.cache_scheduling.eager_schedule);
        assert_eq!(
            config.cache_scheduling.dead_timeout,
            ReadableDuration::secs(2)
        );
        assert_eq!(config.cache_scheduling.allocate_parallelism, 32);

        assert_eq!(config.s3.endpoint, "localhost:9000");
        assert_eq!(config.s3.region, "us-west-2");
        assert_eq!(config.s3.access_key, "testkey");
        assert_eq!(config.s3.secret_key, "testsecret");
        assert!(config.s3.use_path_style);
        assert_eq!(config.s3.bucket, "test-bucket");
        assert_eq!(config.s3.prefix, "test-prefix");

        assert_eq!(config.shard.max_size, ReadableSize::gb(2));
        assert_eq!(config.shard.split.split_threshold, 0.8);
        assert_eq!(config.shard.split.split_by_append_factor, 0.6);
        assert_eq!(config.shard.split.split_point_scan_size_ratio, 0.4);
        assert!(!config.shard.auto_compaction);
        assert_eq!(config.shard.compaction_policy, "uni");
        assert_eq!(config.shard.compactor.compaction_concurrency_shard, 4);
        assert!(!config.shard.enable_split);
        assert!(!config.shard.enable_merge);
        assert_eq!(config.shard.patrol_interval, ReadableDuration::secs(15));
        assert_eq!(config.shard.shard_no_heartbeat_tick, 6);
        assert_eq!(config.shard.check_dead_shard_tick, 2);
        assert_eq!(config.shard.shard_scheduler_runner_cnt, 2);

        assert_eq!(config.shard.merge_strategy.merge_check_tick, 5);
        assert_eq!(config.shard.merge_strategy.merge_threshold, 0.3);
        assert_eq!(config.shard.merge_strategy.write_qps, 0.02);

        assert_eq!(
            config.shard.gc_config.safe_gc_ts_diff,
            ReadableDuration::minutes(20)
        );
        assert_eq!(
            config.shard.gc_config.gc_worker_check_interval,
            ReadableDuration::minutes(90)
        );

        assert_eq!(
            config.worker_node.health_check_interval,
            ReadableDuration::secs(10)
        );
        assert_eq!(
            config.worker_node.worker_node_down_timeout,
            ReadableDuration::secs(60)
        );

        assert_eq!(config.security.cert_path, "/path/to/cert");
        assert_eq!(config.security.key_path, "/path/to/key");
        assert_eq!(config.security.ca_path, "/path/to/ca-cert");

        assert_eq!(config.logger.filename, "tici_meta.log");
        assert_eq!(config.logger.get_level_or(""), "warn");
        assert_eq!(config.logger.size, ReadableSize::mb(64));
        assert_eq!(config.logger.count, 8);

        assert_eq!(config.import.max_concurrency, 256);
    }

    #[test]
    fn test_load_empty_config() {
        const EMPTY_CONFIG: &str = r#"
    "#;

        let mut file = NamedTempFile::new().unwrap();
        file.write_all(EMPTY_CONFIG.as_bytes()).unwrap();

        let result = MetaServiceConfig::load(file.path().to_str().unwrap());
        assert!(result.is_ok());

        let config = result.unwrap();
        assert_eq!(
            config.tidb_server.dsns,
            vec!["mysql://root@localhost:4000".to_string()]
        );
        assert_eq!(config.tidb_server, TiDBServerConfig::default());
        assert_eq!(config.server, ServerConfig::default());
        assert_eq!(config.election, ElectionConfig::default());
        assert_eq!(config.reader_pool, ReaderPoolConfig::default());
        assert_eq!(config.cache_scheduling, CacheSchedulingConfig::default());
        assert_eq!(config.s3, S3Config::default());
        assert_eq!(config.shard, ShardConfig::default());
        assert_eq!(config.worker_node, WorkerNodeMgrConfig::default());
        assert_eq!(config.metrics, MetricsConfig::default());
        assert_eq!(config.security, SecurityConfig::default());
        assert_eq!(config.logger, LogConfig::default());
        assert_eq!(config.import, ImportConfig::default());
    }

    #[test]
    fn test_nonexistent_file() {
        let result = MetaServiceConfig::load("/path/that/does/not/exist.toml");
        assert!(result.is_err(), "Should fail on nonexistent file");
        assert!(
            result.unwrap_err().to_string().contains("No such file"),
            "Error should mention file not found"
        );
    }

    #[test]
    fn test_config_serialization() {
        let config = MetaServiceConfig::default();
        let serialized = toml::to_string(&config).unwrap();
        let deserialized: MetaServiceConfig = toml::from_str(&serialized).unwrap();

        assert_eq!(config.tidb_server.dsns, deserialized.tidb_server.dsns);
        assert_eq!(
            config.tidb_server.cert_path,
            deserialized.tidb_server.cert_path
        );
        assert_eq!(
            config.reader_pool.ttl_duration,
            deserialized.reader_pool.ttl_duration
        );
        assert_eq!(config.cache_scheduling, deserialized.cache_scheduling);
        assert_eq!(config.election, deserialized.election);
        assert_eq!(config.s3.bucket, deserialized.s3.bucket);
        assert_eq!(config.shard.max_size, deserialized.shard.max_size);
    }

    #[test]
    fn test_election_config_effective_keepalive_interval() {
        let default_keepalive_interval = ReadableDuration::micros(3_333_333);
        let config = ElectionConfig::default();
        assert_eq!(
            config.effective_lease_keepalive_interval(),
            default_keepalive_interval
        );

        let config = ElectionConfig {
            campaign_ttl: ReadableDuration::secs(12),
            lease_keepalive_interval: ReadableDuration::secs(5),
            ..Default::default()
        };
        assert_eq!(
            config.effective_lease_keepalive_interval(),
            ReadableDuration::secs(5)
        );

        let config = ElectionConfig::default();
        assert_eq!(
            config.effective_registration_keepalive_interval(),
            default_keepalive_interval
        );
        assert_eq!(config.registration_lease_ttl, ReadableDuration::secs(10));

        let config = ElectionConfig {
            campaign_ttl: ReadableDuration::secs(12),
            registration_lease_ttl: ReadableDuration::secs(7),
            lease_keepalive_interval: ReadableDuration::secs(5),
            registration_keepalive_interval: ReadableDuration::secs(2),
            ..Default::default()
        };
        assert_eq!(config.registration_lease_ttl, ReadableDuration::secs(7));
        assert_eq!(
            config.effective_registration_keepalive_interval(),
            ReadableDuration::secs(2)
        );
    }

    #[test]
    fn test_shard_config_heartbeat_check() {
        let config = ShardConfig {
            patrol_interval: ReadableDuration::secs(10),
            shard_no_heartbeat_tick: 4,
            ..Default::default()
        };

        let now = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .unwrap()
            .as_millis();

        // Recent timestamp should not be abnormal
        let recent_ts = now - 5000; // 5 seconds ago
        assert!(!config.is_abnormal_shard_no_heartbeat(recent_ts));

        // Old timestamp should be abnormal (more than 4 * 10s = 40s)
        let old_ts = now - 50000; // 50 seconds ago
        assert!(config.is_abnormal_shard_no_heartbeat(old_ts));
    }

    #[test]
    fn test_config_update() {
        let mut m = MetaServiceConfig::default();

        {
            let mut m1 = m.clone();
            m1.shard.max_size = ReadableSize(123);
            let mut shard_diff_h: HashMap<String, ConfigValue> = HashMap::new();
            shard_diff_h.insert("max_size".to_string(), ReadableSize(234).into());

            m1.shard.update(shard_diff_h.clone()).unwrap();
            assert_eq!(m1.shard.max_size, ReadableSize(234));
        }

        {
            let mut m1 = m.clone();
            let mut shard_diff_h: HashMap<String, ConfigValue> = HashMap::new();
            shard_diff_h.insert("max_size".to_string(), ReadableSize(234).into());
            m1.shard.max_size = ReadableSize(123);
            m.shard.max_size = ReadableSize(123);
            let mut h: HashMap<String, ConfigValue> = HashMap::new();
            h.insert("shard".to_string(), ConfigValue::Module(shard_diff_h));
            m.update(h).unwrap();
            assert_eq!(m.shard.max_size, ReadableSize(234));
            assert_eq!(m1.shard.max_size, ReadableSize(123));

            let encoded = toml::to_string(&m.get_encoder()).unwrap();
            let decoded: MetaServiceConfig = toml::from_str(encoded.as_str()).unwrap();
            assert_eq!(decoded, m);
        }
    }

    #[test]
    fn test_cmd_overwrite() {
        let default_dsn = vec!["mysql://root@localhost:4000".to_string()];
        let mut cfg = TiDBServerConfig::default();
        assert_eq!(cfg.dsns, default_dsn);
        cfg.overwrite_by_cmdline(&None);
        assert_eq!(cfg.dsns, default_dsn);
        cfg.overwrite_by_cmdline(&Some("".to_string()));
        assert_eq!(cfg.dsns, default_dsn);
        cfg.overwrite_by_cmdline(&Some(
            "mysql://root@127.0.0.1:4000,mysql://root@127.0.0.1:4001".to_string(),
        ));
        assert_eq!(
            cfg.dsns,
            vec!["mysql://root@127.0.0.1:4000", "mysql://root@127.0.0.1:4001",]
        );
    }
}