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

use std::collections::HashMap;

use bytes::Bytes;
use http::StatusCode;
use hyper::{Body, Client, Method, Request};
use test_cloud_server::ServerCluster;
use tokio::runtime::Runtime;

use crate::alloc_node_id_vec;

async fn request_config(
    status_addr: &str,
    method: Method,
    change: Option<HashMap<&str, &str>>,
) -> (StatusCode, Bytes) {
    let body = change
        .map(|change| Body::from(serde_json::to_vec(&change).unwrap()))
        .unwrap_or_else(Body::empty);
    let request = Request::builder()
        .method(method)
        .uri(format!("http://{status_addr}/config"))
        .body(body)
        .unwrap();
    let response = Client::new().request(request).await.unwrap();
    let status = response.status();
    let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
    (status, body)
}

async fn get_config(status_addr: &str) -> serde_json::Value {
    let (status, body) = request_config(status_addr, Method::GET, None).await;
    assert_eq!(status, StatusCode::OK);
    serde_json::from_slice(&body).unwrap()
}

#[test]
fn test_rfengine_lightweight_backup_online_config() {
    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, |_, config| {
        config.rfengine.lightweight_backup = false;
    });
    let status_addr = cluster.status_addr(node_id);
    let engine = cluster.get_rfengine(node_id);
    let runtime = Runtime::new().unwrap();

    // GET keeps exposing the complete rfengine config even though only the
    // lightweight-backup field is online configurable.
    let config = runtime.block_on(get_config(&status_addr));
    assert_eq!(config["rfengine"]["lightweight-backup"], false);
    assert!(config["rfengine"].get("target-file-size").is_some());
    assert!(!engine.is_lightweight_backup_enabled());

    // Enable through the ordinary config endpoint and verify both the HTTP
    // contract and the DFS worker runtime state.
    let (status, body) = runtime.block_on(request_config(
        &status_addr,
        Method::POST,
        Some(HashMap::from([("rfengine.lightweight-backup", "true")])),
    ));
    assert_eq!(status, StatusCode::OK);
    assert!(body.is_empty());
    assert!(engine.is_lightweight_backup_enabled());
    let worker_status = runtime.block_on(engine.get_dfs_worker_status()).unwrap();
    assert!(worker_status.enabled);
    assert!(worker_status.running);
    assert_eq!(
        runtime.block_on(get_config(&status_addr))["rfengine"]["lightweight-backup"],
        true
    );

    // Disable follows the same path and waits until the worker has stopped.
    let (status, body) = runtime.block_on(request_config(
        &status_addr,
        Method::POST,
        Some(HashMap::from([("rfengine.lightweight-backup", "false")])),
    ));
    assert_eq!(status, StatusCode::OK);
    assert!(body.is_empty());
    assert!(!engine.is_lightweight_backup_enabled());
    let worker_status = runtime.block_on(engine.get_dfs_worker_status()).unwrap();
    assert!(!worker_status.enabled);
    assert!(!worker_status.running);
    assert_eq!(
        runtime.block_on(get_config(&status_addr))["rfengine"]["lightweight-backup"],
        false
    );

    // The former write endpoint is gone; runtime changes must use /config.
    let request = Request::post(format!("http://{status_addr}/rfengine/dfs_worker"))
        .header(http::header::CONTENT_TYPE, "application/json")
        .body(Body::from(r#"{"enabled":true}"#))
        .unwrap();
    let response = runtime.block_on(Client::new().request(request)).unwrap();
    assert_eq!(response.status(), StatusCode::NOT_FOUND);
    assert!(!engine.is_lightweight_backup_enabled());

    // Other rfengine fields remain restart-only and retain the endpoint's
    // existing update-failure response.
    let (status, body) = runtime.block_on(request_config(
        &status_addr,
        Method::POST,
        Some(HashMap::from([("rfengine.target-file-size", "1MB")])),
    ));
    assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
    assert!(String::from_utf8_lossy(&body).starts_with("failed to update, error:"));
    assert!(!engine.is_lightweight_backup_enabled());
}