use anyhow::{Context, Result, anyhow, bail, ensure};
use serde::{Deserialize, Serialize};
use tici_common::{
    config::{GrpcConfig, S3Config},
    security::SecurityConfig,
};
use tici_shard::config::{FragReaderConfig, StorageConfig};
use tici_system::readable::ReadableDuration;

// `Config` represents the configuration for a TiFlash compute node that runs
// the tici searchlib.
// To simplify deployment, tici searchlib embeds its settings directly into the
// TiFlash configuration file, eliminating the need for a separate config file
// for the tici searchlib. As a result, Config includes not only TiCI-specific
// settings but also certain TiFlash settings that are not directly part of TiCI
// itself.
#[derive(Debug, Serialize, Deserialize, Clone, Default)]
#[serde(default)]
pub struct Config {
    pub tici: TiCIConfig,

    pub flash: TiFlashServerConfig,

    #[serde(rename = "raft")]
    pub pd: PDConfig,

    // Logger config of the TiFlash compute node.
    pub logger: TiFlashLoggerConfig,
}

impl Config {
    pub fn load(config_path: &str) -> Result<Self> {
        let content = std::fs::read_to_string(config_path)
            .with_context(|| format!("Read {:?}", config_path))?;
        Self::load_from_str(&content)
    }

    pub fn load_from_str(toml_str: &str) -> Result<Self> {
        let config: Config =
            toml::from_str(toml_str).with_context(|| format!("Deserialize {}", toml_str))?;
        config.validate()?;
        Ok(config)
    }

    fn validate(&self) -> Result<()> {
        ensure!(
            !self.tici.storage.data_dir.is_empty(),
            "tici.storage.data_dir is required"
        );
        ensure!(
            self.tici.frag_reader.shard_sample_count > 0,
            "tici.frag_reader.shard_sample_count must be greater than 0"
        );
        ensure!(
            self.tici.frag_reader.frag_sample_count > 0,
            "tici.frag_reader.frag_sample_count must be greater than 0"
        );
        Ok(())
    }

    pub fn resolve_reader_node_addr(&mut self) -> Result<()> {
        if !self.tici.reader_node.addr.trim().is_empty() {
            return Ok(());
        }
        if self.tici.reader_node.port == 0 {
            bail!("reader_node.port must be set when reader_node.addr is empty");
        }

        let mut parts = self.flash.service_addr.split(':');
        let host = parts
            .next()
            .ok_or_else(|| anyhow!("flash.service_addr is invalid: {}", self.flash.service_addr))?;
        // use the host from flash.service_addr and the port from reader_node.port for
        // setting reader_node.advertise_addr
        self.tici.reader_node.advertise_addr = format!("{}:{}", host, self.tici.reader_node.port);
        self.tici.reader_node.addr = format!("0.0.0.0:{}", self.tici.reader_node.port);
        Ok(())
    }
}

// `TiFlashServerConfig` represents the TiFlash server configuration for the
// TiFlash compute node. A helper to parse the service address from the TiFlash
// config file.
// So do not use kebab-case for TiFlashServerConfig.
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Default)]
#[serde(default)]
pub struct TiFlashServerConfig {
    #[serde(default)]
    pub service_addr: String,
}

// `PDConfig` represents the PD configuration for the TiFlash compute node.
// A helper to parse the PD addresses from the TiFlash config file.
// So do not use kebab-case for PDConfig.
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Default)]
#[serde(default)]
pub struct PDConfig {
    pub pd_addr: String,
}

impl PDConfig {
    pub fn get_pd_addresses(&self) -> Vec<String> {
        self.pd_addr
            .split(',')
            .map(|addr| addr.trim().to_string())
            .filter(|addr| !addr.is_empty())
            .collect()
    }
}

// `TiFlashLoggerConfig` represents the logger configuration for the TiFlash
// compute node. A helper to parse the logger level from the TiFlash config
// file.
// So do not use kebab-case for TiFlashLoggerConfig.
#[derive(Clone, Debug, Serialize, Deserialize, Default)]
#[serde(default)]
pub struct TiFlashLoggerConfig {
    pub level: String,
}

// `TiCIConfig` represents the ** top-level ** configuration for tici searchlib
// within the TiFlash compute node.
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct TiCIConfig {
    pub s3: S3Config,

    pub reader_node: ReaderNodeConfig,

    pub frag_reader: FragReaderConfig,

    pub security: SecurityConfig,

    pub logger: tici_common::config::LogConfig,

    pub storage: StorageConfig,
}

impl Default for TiCIConfig {
    fn default() -> Self {
        Self {
            s3: S3Config::default(),
            reader_node: ReaderNodeConfig::default(),
            frag_reader: FragReaderConfig::default(),
            security: SecurityConfig::default(),
            logger: tici_common::config::LogConfig::new("tici_searchlib.log".to_string()),
            storage: StorageConfig::default(),
        }
    }
}

#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct ReaderNodeConfig {
    pub addr: String,

    // If addr is empty, use flash.service_addr and port to bind.
    pub port: u64,

    pub advertise_addr: String,

    pub heartbeat_interval: ReadableDuration,

    pub max_heartbeat_retries: u32,

    pub heartbeat_worker_count: u32,

    pub heartbeat_queue_size: u32,

    pub grpc: GrpcConfig,
    pub grpc_client: GrpcConfig,
}

impl Default for ReaderNodeConfig {
    fn default() -> Self {
        Self {
            addr: "".to_string(),
            port: 8520,
            advertise_addr: "".to_string(),
            heartbeat_interval: ReadableDuration::secs(3),
            max_heartbeat_retries: 3,
            heartbeat_worker_count: 8,
            heartbeat_queue_size: 1024,
            grpc: GrpcConfig::default(),
            grpc_client: GrpcConfig::default(),
        }
    }
}

#[cfg(test)]
mod tests {

    use anyhow::Result;
    use tici_common::config::S3Config;
    use tici_shard::{
        config::default_global_doc_cache_size,
        reader::shard_reader::{DEFAULT_FRAG_SAMPLE_COUNT, DEFAULT_SHARD_SAMPLE_COUNT},
    };
    use tici_system::readable::ReadableSize;

    use super::*;

    #[test]
    fn test_new_with_data_path() -> Result<()> {
        let config = Config::default();
        assert_eq!(
            config.tici.frag_reader.global_doc_cache_size,
            default_global_doc_cache_size()
        );
        assert_eq!(
            config.tici.frag_reader.disk_reserve_size,
            ReadableSize::gb(10)
        );
        assert_eq!(
            config.tici.frag_reader.shard_sample_count,
            DEFAULT_SHARD_SAMPLE_COUNT
        );
        assert_eq!(
            config.tici.frag_reader.frag_sample_count,
            DEFAULT_FRAG_SAMPLE_COUNT
        );
        assert_eq!(config.tici.s3, S3Config::default());
        assert_eq!(config.tici.reader_node, ReaderNodeConfig::default());
        assert_eq!(config.flash, TiFlashServerConfig::default());
        assert_eq!(config.pd, PDConfig::default());
        Ok(())
    }

    #[test]
    fn test_load() -> Result<()> {
        let config = Config::load("../../ci/tiflash-reader-template.toml")?;
        assert_eq!(
            config.tici.logger.filename,
            "tici_searchlib.log".to_string()
        );
        assert_eq!(config.tici.reader_node.addr, "");
        assert_eq!(config.flash.service_addr, "");
        assert_eq!(config.pd.pd_addr, "");
        assert_eq!(config.logger.level, "");

        // validate should success
        assert!(config.validate().is_ok());
        Ok(())
    }

    #[test]
    fn test_resolve_addr() -> Result<()> {
        const EXAMPLE_CONFIG: &str = r#"
        [flash]
        service_addr = "compute-tiflash-yfxrk2.db-cluster.tidb12345:8520"
        [tici]
        [tici.reader-node]
        port = 1818
    "#;

        let mut config = Config::load_from_str(EXAMPLE_CONFIG)?;
        config.resolve_reader_node_addr()?;
        // check the resolved addr and advertise_addr
        assert_eq!(config.tici.reader_node.addr, "0.0.0.0:1818");
        assert_eq!(
            config.tici.reader_node.advertise_addr,
            "compute-tiflash-yfxrk2.db-cluster.tidb12345:1818"
        );
        Ok(())
    }

    #[test]
    fn parse_configs() -> Result<()> {
        const EXAMPLE_CONFIG: &str = r#"
        [flash]
        service_addr = "10.0.9.1:9000"
        [raft]
        pd_addr = "10.0.1.1:2379,10.0.1.2:2379,10.0.1.3:2379"
        [logger]
        level = "debug"
        some-unknown-field = "should be ignored"

        [tici]
        [tici.s3]
        endpoint = "http://localhost:9000"
        region = "us-east-1"
        access-key = "minioadmin"
        secret-key = "minioadmin"
        use-path-style = true

        [tici.reader-node]
        addr = "10.0.9.1:8520"
        port = 8520
        advertise-addr = "reader-node.example.com:8520"
        heartbeat-interval = "5s"
        max-heartbeat-retries = 5
        heartbeat-worker-count = 8
        heartbeat-queue-size = 1024

        [tici.frag-reader]
        global-doc-cache-size = "128MB"
        remove-zero-ref-frag-delay = "300s"
        disk-reserve-size = "12GB"
        shard-sample-count = 5
        frag-sample-count = 7

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

        [tici.logger]
        filename = "tici_searchlib.log"
        level = "info"

        [tici.storage]
        data-dir = "/var/lib/tici/searchlib"
    "#;
        let config = Config::load_from_str(EXAMPLE_CONFIG)?;
        assert_eq!(config.flash.service_addr, "10.0.9.1:9000");
        assert_eq!(
            config.pd.get_pd_addresses(),
            vec![
                "10.0.1.1:2379".to_string(),
                "10.0.1.2:2379".to_string(),
                "10.0.1.3:2379".to_string(),
            ]
        );
        assert_eq!(config.logger.level, "debug".to_string());

        // Check tici.s3
        assert_eq!(config.tici.s3.endpoint, "http://localhost:9000");
        assert_eq!(config.tici.s3.region, "us-east-1");
        assert_eq!(config.tici.s3.access_key, "minioadmin");
        assert_eq!(config.tici.s3.secret_key, "minioadmin");
        assert!(config.tici.s3.use_path_style);

        // Check tici.reader_node
        assert_eq!(config.tici.reader_node.addr, "10.0.9.1:8520");
        assert_eq!(config.tici.reader_node.port, 8520);
        assert_eq!(
            config.tici.reader_node.advertise_addr,
            "reader-node.example.com:8520"
        );
        assert_eq!(
            config.tici.reader_node.heartbeat_interval,
            ReadableDuration::secs(5)
        );
        assert_eq!(config.tici.reader_node.max_heartbeat_retries, 5);
        assert_eq!(config.tici.reader_node.heartbeat_worker_count, 8);
        assert_eq!(config.tici.reader_node.heartbeat_queue_size, 1024);

        // Check tici.frag_reader
        assert_eq!(
            config.tici.frag_reader.global_doc_cache_size,
            ReadableSize::mb(128)
        );
        assert_eq!(
            config.tici.frag_reader.remove_zero_ref_frag_delay,
            ReadableDuration::secs(300)
        );
        assert_eq!(
            config.tici.frag_reader.disk_reserve_size,
            ReadableSize::gb(12)
        );
        assert_eq!(config.tici.frag_reader.shard_sample_count, 5);
        assert_eq!(config.tici.frag_reader.frag_sample_count, 7);

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

        // Check tici.logger
        assert_eq!(config.tici.logger.filename, "tici_searchlib.log");
        assert_eq!(config.tici.logger.get_level_or(""), "info");

        assert_eq!(config.tici.storage.data_dir, "/var/lib/tici/searchlib");
        Ok(())
    }
}