// Copyright 2026 TiKV Project Authors. Licensed under Apache-2.0.

use std::{collections::HashMap, sync::Arc};

use bytes::Bytes;
use hyper::Method;
use security::{RestfulClient, SecurityConfig, SecurityManager};
use test_cloud_server::ServerCluster;

use crate::alloc_node_id_vec;

fn update_config(status_addr: String, key: &str, value: &str) {
    let security_mgr = Arc::new(SecurityManager::new(&SecurityConfig::default()).unwrap());
    let rt = tokio::runtime::Runtime::new().unwrap();
    let change = HashMap::from([(key.to_owned(), value.to_owned())]);
    rt.block_on(async move {
        RestfulClient::new("kvengine-config", vec![status_addr], security_mgr)
            .unwrap()
            .request(
                "config",
                Method::POST,
                Some(Bytes::from(serde_json::to_vec(&change).unwrap())),
            )
            .await
            .unwrap();
    });
}

fn get_config(status_addr: String) -> serde_json::Value {
    let security_mgr = Arc::new(SecurityManager::new(&SecurityConfig::default()).unwrap());
    let rt = tokio::runtime::Runtime::new().unwrap();
    rt.block_on(async move {
        let resp = RestfulClient::new("kvengine-config", vec![status_addr], security_mgr)
            .unwrap()
            .request("config", Method::GET, None)
            .await
            .unwrap();
        serde_json::from_slice(&resp).unwrap()
    })
}

#[test]
fn test_kvengine_online_config_via_http() {
    test_util::init_log_for_test();
    let node_ids = alloc_node_id_vec(1);
    let node_id = node_ids[0];
    let cluster = ServerCluster::new(node_ids, |_, _| {});
    let status_addr = cluster.status_addr(node_id);

    // Verify initial defaults.
    let cfg = get_config(status_addr.clone());
    assert_eq!(cfg["kvengine"]["remote-worker-addr"], "");
    assert_eq!(cfg["kvengine"]["remote-coprocessor-addr"], "");
    assert_eq!(cfg["kvengine"]["compaction-old-data-ratio"], 0.3);
    assert_eq!(cfg["kvengine"]["compaction-old-data-size"], "64MiB");

    // Update remote-worker-addr twice to verify repeated changes.
    update_config(
        status_addr.clone(),
        "kvengine.remote-worker-addr",
        "http://worker:1",
    );
    let cfg = get_config(status_addr.clone());
    assert_eq!(cfg["kvengine"]["remote-worker-addr"], "http://worker:1");

    update_config(
        status_addr.clone(),
        "kvengine.remote-worker-addr",
        "http://worker:2",
    );
    let cfg = get_config(status_addr.clone());
    assert_eq!(cfg["kvengine"]["remote-worker-addr"], "http://worker:2");

    // Clear to empty (operator local-fallback signal).
    update_config(status_addr.clone(), "kvengine.remote-worker-addr", "");
    let cfg = get_config(status_addr.clone());
    assert_eq!(cfg["kvengine"]["remote-worker-addr"], "");

    // Update the remaining online fields.
    update_config(
        status_addr.clone(),
        "kvengine.remote-coprocessor-addr",
        "http://cop:3",
    );
    update_config(
        status_addr.clone(),
        "kvengine.remote-coprocessor-min-blocks-size",
        "4096",
    );
    update_config(
        status_addr.clone(),
        "kvengine.remote-coprocessor-num-ranges",
        "16",
    );
    update_config(
        status_addr.clone(),
        "kvengine.remote-coprocessor-min-process-duration",
        "500ms",
    );
    update_config(
        status_addr.clone(),
        "kvengine.remote-coprocessor-topn-max-pushdown-rows",
        "12345",
    );
    update_config(
        status_addr.clone(),
        "kvengine.compaction-old-data-ratio",
        "0.5",
    );
    update_config(
        status_addr.clone(),
        "kvengine.compaction-old-data-size",
        "128MB",
    );
    let cfg = get_config(status_addr);
    assert_eq!(cfg["kvengine"]["remote-coprocessor-addr"], "http://cop:3");
    assert_eq!(cfg["kvengine"]["remote-coprocessor-min-blocks-size"], 4096);
    assert_eq!(cfg["kvengine"]["remote-coprocessor-num-ranges"], 16);
    assert_eq!(
        cfg["kvengine"]["remote-coprocessor-min-process-duration"],
        "500ms"
    );
    assert_eq!(
        cfg["kvengine"]["remote-coprocessor-topn-max-pushdown-rows"],
        12345
    );
    assert_eq!(cfg["kvengine"]["compaction-old-data-ratio"], 0.5);
    assert_eq!(cfg["kvengine"]["compaction-old-data-size"], "128MiB");
    let opts = &cluster.get_kvengine(node_id).opts;
    assert_eq!(opts.compaction_old_data_ratio(), 0.5);
    assert_eq!(opts.compaction_old_data_size(), 128 << 20);
}