// Copyright 2022 PingCAP, Inc.

use std::{path::Path, sync::Arc};

use kvengine::dfs::{DFSConfig, Dfs, FileType, Options, new_dfs_from_config};
use kvproto::metapb::Store;
use pd_client::util::get_all_stores_except_tiflash;
use security::{GetSecurityManager, SecurityConfig, SecurityManager};

#[derive(Clone, Serialize, Deserialize, PartialEq, Debug, Default)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct CommonConfig {
    pub pd: pd_client::Config,
    pub security: SecurityConfig,
    pub dfs: DFSConfig,
}

impl CommonConfig {
    pub fn from_args(path: &Path, pd: &str) -> Self {
        let mut config = Self::default();
        if path.exists() {
            let data = std::fs::read(path).expect("failed to read config file");
            config = toml::from_slice(&data).unwrap();
        } else if !path.as_os_str().is_empty() {
            panic!("config file not found");
        }
        if !pd.is_empty() {
            config.pd.endpoints = pd.split(',').map(|x| x.to_owned()).collect();
        }
        // override from ENV
        config.dfs.override_from_env();
        config.security.override_from_env();
        config
    }

    pub fn create_context(&self) -> CtlContext {
        let dfs = new_dfs_from_config(self.dfs.clone());
        let mut pd_cfg = self.pd.clone();
        // cse-ctl is a one-shot CLI tool, fail fast instead of retrying forever.
        pd_cfg.retry_max_count = 3;
        let pd_client = Arc::new(native_br::common::create_pd_client(&self.security, &pd_cfg));
        let security_mgr = pd_client.get_security_mgr();
        let http_client = security_mgr.http_client(hyper::Client::builder()).unwrap();
        CtlContext {
            dfs,
            security_mgr,
            pd_client,
            http_client,
        }
    }
}

#[derive(Clone)]
pub struct CtlContext {
    pub dfs: Arc<dyn Dfs>,
    pub security_mgr: Arc<SecurityManager>,
    pub pd_client: Arc<pd_client::RpcClient>,
    pub http_client: security::HttpClient,
}

impl CtlContext {
    pub fn get_tikv_stores(&self, exclude_store_arg: &str) -> Vec<Store> {
        let excludes: Vec<u64> = exclude_store_arg
            .split(',')
            .filter_map(|x| x.parse().ok())
            .collect();
        let mut stores = get_all_stores_except_tiflash(self.pd_client.as_ref()).unwrap();
        stores.retain(|store| !excludes.contains(&store.get_id()));
        stores
    }
}

pub fn get_file_data_from_dfs(id: u64, file_type: FileType, config: &DFSConfig) -> bytes::Bytes {
    let dfs = new_dfs_from_config(config.clone());
    let runtime = dfs.get_runtime();
    runtime
        .block_on(dfs.read_file(id, Options::default().with_type(file_type)))
        .expect("failed to read file from dfs")
}