// Copyright 2026 PingCAP, Inc.

use std::{
    sync::{
        Arc,
        atomic::{AtomicUsize, Ordering},
    },
    thread,
    time::Duration,
};

use bytes::{BufMut, BytesMut};
use cloud_server::status_server::SstMeta;
use http::{Method, Request, StatusCode};
use hyper::{Body, Client};
use kvengine::{Engine, WRITE_CF, WRITE_CF_BOTTOM_LEVEL, dfs::DFSConfig};
use kvproto::errorpb;
use pd_client::PdClient;
use protobuf::Message;
use serde_json::{Value as JsonValue, json};
use tempfile::TempDir;
use test_cloud_server::{
    ServerCluster, ServerClusterBuilder, TikvWorkerOptions, alloc_node_id_vec,
    oss::ObjectStorageService, util::request_major_compact_on_store,
};
use tikv::config::TikvConfig;

use crate::{alloc_node_id, i_to_key_with_keyspace, i_to_val};

const NODES_SIZE: usize = 3;
const KEYSPACE_ID: u32 = 42;

/// Test cluster context that owns all necessary resources.
/// Fields are ordered to ensure correct destruction order - later fields drop
/// first.
struct TestCluster {
    // Must be kept alive for the entire test
    _base_dir: TempDir,
    _oss: ObjectStorageService,

    // Cluster and its metadata (public for direct access)
    pub cluster: ServerCluster,
    pub nodes: Vec<u16>,
}

/// Setup a basic test cluster
fn setup_test_cluster(test_name: &str) -> TestCluster {
    let base_dir = tempfile::Builder::new()
        .prefix(test_name)
        .tempdir()
        .unwrap();

    // Start mock S3 service
    let oss_dir = base_dir.path().join("oss");
    let mut oss = ObjectStorageService::new(oss_dir.clone());
    oss.start_server();

    // Configure DFS
    let dfs_config = DFSConfig {
        prefix: test_name.to_string(),
        s3_endpoint: format!("http://127.0.0.1:{}", oss.port()),
        s3_key_id: "admin".to_string(),
        s3_secret_key: "admin".to_string(),
        s3_bucket: test_name.to_string(),
        s3_region: "local".to_string(),
        zstd_compression_level: "3".to_string(),
        ..Default::default()
    };

    // Start TiKV cluster
    let nodes: Vec<_> = (0..NODES_SIZE).map(|_| alloc_node_id()).collect();
    let mut cluster = ServerClusterBuilder::new(nodes.clone(), |_, conf: &mut TikvConfig| {
        conf.dfs = dfs_config.clone();
    })
    .pd_server_cnt(1)
    .tikv_worker_cnt(1)
    .build();

    cluster.start_tikv_workers(alloc_node_id_vec(1), TikvWorkerOptions::default());
    cluster.wait_region_replicated(&[], NODES_SIZE);

    // /write_sst requires API v2 keys with a keyspace prefix. Split regions so
    // the keyspace range is properly bounded; otherwise the worker cannot fetch
    // shard meta for the keyspace.
    {
        let mut client = cluster.new_client();
        client.split_keyspace(KEYSPACE_ID);
    }
    let keyspace_probe_key = i_to_key_with_keyspace(KEYSPACE_ID)(0);
    cluster.wait_region_replicated(&keyspace_probe_key, NODES_SIZE);

    TestCluster {
        _base_dir: base_dir,
        _oss: oss,
        cluster,
        nodes,
    }
}

/// Prepare SST data for a key range.
fn prepare_sst_data(start: usize, end: usize, value_offset: usize) -> Vec<(Vec<u8>, Vec<u8>)> {
    let i_to_key = i_to_key_with_keyspace(KEYSPACE_ID);
    (start..end)
        .map(|i| (i_to_key(i), i_to_val(i + value_offset)))
        .collect()
}

/// Verify data correctness by reading and comparing values.
fn verify_data(tikv_client: &mut crate::ClusterClient, kvs: &[(Vec<u8>, Vec<u8>)]) {
    for (key, expected_val) in kvs {
        let (val, _) = tikv_client.must_get_key(key, tikv_util::time::Instant::now());
        assert_eq!(val, *expected_val);
    }
}

/// Check for same-level overlap and assert no overlap exists.
fn check_and_assert_no_overlap(engine: &Engine, region_id: u64) {
    let target_level = WRITE_CF_BOTTOM_LEVEL as usize;
    let all_levels = engine.get_cf_levels_files(region_id, WRITE_CF);

    // Check if the target level exists and has files
    if target_level >= all_levels.len() || all_levels[target_level].is_empty() {
        return; // No files at this level, nothing to check
    }

    let files = &all_levels[target_level];

    // Check for overlapping files at the same level
    // Two ranges overlap if: start1 <= end2 && start2 <= end1
    for i in 0..files.len() {
        for j in (i + 1)..files.len() {
            let (id1, start1, end1) = &files[i];
            let (id2, start2, end2) = &files[j];

            let has_overlap = start1.as_ref() <= end2.as_ref() && start2.as_ref() <= end1.as_ref();

            assert!(
                !has_overlap,
                "Same-level overlap detected in L{}: File {} [{:?}..{:?}] overlaps with File {} [{:?}..{:?}], violates LSM invariants.\nAll L{} files: {:?}",
                target_level,
                id1,
                start1.as_ref(),
                end1.as_ref(),
                id2,
                start2.as_ref(),
                end2.as_ref(),
                target_level,
                files
                    .iter()
                    .map(|(id, s, b)| (id, s.as_ref(), b.as_ref()))
                    .collect::<Vec<_>>()
            );
        }
    }
}

// Helper: write SST via tikv-worker /write_sst endpoint
fn write_sst(
    cluster: &test_cloud_server::ServerCluster,
    kvs: &[(Vec<u8>, Vec<u8>)],
    commit_ts: u64,
) -> SstMeta {
    let pd_client = cluster.get_pd_client();
    let dfs = cluster.get_dfs().unwrap();
    let runtime = dfs.get_runtime();
    let cluster_id = pd_client.get_cluster_id().unwrap();

    // Encode data in write_sst format
    let mut body_buf = BytesMut::new();
    for (key, val) in kvs {
        body_buf.put_u16_le(key.len() as u16);
        body_buf.put_slice(key);
        body_buf.put_u32_le(val.len() as u32);
        body_buf.put_slice(val);
    }
    let body_bytes = body_buf.freeze();

    // Send write_sst request
    let write_url = format!(
        "http://{}/write_sst?cluster_id={}&commit_ts={}",
        cluster.tikv_worker_endpoints()[0],
        cluster_id,
        commit_ts
    );

    let http_client = Client::new();
    let req = Request::builder()
        .method(Method::PUT)
        .uri(&write_url)
        .header("Content-Type", "application/octet-stream")
        .body(Body::from(body_bytes))
        .unwrap();

    let resp = runtime.block_on(http_client.request(req)).unwrap();
    assert_eq!(resp.status(), StatusCode::OK);

    let bytes = runtime
        .block_on(hyper::body::to_bytes(resp.into_body()))
        .unwrap();

    let resp_body: JsonValue = serde_json::from_slice(&bytes).unwrap();
    let sst_meta_json = resp_body.get("sst_meta").unwrap();
    serde_json::from_value(sst_meta_json.clone()).unwrap()
}

// Helper: ingest SST via TiKV status server /ingest_s3 endpoint
fn ingest_sst(
    cluster: &test_cloud_server::ServerCluster,
    nodes: &[u16],
    sst_meta: &SstMeta,
    target_key: &[u8],
) {
    let pd_client = cluster.get_pd_client();
    let dfs = cluster.get_dfs().unwrap();
    let runtime = dfs.get_runtime();
    let cluster_id = pd_client.get_cluster_id().unwrap();

    let region = pd_client.get_region(target_key).unwrap();
    let region_id = region.get_id();
    let epoch = region.get_region_epoch().clone();

    let node_id = nodes[0];
    let status_addr = cluster.status_addr(node_id);
    let ingest_url = format!(
        "http://{}/ingest_s3?cluster_id={}&region_id={}&epoch_version={}",
        status_addr,
        cluster_id,
        region_id,
        epoch.get_version()
    );

    let http_client = Client::new();
    for attempt in 0..100 {
        let ingest_req = Request::builder()
            .method(Method::POST)
            .uri(&ingest_url)
            .header("Content-Type", "application/json")
            .body(Body::from(json!(sst_meta).to_string()))
            .unwrap();

        let ingest_resp = runtime.block_on(http_client.request(ingest_req)).unwrap();
        let status = ingest_resp.status();
        let bytes = runtime
            .block_on(hyper::body::to_bytes(ingest_resp.into_body()))
            .unwrap();
        if status == StatusCode::OK {
            return;
        }

        let mut error = errorpb::Error::default();
        error.merge_from_bytes(&bytes).unwrap();
        if (error.has_region_not_initialized() || error.has_not_leader()) && attempt < 99 {
            thread::sleep(Duration::from_millis(20));
            continue;
        }
        panic!("ingest failed with status {:?}: {:?}", status, error);
    }
}

/// Send concurrent ingest requests and return responses.
fn send_concurrent_ingest_requests(
    status_addr: String,
    cluster_id: u64,
    region_id: u64,
    epoch_version: u64,
    sst: SstMeta,
    count: usize,
) -> Vec<(usize, StatusCode, String)> {
    let mut handles = Vec::new();
    for i in 0..count {
        let status_addr_clone = status_addr.clone();
        let sst_clone = sst.clone();

        let handle = thread::spawn(move || {
            let runtime = tokio::runtime::Runtime::new().unwrap();
            let http_client = Client::new();
            let ingest_url = format!(
                "http://{}/ingest_s3?cluster_id={}&region_id={}&epoch_version={}",
                status_addr_clone, cluster_id, region_id, epoch_version
            );

            let req_body = json!(sst_clone).to_string();
            let ingest_req = Request::builder()
                .method(Method::POST)
                .uri(&ingest_url)
                .header("Content-Type", "application/json")
                .body(Body::from(req_body))
                .unwrap();

            let ingest_resp = runtime.block_on(http_client.request(ingest_req)).unwrap();
            let status = ingest_resp.status();
            let bytes = runtime
                .block_on(hyper::body::to_bytes(ingest_resp.into_body()))
                .unwrap();
            let resp_text = String::from_utf8_lossy(&bytes).to_string();

            (i, status, resp_text)
        });
        handles.push(handle);
    }

    handles.into_iter().map(|h| h.join().unwrap()).collect()
}

/// Send a single ingest request and return (status, response_text).
fn send_ingest_request(
    cluster: &ServerCluster,
    nodes: &[u16],
    pd_client: &impl PdClient,
    runtime: &tokio::runtime::Runtime,
    cluster_id: u64,
    target_key: &[u8],
    sst: &SstMeta,
) -> (StatusCode, String) {
    let region = pd_client.get_region(target_key).unwrap();
    let region_id = region.get_id();
    let epoch = region.get_region_epoch().clone();
    let status_addr = cluster.status_addr(nodes[0]);
    let ingest_url = format!(
        "http://{}/ingest_s3?cluster_id={}&region_id={}&epoch_version={}",
        status_addr,
        cluster_id,
        region_id,
        epoch.get_version()
    );

    let req_body = json!(sst).to_string();
    let http_client = Client::new();
    let ingest_req = Request::builder()
        .method(Method::POST)
        .uri(&ingest_url)
        .header("Content-Type", "application/json")
        .body(Body::from(req_body))
        .unwrap();

    let ingest_resp = runtime.block_on(http_client.request(ingest_req)).unwrap();
    let status = ingest_resp.status();
    let bytes = runtime
        .block_on(hyper::body::to_bytes(ingest_resp.into_body()))
        .unwrap();
    let resp_text = String::from_utf8_lossy(&bytes).to_string();

    (status, resp_text)
}

/// Retry ingest until success, return (first_attempt_got_server_is_busy,
/// total_server_is_busy_count).
fn retry_ingest_until_success(
    cluster: &ServerCluster,
    nodes: &[u16],
    pd_client: &impl PdClient,
    runtime: &tokio::runtime::Runtime,
    cluster_id: u64,
    target_key: &[u8],
    sst: &SstMeta,
) -> (bool, u32) {
    let mut first_attempt_got_server_is_busy = false;
    let mut total_server_is_busy_count = 0;

    // Upper bound the total wait time to keep the test reasonably fast.
    // Compaction should normally finish within a few retries (on the order of
    // seconds).
    for attempt in 0..=8 {
        let (status, resp_text) = send_ingest_request(
            cluster, nodes, pd_client, runtime, cluster_id, target_key, sst,
        );

        if status == StatusCode::OK {
            break;
        }

        assert_eq!(
            status,
            StatusCode::INTERNAL_SERVER_ERROR,
            "Unexpected status at attempt {}: body={}",
            attempt + 1,
            resp_text
        );

        let is_server_is_busy = resp_text.contains("ServerIsBusy")
            || resp_text.contains("server_is_busy")
            || resp_text.contains("compaction in progress");

        assert!(
            is_server_is_busy,
            "Unexpected error at attempt {}: status={}, body={}",
            attempt + 1,
            status,
            resp_text
        );

        total_server_is_busy_count += 1;
        if attempt == 0 {
            first_attempt_got_server_is_busy = true;
        }

        assert!(
            attempt < 8,
            "Ingest failed after 9 attempts, ServerIsBusy count: {}",
            total_server_is_busy_count
        );

        // Wait for compaction to complete before retry.
        thread::sleep(Duration::from_millis(500));
    }

    (first_attempt_got_server_is_busy, total_server_is_busy_count)
}

// =============================================================================
// Tests (ordered from simple to complex)
// =============================================================================

/// Test normal ingest workflow when no overlap exists.
///
/// When ingest data has no overlap with existing files, get_ingest_level()
/// returns a level > 0 (typically L3). In this case, no compaction is triggered
/// and ingest succeeds immediately. All files should be placed at the
/// bottommost level (L3).
#[test]
fn test_ingest_without_overlap() {
    let test_ctx = setup_test_cluster("test_ingest_without_overlap");
    let cluster = &test_ctx.cluster;
    let nodes = &test_ctx.nodes;
    let mut tikv_client = cluster.new_client();

    // Ingest non-overlapping data - should succeed immediately at L3
    let data = prepare_sst_data(0, 100, 0);
    let target_key = &data[0].0;
    let sst = write_sst(cluster, &data, 200);
    ingest_sst(cluster, nodes, &sst, target_key);
    verify_data(&mut tikv_client, &data);

    // Ingest another non-overlapping range - should succeed immediately at L3
    // This tests the else branch (ingest_level > 0), no compaction triggered
    let data_new = prepare_sst_data(200, 250, 0);
    let sst_new = write_sst(cluster, &data_new, 300);
    ingest_sst(cluster, nodes, &sst_new, &data_new[0].0);
    verify_data(&mut tikv_client, &data_new);

    // Verify only L3 has files (no overlap, so all files go to bottommost level)
    let pd_client = cluster.get_pd_client();
    let region = pd_client.get_region(target_key).unwrap();
    let region_id = region.get_id();
    let engine = cluster.get_kvengine(nodes[0]);
    let levels_files = engine.get_cf_levels_files(region_id, WRITE_CF);
    assert!(
        !levels_files[WRITE_CF_BOTTOM_LEVEL as usize].is_empty(),
        "L3 should have files"
    );
    assert!(
        levels_files[0].is_empty(),
        "L0 should be empty (no overlap, no compaction triggered)"
    );
    assert!(
        levels_files[1].is_empty(),
        "L1 should be empty (no overlap, no compaction triggered)"
    );
    assert!(
        levels_files[2].is_empty(),
        "L2 should be empty (no overlap, no compaction triggered)"
    );
}

/// Test "Compact on Ingest Overlap" workflow when ingest would go to L0.
///
/// When get_ingest_level() returns 0 (ingest would go to L0), we trigger
/// compaction to compact all data to L3, allowing retry to ingest at L2 or L3.
///
/// T1: First ingest to L3 (no overlap)
/// T2: Second ingest to L2 (overlaps with L3)
/// T3: Third ingest to L1 (overlaps with L2). L1 can ingest normally,
///       no compaction triggered.
/// T4: Fourth ingest overlaps with L1, ingest_level==0 triggers compaction,
///       returns ServerIsBusy.
/// T5: Client retries, compaction completes, ingest succeeds.
#[test]
fn test_ingest_to_l0_triggers_compaction() {
    let test_ctx = setup_test_cluster("test_ingest_to_l0_triggers_compaction");
    let cluster = &test_ctx.cluster;
    let nodes = &test_ctx.nodes;
    let mut tikv_client = cluster.new_client();
    let pd_client = cluster.get_pd_client();
    let dfs = cluster.get_dfs().unwrap();
    let runtime = dfs.get_runtime();

    // T1: First ingest to L3 (no overlap)
    let data = prepare_sst_data(0, 100, 0);
    let target_key = &data[0].0;
    let sst_1 = write_sst(cluster, &data, 200);
    ingest_sst(cluster, nodes, &sst_1, target_key);
    verify_data(&mut tikv_client, &data);

    let region = pd_client.get_region(target_key).unwrap();
    let region_id = region.get_id();
    let engine = cluster.get_kvengine(nodes[0]);
    let levels_files = engine.get_cf_levels_files(region_id, WRITE_CF);
    assert!(
        !levels_files[WRITE_CF_BOTTOM_LEVEL as usize].is_empty(),
        "First ingest should create file at L3"
    );

    // T2: Second ingest to L2 (overlaps with L3)
    let sst_2 = write_sst(cluster, &data, 201);
    ingest_sst(cluster, nodes, &sst_2, target_key);
    // Wait for ingest to complete and file to be created.
    thread::sleep(Duration::from_millis(500));
    let levels_files = engine.get_cf_levels_files(region_id, WRITE_CF);
    assert!(
        !levels_files[2].is_empty(),
        "Second ingest should create file at L2"
    );

    // T3: Third ingest to L1 (overlaps with L2). L1 can ingest normally,
    //       no compaction triggered. This step ensures L1 has files so next
    //       ingest will trigger L0.
    // NOTE: Each SST must use a unique version to avoid triggering the
    //       debug_assert in MergeIterator::fix(). See #3985 for details.
    // TODO: Add a test case for same-version ingest after #3985 is resolved.
    let sst_2_5 = write_sst(cluster, &data, 202);
    ingest_sst(cluster, nodes, &sst_2_5, target_key);
    // Wait for ingest to complete and file to be created.
    thread::sleep(Duration::from_millis(500));
    let levels_files = engine.get_cf_levels_files(region_id, WRITE_CF);
    assert!(
        !levels_files[1].is_empty(),
        "Third ingest should create file at L1"
    );

    // T4: Fourth ingest triggers L0 overlap detection. get_ingest_level checks
    //       L1 first, finds overlap, returns 0. Since ingest_level==0,
    //       compaction is triggered (L0 cannot ingest), returns ServerIsBusy.
    let sst_3 = write_sst(cluster, &data, 202);
    // Wait before first attempt to ensure previous operations are stable.
    thread::sleep(Duration::from_millis(500));

    // T5: Client retries until compaction completes and ingest succeeds
    let cluster_id = pd_client.get_cluster_id().unwrap();
    let (first_attempt_got_server_is_busy, _total_server_is_busy_count) =
        retry_ingest_until_success(
            cluster,
            nodes,
            pd_client.as_ref(),
            runtime,
            cluster_id,
            target_key,
            &sst_3,
        );

    // Verify first attempt received ServerIsBusy (proves compaction was triggered)
    assert!(
        first_attempt_got_server_is_busy,
        "First attempt did not receive ServerIsBusy - compaction may not have been triggered"
    );
    verify_data(&mut tikv_client, &data);

    // Verify compaction completed: L2 and L3 should have files after compaction
    let levels_files = engine.get_cf_levels_files(region_id, WRITE_CF);
    assert!(
        !levels_files[2].is_empty(),
        "L2 should have files after compaction"
    );
    assert!(
        !levels_files[WRITE_CF_BOTTOM_LEVEL as usize].is_empty(),
        "L3 should have files after compaction"
    );
}

/// Test concurrent ingest requests that all trigger L0 overlap.
///
/// When multiple ingest requests arrive concurrently and all detect
/// ingest_level==0, only one compaction is triggered (guaranteed by Raft's
/// linearizability). All requests receive ServerIsBusy.
///
/// T1: Setup data so the next ingest hits L0 overlap.
/// T2: Fire 5 concurrent ingest_s3 requests; each proposes
///     MANUAL_MAJOR_COMPACTION because there is no in-progress gate.
/// T3: Verify all 5 requests receive ServerIsBusy.
/// T4: Verify CompactionRunner deduped to exactly 1 major compaction.
/// T5: Release the failpoint and verify follow-up ingest succeeds.
#[test]
fn test_concurrent_ingest_trigger_single_compaction() {
    let test_ctx = setup_test_cluster("test_concurrent_ingest_trigger_single_compaction");
    let cluster = &test_ctx.cluster;
    let nodes = &test_ctx.nodes;
    let pd_client = cluster.get_pd_client();
    let dfs = cluster.get_dfs().unwrap();
    let runtime = dfs.get_runtime();

    // Counter increments every time trigger_major_compaction reaches the
    // failpoint. Pause 200ms so that all 5 concurrent CompactMsg::Compact
    // messages reach CompactionRunner and get deduped before the first
    // compaction finishes.
    let major_compaction_count = Arc::new(AtomicUsize::new(0));
    let count_clone = major_compaction_count.clone();
    fail::cfg_callback("major_compaction_after_input_determined", move || {
        count_clone.fetch_add(1, Ordering::SeqCst);
        thread::sleep(Duration::from_millis(200));
    })
    .unwrap();

    // T1: Setup data so the next ingest hits L0 overlap.
    let data = prepare_sst_data(0, 100, 0);
    let target_key = &data[0].0;
    let sst_1 = write_sst(cluster, &data, 200);
    ingest_sst(cluster, nodes, &sst_1, target_key);

    let sst_2 = write_sst(cluster, &data, 201);
    ingest_sst(cluster, nodes, &sst_2, target_key);
    thread::sleep(Duration::from_millis(500));

    let sst_3 = write_sst(cluster, &data, 202);
    ingest_sst(cluster, nodes, &sst_3, target_key);
    thread::sleep(Duration::from_millis(500));

    // T2: Fire 5 concurrent ingest_s3 requests; each one proposes a
    // MANUAL_MAJOR_COMPACTION change set because there is no in-progress
    // gate.
    let cluster_id = pd_client.get_cluster_id().unwrap();
    let sst_4 = write_sst(cluster, &data, 203);
    let status_addr = cluster.status_addr(nodes[0]);
    let region = pd_client.get_region(target_key).unwrap();
    let region_id = region.get_id();
    let epoch_version = region.get_region_epoch().get_version();

    let responses = send_concurrent_ingest_requests(
        status_addr,
        cluster_id,
        region_id,
        epoch_version,
        sst_4,
        5,
    );

    // T3: Every response should be ServerIsBusy. The status server serializes
    // the protobuf error so the body carries the reason string we set in
    // handle_l0_ingest_overlap rather than the literal `ServerIsBusy` tag.
    let server_is_busy_count = responses
        .iter()
        .filter(|(_, status, body)| {
            *status == StatusCode::INTERNAL_SERVER_ERROR
                && (body.contains("ServerIsBusy")
                    || body.contains("server_is_busy")
                    || body.contains("compaction in progress"))
        })
        .count();
    assert_eq!(
        server_is_busy_count, 5,
        "All 5 concurrent requests should receive ServerIsBusy, got responses: {:?}",
        responses,
    );

    // T4: Give the runner enough time to drain the CompactMsg queue and
    // verify only one major compaction was actually triggered.
    thread::sleep(Duration::from_secs(2));
    let triggered = major_compaction_count.load(Ordering::SeqCst);
    assert_eq!(
        triggered, 1,
        "exactly 1 major compaction should be triggered, but {} were triggered",
        triggered,
    );

    // T5: Release the failpoint and check that the next ingest succeeds.
    fail::remove("major_compaction_after_input_determined");
    let sst_5 = write_sst(cluster, &data, 204);
    let _ = retry_ingest_until_success(
        cluster,
        nodes,
        pd_client.as_ref(),
        runtime,
        cluster_id,
        target_key,
        &sst_5,
    );
}

/// Test ingest behavior when compaction fails.
///
/// When compaction fails (e.g., due to disk full), subsequent ingest requests
/// should continue to receive ServerIsBusy until compaction succeeds.
///
/// T1: Setup data to trigger L0.
/// T2: Configure failpoint to make compaction fail.
/// T3: Trigger ingest that would go to L0, compaction triggered but fails.
/// T4: Verify subsequent ingest requests still receive ServerIsBusy.
/// T5: Remove failpoint, compaction succeeds, ingest can proceed.
#[test]
fn test_ingest_when_compaction_fails() {
    let test_ctx = setup_test_cluster("test_ingest_when_compaction_fails");
    let cluster = &test_ctx.cluster;
    let nodes = &test_ctx.nodes;
    let mut tikv_client = cluster.new_client();
    let pd_client = cluster.get_pd_client();
    let dfs = cluster.get_dfs().unwrap();
    let runtime = dfs.get_runtime();

    // T1: Setup data to trigger L0 (L1 has files)
    let data = prepare_sst_data(0, 100, 0);
    let target_key = &data[0].0;
    let sst_1 = write_sst(cluster, &data, 200);
    ingest_sst(cluster, nodes, &sst_1, target_key);

    // Create L2 and L1 files so next ingest will trigger L0
    let sst_2 = write_sst(cluster, &data, 201);
    ingest_sst(cluster, nodes, &sst_2, target_key);
    thread::sleep(Duration::from_millis(500));

    let sst_3 = write_sst(cluster, &data, 202);
    ingest_sst(cluster, nodes, &sst_3, target_key);
    thread::sleep(Duration::from_millis(500));

    // T2: Configure failpoint to make compaction fail
    fail::cfg("kvengine_compact", "return").unwrap();

    // T3: Trigger ingest that would go to L0 → compaction triggered but fails
    let sst_4 = write_sst(cluster, &data, 203);
    let cluster_id = pd_client.get_cluster_id().unwrap();
    let (status, resp_text) = send_ingest_request(
        cluster,
        nodes,
        pd_client.as_ref(),
        runtime,
        cluster_id,
        target_key,
        &sst_4,
    );

    // First request should receive ServerIsBusy
    assert!(
        status == StatusCode::INTERNAL_SERVER_ERROR
            && (resp_text.contains("ServerIsBusy")
                || resp_text.contains("server_is_busy")
                || resp_text.contains("compaction in progress")),
        "First ingest should receive ServerIsBusy: status={}, body={}",
        status,
        resp_text
    );

    // T4: Verify subsequent ingest requests still receive ServerIsBusy
    thread::sleep(Duration::from_millis(500));
    for attempt in 0..3 {
        let (status, resp_text) = send_ingest_request(
            cluster,
            nodes,
            pd_client.as_ref(),
            runtime,
            cluster_id,
            target_key,
            &sst_4,
        );
        assert!(
            status == StatusCode::INTERNAL_SERVER_ERROR
                && (resp_text.contains("ServerIsBusy")
                    || resp_text.contains("server_is_busy")
                    || resp_text.contains("compaction in progress")),
            "Attempt {} should still receive ServerIsBusy while compaction fails: status={}, body={}",
            attempt + 1,
            status,
            resp_text
        );
        thread::sleep(Duration::from_millis(500));
    }

    // T5: Remove failpoint → compaction succeeds → ingest can proceed
    fail::remove("kvengine_compact");
    // Wait for compaction to complete
    thread::sleep(Duration::from_secs(2));

    // Retry ingest - should succeed now
    let (_first_attempt_got_server_is_busy, _) = retry_ingest_until_success(
        cluster,
        nodes,
        pd_client.as_ref(),
        runtime,
        cluster_id,
        target_key,
        &sst_4,
    );
    verify_data(&mut tikv_client, &data);
}

// =============================================================================
// Ingest-Compaction Race Tests
//
// When ingest and compaction race on the same region, there are two
// theoretical scenarios that could lead to same-level file overlap in the
// KvEngine LSM-tree:
//   - Scenario 1: Compaction proposes → Ingest picks level and proposes →
//     Compaction applies → Ingest applies. Timeline:
//       1. Compaction proposes its changeset to Raft (enters Raft log)
//       2. Ingest picks target level L3 in peer_fsm based on current LSM state
//          (compaction not yet applied)
//       3. Ingest proposes its changeset to Raft (enters Raft log)
//       4. Compaction applies in apply thread, merging files and changing L3's
//          structure
//       5. Ingest applies in apply thread to L3 with outdated level choice
//          - Result: Creates overlap if compaction's output overlaps ingest's
//            range
//   - Scenario 2: Compaction picks inputs → Ingest propose + applies →
//     Compaction applies. Timeline:
//       1. Compaction picks input files based on current LSM state
//       2. Before compaction applies, ingest runs and applies to L3
//          - A new file appears in L3
//       3. Compaction propose + applies using its old input selection,
//          outputting to L3
//          - Result: Compaction's output could overlap with the newly ingested
//            file
// =============================================================================

// TODO: Add test for Scenario 1 after Raft message filter is implemented.
//
// This test requires Raft message filtering to control message delivery timing:
// 1. Compaction proposes to Raft
// 2. Filter blocks Raft AppendEntries messages for compaction
// 3. Ingest picks level L3 (sees old state with gap) + proposes to Raft
// 4. Unblock compaction → compaction commit -> compaction applies, fills gap at
//    L3
// 5. Ingest applies → overlap detection triggers
//
// Without message filter, we cannot reliably control the timing to create this
// race condition.

/// Test Scenario 2: Compaction picks inputs before ingest applies.
///
/// Setup: Create gap [50, 100) in L3 with files [0, 50) and [100, 150)
///
/// Execution:
/// 1. Start major compaction, pause at major_compaction_after_input_determined
///    (after input selection)
/// 2. While compaction is paused, ingest [50, 100) starts and completes
/// 3. Resume compaction, which proposes using outdated input selection
///
/// Result:
/// This scenario CAN happen. Overlap detection prevents the race.
#[test]
fn test_ingest_compaction_race_when_ingest_apply_first() {
    let test_ctx = setup_test_cluster("test_ingest_compaction_race_when_ingest_apply_first");
    let cluster = &test_ctx.cluster;
    let nodes = &test_ctx.nodes;
    let mut tikv_client = cluster.new_client();
    let pd_client = cluster.get_pd_client();
    let dfs = cluster.get_dfs().unwrap();
    let runtime = dfs.get_runtime();

    // 1. Prepare SST data: [0, 50), [50, 100), [100, 150)
    let data_0_50 = prepare_sst_data(0, 50, 0);
    let data_50_100 = prepare_sst_data(50, 100, 1000);
    let data_100_150 = prepare_sst_data(100, 150, 0);

    // 2. Write SST files
    let sst_0_50 = write_sst(cluster, &data_0_50, 100);
    let sst_50_100 = write_sst(cluster, &data_50_100, 200);
    let sst_100_150 = write_sst(cluster, &data_100_150, 101);

    // 3. Ingest [0, 50) and [100, 150) → L3 has gap [50, 100)
    ingest_sst(cluster, nodes, &sst_0_50, &data_0_50[0].0);
    ingest_sst(cluster, nodes, &sst_100_150, &data_100_150[0].0);
    verify_data(&mut tikv_client, &data_0_50);
    verify_data(&mut tikv_client, &data_100_150);

    let region = pd_client.get_region(&data_0_50[0].0).unwrap();
    let region_id = region.get_id();
    let stores = pd_client.get_all_stores(true).unwrap();
    let engine = cluster.get_kvengine(nodes[0]);

    // 4. Setup failpoint to pause compaction after input determination
    fail::cfg("major_compaction_after_input_determined", "pause").unwrap();

    // 5. Trigger compaction in background (will pause after selecting input files)
    let compaction_task = {
        let stores = stores.clone();
        runtime.spawn(async move {
            for store in stores.iter() {
                request_major_compact_on_store(
                    store,
                    &format!("major_compact=true&region_id={}", region_id),
                    false,
                )
                .await;
            }
        })
    };
    // Wait for compaction to reach failpoint (paused after input selection)
    thread::sleep(Duration::from_millis(500));

    // 6. Ingest [50, 100) while compaction is paused L3 now has: [0, 50), [50,
    //    100), [100, 150)
    ingest_sst(cluster, nodes, &sst_50_100, &data_50_100[0].0);

    // 7. Resume compaction, merges [0, 50) + [100, 150) → [0, 150) L3 now has: [0,
    //    150) and [50, 100) → OVERLAP!
    fail::remove("major_compaction_after_input_determined");
    runtime.block_on(async {
        tokio::time::timeout(Duration::from_secs(10), compaction_task)
            .await
            .expect("compaction timed out")
            .expect("compaction failed");
    });
    // Wait for compaction to be fully applied to shard meta
    thread::sleep(Duration::from_secs(1));

    // 8. Assert no overlap exists - the detection mechanism should have prevented
    //    it
    check_and_assert_no_overlap(&engine, region_id);

    // 9. Verify meta_committed was called by checking shard.compacting is false If
    //    meta_committed was NOT called when overlap detected, shard.compacting
    //    would remain true and block future compactions
    let is_compacting = engine.is_shard_compacting(region_id);
    assert_eq!(
        is_compacting,
        Some(false),
        "shard.compacting should be false after rejected compaction, \
         indicating meta_committed was properly called"
    );
}