// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
// Modifications Copyright 2022 PingCAP, Inc.

use std::{
    collections::HashMap,
    fs::File,
    io::{Read, Write},
    sync::{Arc, Mutex},
};

use online_config::{ConfigChange, OnlineConfig};
use raftstore::store::Config as RaftstoreConfig;
use tikv::config::*;
use tikv_util::config::ReadableDuration;

fn change(name: &str, value: &str) -> HashMap<String, String> {
    let mut m = HashMap::new();
    m.insert(name.to_owned(), value.to_owned());
    m
}

#[test]
fn test_update_config() {
    let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap();
    cfg.validate().unwrap();
    let cfg_controller = ConfigController::new(cfg);
    let mut cfg = cfg_controller.get_current();

    // normal update
    cfg_controller
        .update(change("raftstore.raft-log-gc-threshold", "2000"))
        .unwrap();
    cfg.raft_store.raft_log_gc_threshold = 2000;
    cfg_controller
        .update(change("storage.max-ts.max-drift", "365s"))
        .unwrap();
    cfg.storage.max_ts.max_drift = ReadableDuration::secs(365);
    assert_eq!(cfg_controller.get_current(), cfg);

    // update that fails the validation
    assert!(
        cfg_controller
            .update(change("storage.max-ts.max-drift", "3s"))
            .unwrap_err()
            .to_string()
            .contains("smaller than or equal to storage.max-ts.cache-sync-interval")
    );

    // update not support config
    let res = cfg_controller.update(change("server.addr", "localhost:3000"));
    res.unwrap_err();
    let res = cfg_controller.update(change("storage.max-ts.cache-sync-interval", "3s"));
    res.unwrap_err();
    assert_eq!(cfg_controller.get_current(), cfg);

    // update to invalid config
    let res = cfg_controller.update(change("raftstore.raft-log-gc-threshold", "0"));
    res.unwrap_err();
    assert_eq!(cfg_controller.get_current(), cfg);

    // bad update request
    let res = cfg_controller.update(change("xxx.yyy", "0"));
    res.unwrap_err();
    let res = cfg_controller.update(change("raftstore.xxx", "0"));
    res.unwrap_err();
    let res = cfg_controller.update(change("raftstore.raft-log-gc-threshold", "10MB"));
    res.unwrap_err();
    let res = cfg_controller.update(change("raft-log-gc-threshold", "10MB"));
    res.unwrap_err();
    assert_eq!(cfg_controller.get_current(), cfg);
}

#[test]
fn test_update_config_txn_file_lock_limit() {
    let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap();
    cfg.validate().unwrap();
    let cfg_controller = ConfigController::new(cfg);

    cfg_controller
        .update(change("storage.max-txn-file-locks-per-region", "0"))
        .unwrap();
    assert_eq!(
        cfg_controller
            .get_current()
            .storage
            .max_txn_file_locks_per_region,
        0
    );

    cfg_controller
        .update(change(
            "storage.scheduler-async-resolve-lock-batch-interval",
            "0s",
        ))
        .unwrap();
    assert_eq!(
        cfg_controller
            .get_current()
            .storage
            .scheduler_async_resolve_lock_batch_interval,
        ReadableDuration::ZERO
    );
}

#[test]
fn test_dispatch_change() {
    use std::{error::Error, result::Result};

    use online_config::ConfigManager;

    #[derive(Clone)]
    struct CfgManager(Arc<Mutex<RaftstoreConfig>>);

    impl ConfigManager for CfgManager {
        fn dispatch(&mut self, c: ConfigChange) -> Result<(), Box<dyn Error>> {
            self.0.lock().unwrap().update(c)
        }
    }

    let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap();
    cfg.validate().unwrap();
    let raftstore_cfg = TikvConfig::compatible_adjust_to_raftstore(&cfg.raft_store);
    let cfg_controller = ConfigController::new(cfg);
    let mut cfg = cfg_controller.get_current();
    let mgr = CfgManager(Arc::new(Mutex::new(raftstore_cfg.clone())));
    cfg_controller.register(Module::Raftstore, Box::new(mgr.clone()));

    cfg_controller
        .update(change("raftstore.raft-log-gc-threshold", "2000"))
        .unwrap();

    // config update
    cfg.raft_store.raft_log_gc_threshold = 2000;
    assert_eq!(cfg_controller.get_current(), cfg);

    // config change should also dispatch to raftstore config manager
    assert_eq!(mgr.0.lock().unwrap().raft_log_gc_threshold, 2000);
}

#[test]
fn test_dispatch_change_kvengine() {
    use std::time::Duration;

    use kvengine::{
        KvEngineConfig, Options, config_manager::KvEngineConfigManager, context::IaCtx,
    };
    use tikv_util::config::VersionTrack;

    let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap();
    cfg.validate().unwrap();
    let cfg_controller = ConfigController::new(cfg);

    let kvengine_config = KvEngineConfig::default();
    let ia_ctx = IaCtx::disabled(&kvengine_config.ia);
    let track = Arc::new(VersionTrack::new(kvengine_config));
    let opts = Arc::new(Options::default());
    let mgr = KvEngineConfigManager::new(track.clone(), ia_ctx, opts.clone());
    cfg_controller.register(Module::Kvengine, Box::new(mgr));

    // Each top-level kvengine online field routes through `/config`
    // end-to-end into the snapshot and any runtime side effects.
    cfg_controller
        .update(change("kvengine.remote-worker-addr", "http://worker:1"))
        .unwrap();
    assert_eq!(track.value().remote_worker_addr, "http://worker:1");

    cfg_controller
        .update(change("kvengine.remote-coprocessor-addr", "http://cop:2"))
        .unwrap();
    assert_eq!(track.value().remote_coprocessor_addr, "http://cop:2");

    cfg_controller
        .update(change(
            "kvengine.remote-coprocessor-min-blocks-size",
            "4096",
        ))
        .unwrap();
    assert_eq!(track.value().remote_coprocessor_min_blocks_size, 4096);

    cfg_controller
        .update(change("kvengine.remote-coprocessor-num-ranges", "8"))
        .unwrap();
    assert_eq!(track.value().remote_coprocessor_num_ranges, 8);

    cfg_controller
        .update(change(
            "kvengine.remote-coprocessor-min-process-duration",
            "200ms",
        ))
        .unwrap();
    assert_eq!(
        track.value().remote_coprocessor_min_process_duration.0,
        Duration::from_millis(200)
    );

    cfg_controller
        .update(change(
            "kvengine.remote-coprocessor-topn-max-pushdown-rows",
            "12345",
        ))
        .unwrap();
    assert_eq!(
        track.value().remote_coprocessor_topn_max_pushdown_rows,
        12_345
    );

    cfg_controller
        .update(change("kvengine.compaction-old-data-ratio", "0.5"))
        .unwrap();
    assert_eq!(track.value().compaction_old_data_ratio, 0.5);
    assert_eq!(opts.compaction_old_data_ratio(), 0.5);

    cfg_controller
        .update(change("kvengine.compaction-old-data-size", "128MB"))
        .unwrap();
    assert_eq!(track.value().compaction_old_data_size.0, 128 * 1024 * 1024);
    assert_eq!(opts.compaction_old_data_size(), 128 * 1024 * 1024);

    // Skip-attributed fields are rejected at the ConfigController layer.
    let snapshot_before = track.value().clone();
    let err = cfg_controller
        .update(change("kvengine.flush-split-l0", "true"))
        .unwrap_err();
    assert!(
        err.to_string().contains("can not be changed"),
        "expected reject for skip field, got: {}",
        err
    );
    assert_eq!(track.value().clone(), snapshot_before);

    // Unknown sub-key under kvengine is also rejected.
    cfg_controller
        .update(change("kvengine.no-such-knob", "1"))
        .unwrap_err();
}

#[test]
fn test_write_update_to_file() {
    let (mut cfg, tmp_dir) = TikvConfig::with_tmp().unwrap();
    cfg.cfg_path = tmp_dir.path().join("cfg_file").to_str().unwrap().to_owned();
    {
        let c = r#"
## comment should be reserve
[raftstore]

# config that comment out by one `#` should be update in place
## pd-heartbeat-tick-interval = "30s"
# pd-heartbeat-tick-interval = "30s"

[rocksdb.defaultcf]
## config should be update in place
block-cache-size = "10GB"

[rocksdb.lockcf]
## this config will not update even it has the same last 
## name as `rocksdb.defaultcf.block-cache-size`
block-cache-size = "512MB"

[coprocessor]
## the update to `coprocessor.region-split-keys`, which do not show up 
## as key-value pair after [coprocessor], will be written at the end of [coprocessor]

[gc]
## config should be update in place
max-write-bytes-per-sec = "1KB"

[rocksdb.defaultcf.titan]
blob-run-mode = "normal"
"#;
        let mut f = File::create(&cfg.cfg_path).unwrap();
        f.write_all(c.as_bytes()).unwrap();
        f.sync_all().unwrap();
    }
    let cfg_controller = ConfigController::new(cfg);
    let change = {
        let mut change = HashMap::new();
        change.insert(
            "raftstore.pd-heartbeat-tick-interval".to_owned(),
            "1h".to_owned(),
        );
        change.insert(
            "coprocessor.region-split-keys".to_owned(),
            "10000".to_owned(),
        );
        change.insert("gc.max-write-bytes-per-sec".to_owned(), "100MB".to_owned());
        change.insert(
            "rocksdb.defaultcf.block-cache-size".to_owned(),
            "1GB".to_owned(),
        );
        change.insert(
            "rocksdb.defaultcf.titan.blob-run-mode".to_owned(),
            "read-only".to_owned(),
        );
        change
    };
    cfg_controller.update(change).unwrap();
    let res = {
        let mut buf = Vec::new();
        let mut f = File::open(cfg_controller.get_current().cfg_path).unwrap();
        f.read_to_end(&mut buf).unwrap();
        buf
    };

    let expect = r#"
## comment should be reserve
[raftstore]

# config that comment out by one `#` should be update in place
## pd-heartbeat-tick-interval = "30s"
pd-heartbeat-tick-interval = "1h"

[rocksdb.defaultcf]
## config should be update in place
block-cache-size = "1GB"

[rocksdb.lockcf]
## this config will not update even it has the same last 
## name as `rocksdb.defaultcf.block-cache-size`
block-cache-size = "512MB"

[coprocessor]
## the update to `coprocessor.region-split-keys`, which do not show up 
## as key-value pair after [coprocessor], will be written at the end of [coprocessor]

region-split-keys = 10000
[gc]
## config should be update in place
max-write-bytes-per-sec = "100MB"

[rocksdb.defaultcf.titan]
blob-run-mode = "read-only"
"#;
    assert_eq!(expect.as_bytes(), res.as_slice());
}

#[test]
fn test_update_from_toml_file() {
    use std::{error::Error, result::Result};

    use online_config::ConfigManager;

    #[derive(Clone)]
    struct CfgManager(Arc<Mutex<RaftstoreConfig>>);

    impl ConfigManager for CfgManager {
        fn dispatch(&mut self, c: ConfigChange) -> Result<(), Box<dyn Error>> {
            self.0.lock().unwrap().update(c)
        }
    }

    let (cfg, _dir) = TikvConfig::with_tmp().unwrap();
    let cfg_controller = ConfigController::new(cfg);
    let cfg = cfg_controller.get_current();
    let mgr = CfgManager(Arc::new(Mutex::new(
        TikvConfig::compatible_adjust_to_raftstore(&cfg.raft_store),
    )));
    cfg_controller.register(Module::Raftstore, Box::new(mgr));

    // update config file
    let c = r#"
[raftstore]
raft-log-gc-threshold = 2000
"#;
    let mut f = File::create(&cfg.cfg_path).unwrap();
    f.write_all(c.as_bytes()).unwrap();
    // before update this configuration item should be the default value
    assert_eq!(
        cfg_controller
            .get_current()
            .raft_store
            .raft_log_gc_threshold,
        50
    );
    // config update from config file
    cfg_controller.update_from_toml_file().unwrap();
    // after update this configration item should be constant with the modified
    // configuration file
    assert_eq!(
        cfg_controller
            .get_current()
            .raft_store
            .raft_log_gc_threshold,
        2000
    );
}