// Copyright 2025 PingCAP, Inc.

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

use api_version::ApiV2;
use kvproto::kvrpcpb::Op;
use rfengine::RFENGINE_DFS_WORKER_HEALTHY_GAUGE;
use test_cloud_server::{
    ServerCluster, alloc_node_id_vec,
    client::{MutateOptions, PrewriteExt, TxnMutations},
    must_wait,
    oss::prepare_dfs,
    util::Mutation,
};
use test_pd_client::PdClientExt;
use tikv::config::TikvConfig;
use tikv_util::{config::ReadableSize, info};

use crate::cases::{i_to_key, i_to_keyspace_key, random_value_1kb};

fn new_lock_mutation(key: Vec<u8>) -> Mutation {
    Mutation {
        key: key.into(),
        value: vec![].into(),
        op: Op::Lock,
        ..Default::default()
    }
}

fn new_shared_lock_mutation(key: Vec<u8>) -> Mutation {
    Mutation {
        key: key.into(),
        value: vec![].into(),
        op: Op::SharedLock,
        ..Default::default()
    }
}

#[test]
fn test_rfengine_recover_from_unhealthy() {
    const KEYSPACE_ID: u32 = 1;
    let s3fs_put_wal_chunks_error_fp = "s3fs_put_wal_chunks_error";

    let (_temp_dir, mut oss, dfs_config) = prepare_dfs("t");

    // Create a cluster with a single store. As we use the metric
    // RFENGINE_DFS_WORKER_HEALTHY_GAUGE to detect the unhealthy, which is shared by
    // all stores.
    let mut cluster = ServerCluster::new(alloc_node_id_vec(1), |_, conf: &mut TikvConfig| {
        conf.dfs = dfs_config.clone();
        conf.rfengine.target_file_size = ReadableSize::kb(512);
        conf.rfengine.wal_chunk_target_file_size = ReadableSize::kb(32);
        conf.rfengine.wal_sync_dir = "${data-dir}/raft-wal".to_string();
        conf.rfengine.lightweight_backup = true;
    });
    cluster.wait_region_replicated(&[], 1);
    let mut client = cluster.new_client();
    client.split_keyspace(KEYSPACE_ID);

    let gen_key = i_to_keyspace_key(KEYSPACE_ID);
    client.put_kv(0..1000, &gen_key, random_value_1kb);
    client.verify_data_with_ref_store();

    // Workload to continuously generate WAL.
    let (tx, rx) = std::sync::mpsc::sync_channel(0);
    let workload_handle = {
        let mut client = cluster.new_client();
        std::thread::spawn(move || {
            while let Err(std::sync::mpsc::TryRecvError::Empty) = rx.try_recv() {
                let _ = client.try_put_kv(
                    10000..10050,
                    &gen_key,
                    random_value_1kb,
                    MutateOptions::default(),
                );
                let _ = client.try_del_kv(10000..10050, &gen_key, MutateOptions::default());
            }
        })
    };

    fail::cfg(s3fs_put_wal_chunks_error_fp, "return").unwrap();
    must_wait(
        || RFENGINE_DFS_WORKER_HEALTHY_GAUGE.get() == 0,
        30,
        || "wait for dfs worker unhealthy timeout".to_string(),
    );
    info!("DFS worker is unhealthy");

    fail::remove(s3fs_put_wal_chunks_error_fp);
    must_wait(
        || RFENGINE_DFS_WORKER_HEALTHY_GAUGE.get() == 1,
        30,
        || "wait for dfs worker healthy timeout".to_string(),
    );
    info!("DFS worker is healthy");

    tx.send(()).unwrap();
    workload_handle.join().unwrap();

    client.verify_data_with_ref_store();
    cluster.stop();
    oss.shutdown();
}

// Background (Linux durability model):
// - `fsync(2)` docs: https://man7.org/linux/man-pages/man2/fsync.2.html
// - `fsync` on a regular file persists file content/metadata, but does not
//   guarantee durability of the parent directory entry (filename -> inode).
// - To make a newly created filename durable, the parent directory must be
//   fsynced.
// - In this codebase, `sync_dir` is that parent-directory fsync helper
//   (`components/file_system/src/lib.rs::sync_dir`).
//
// Bug window under test:
// - rfengine compaction does: write `.rlog` -> fsync `.rlog` data
//   (`sync_files`) -> `sync_dir` -> write+fsync MANIFEST.
// - If `sync_dir` fails and the error is swallowed, compaction may still
//   write+fsync MANIFEST that references the new `.rlog`.
// - If crash happens before any later successful directory fsync, restart can
//   see MANIFEST durable but `.rlog` filename missing (ENOENT), which can
//   surface as `EntriesUnavailable` and recovery panic.
//
// Why this is important:
// - Probability is low but real (transient storage fault + crash timing).
// - Impact is high (data-loss risk for that log range, recovery panic, node
//   unavailability).
//
// Test steps:
// 1) Set up two failpoint callbacks, both gated by a compact counter:
//    - `compact_before_sync_dir`: On first compaction (count==1), inject
//      `sync_dir_error` to simulate a transient storage fault.
//    - `compact_after_manifest_persist`: On first compaction (count==1), delete
//      the target peer's `.rlog` file to simulate post-crash state.
// 2) Trigger multiple compactions by writing data. Only the first compaction
//    has the fault injected.
// 3) Clean up failpoints and restart the target node.
//
// Expected behavior:
// - Bug case (sync_dir error ignored): First compaction continues despite
//   sync_dir failure, MANIFEST is persisted. After rlog deletion, restart finds
//   MANIFEST references to missing file and panics.
// - Fixed case (sync_dir error propagated): First compaction fails early,
//   MANIFEST is never written. The `compact_after_manifest_persist` callback
//   never fires (count stays 0). Restart succeeds because MANIFEST is clean.
//
// Scope note:
// - We cannot deterministically force OS power-loss/page-cache loss in a
//   failpoint test, so step (2) uses state-equivalent surgery.
// - Other rfengine `sync_dir` call sites already propagate errors (`engine.rs`,
//   `writer.rs`, `manifest.rs`); this compact path is the one under test.
#[test]
fn test_rlog_loss_from_ignored_sync_dir_error() {
    const COMPACT_BEFORE_SYNC_DIR_FP: &str = "rfengine::compact_before_sync_dir";
    const COMPACT_AFTER_MANIFEST_PERSIST_FP: &str = "rfengine::compact_after_manifest_persist";
    const SYNC_DIR_ERROR_FP: &str = "file_system::sync_dir_error";

    let nodes = alloc_node_id_vec(3);
    let mut cluster = ServerCluster::new(nodes.clone(), |_, conf: &mut TikvConfig| {
        conf.raft_store.raft_log_gc_count_limit = Some(1_000_000);
        conf.raft_store.raft_log_gc_threshold = 100_000;
        conf.raft_store.raft_log_reserve_max_ticks = 1_000_000;
        // Force frequent compaction/offload with tiny rfengine file/memory limits.
        conf.rfengine.target_file_size = ReadableSize::kb(2);
        conf.rfengine.rlog_soft_memory_limit = ReadableSize::kb(8);
    });
    cluster.wait_region_replicated(&[], nodes.len());
    cluster.get_pd_client().disable_default_operator();

    let node_store_ids: Vec<(u16, u64)> = nodes
        .iter()
        .map(|&node_id| (node_id, cluster.get_store_id(node_id)))
        .collect();
    let mut client = cluster.new_client();
    client.put_kv(0..1, crate::cases::i_to_key, |_| b"v0000".to_vec());

    let region = client.get_region_by_key(&[]);
    let leader_store_id = region.get_leader().get_store_id();
    let target_store_id = region
        .peers
        .iter()
        .find_map(|peer| {
            let store_id = peer.get_store_id();
            (store_id != leader_store_id).then_some(store_id)
        })
        .unwrap();
    let target_peer = region
        .peers
        .iter()
        .find(|peer| peer.get_store_id() == target_store_id)
        .cloned()
        .unwrap();
    let target_peer_id = target_peer.get_id();
    let target_peer_prefix = format!("{:016x}_", target_peer_id);
    cluster
        .get_pd_client()
        .must_transfer_leader(region.id, target_peer);

    let target_node_id = node_store_ids
        .iter()
        .find_map(|&(node_id, store_id)| (store_id == target_store_id).then_some(node_id))
        .unwrap();
    let lagging_node_id = node_store_ids
        .iter()
        .find_map(|&(node_id, store_id)| {
            (store_id != leader_store_id && store_id != target_store_id).then_some(node_id)
        })
        .unwrap();
    let target_rfengine_dir = cluster
        .get_rfengine(target_node_id)
        .db_dir()
        .unwrap()
        .to_path_buf();

    // Compact counter: track how many times compact() has been called
    let compact_count = Arc::new(AtomicUsize::new(0));

    // Step 1: compact_before_sync_dir
    // - Increment compact counter
    // - Only inject sync_dir_error when count==1
    let count_clone1 = compact_count.clone();
    fail::cfg_callback(COMPACT_BEFORE_SYNC_DIR_FP, move || {
        let count = count_clone1.fetch_add(1, Ordering::SeqCst) + 1;
        if count == 1 {
            fail::cfg(SYNC_DIR_ERROR_FP, "1*return").unwrap();
        }
    })
    .unwrap();

    // Step 2: compact_after_manifest_persist
    // - Only delete rlog when count==1 (first compact that reaches here)
    let count_clone2 = compact_count.clone();
    let rfengine_dir_clone = target_rfengine_dir.clone();
    let peer_prefix_clone = target_peer_prefix.clone();
    fail::cfg_callback(COMPACT_AFTER_MANIFEST_PERSIST_FP, move || {
        let count = count_clone2.load(Ordering::SeqCst);
        if count != 1 {
            return;
        }

        // Delete target peer's rlog files (simulate data loss)
        let entries = match fs::read_dir(rfengine_dir_clone.as_path()) {
            Ok(entries) => entries,
            Err(_) => {
                return;
            }
        };

        for entry in entries {
            let entry = match entry {
                Ok(entry) => entry,
                Err(_) => continue,
            };
            let file_name = entry.file_name();
            let file_name_str = file_name.to_string_lossy();

            if file_name_str.ends_with(".rlog") && file_name_str.starts_with(&peer_prefix_clone) {
                fs::remove_file(entry.path()).unwrap();
                return;
            }
        }
    })
    .unwrap();

    // Trigger multiple compactions by writing data
    cluster.stop_node(lagging_node_id);
    let large_value = vec![b'x'; 1024];
    client.put_kv(1..1000, crate::cases::i_to_key, |_| large_value.clone());
    must_wait(
        || compact_count.load(Ordering::SeqCst) > 0,
        30,
        || "wait for compact failpoint to trigger timeout".to_string(),
    );

    // Cleanup failpoints
    fail::remove(SYNC_DIR_ERROR_FP);
    fail::remove(COMPACT_BEFORE_SYNC_DIR_FP);
    fail::remove(COMPACT_AFTER_MANIFEST_PERSIST_FP);

    // Restart target node. With the fix, the failed compaction should not leave
    // a manifest that references the deleted rlog file.
    cluster.stop_node(target_node_id);
    cluster.start_node(target_node_id, |_, _| {});
    client.verify_data_with_ref_store();
    cluster.stop();
}

#[test]
fn test_shared_lock_logs_survive_compaction_and_restart() {
    const COMPACT_BEFORE_SYNC_DIR_FP: &str = "rfengine::compact_before_sync_dir";

    test_util::init_log_for_test();

    let compact_count = Arc::new(AtomicUsize::new(0));
    let compact_count_clone = compact_count.clone();
    fail::cfg_callback(COMPACT_BEFORE_SYNC_DIR_FP, move || {
        compact_count_clone.fetch_add(1, Ordering::SeqCst);
    })
    .unwrap();
    tikv_util::defer!({
        fail::remove(COMPACT_BEFORE_SYNC_DIR_FP);
    });

    let nodes = alloc_node_id_vec(1);
    let node_id = nodes[0];
    let mut cluster = ServerCluster::new(nodes, |_, conf: &mut TikvConfig| {
        conf.pessimistic_txn.pipelined = false;
        conf.raft_store.raft_log_gc_count_limit = Some(1_000_000);
        conf.raft_store.raft_log_gc_threshold = 100_000;
        conf.raft_store.raft_log_reserve_max_ticks = 1_000_000;
        conf.rfengine.target_file_size = ReadableSize::kb(2);
        conf.rfengine.rlog_soft_memory_limit = ReadableSize::kb(8);
    });
    cluster.wait_region_replicated(&[], 1);

    let mut client = cluster.new_client();
    let keyspace_id = ApiV2::get_u32_keyspace_id_by_key(&i_to_key(0)).unwrap();
    client.split_keyspace(keyspace_id);

    let key = i_to_key(240);
    let pk1 = i_to_key(241);
    let pk2 = i_to_key(242);

    let start_ts_1 = client.get_ts();
    let for_update_ts_1 = client.get_ts();
    client
        .kv_pessimistic_lock(
            pk1.clone().into(),
            vec![pk1.clone().into()],
            start_ts_1,
            3000,
            for_update_ts_1,
            None,
        )
        .unwrap();
    client
        .kv_shared_pessimistic_lock(
            pk1.clone().into(),
            vec![key.clone().into()],
            start_ts_1,
            3000,
            for_update_ts_1,
            None,
        )
        .unwrap();

    let start_ts_2 = client.get_ts();
    let for_update_ts_2 = client.get_ts();
    client
        .kv_pessimistic_lock(
            pk2.clone().into(),
            vec![pk2.clone().into()],
            start_ts_2,
            3000,
            for_update_ts_2,
            None,
        )
        .unwrap();
    client
        .kv_shared_pessimistic_lock(
            pk2.clone().into(),
            vec![key.clone().into()],
            start_ts_2,
            3000,
            for_update_ts_2,
            None,
        )
        .unwrap();

    let mut txn_1 = client.begin_transaction(Some(start_ts_1));
    client
        .kv_prewrite_ext(
            pk1.clone().into(),
            Some(&vec![key.clone().into()]),
            TxnMutations::from_normal(vec![
                new_lock_mutation(pk1.clone()),
                new_shared_lock_mutation(key.clone()),
            ]),
            &mut txn_1,
            PrewriteExt {
                for_update_ts: for_update_ts_1,
                ..Default::default()
            },
        )
        .unwrap();
    let mut txn_2 = client.begin_transaction(Some(start_ts_2));
    client
        .kv_prewrite_ext(
            pk2.clone().into(),
            Some(&vec![key.clone().into()]),
            TxnMutations::from_normal(vec![
                new_lock_mutation(pk2.clone()),
                new_shared_lock_mutation(key.clone()),
            ]),
            &mut txn_2,
            PrewriteExt {
                for_update_ts: for_update_ts_2,
                ..Default::default()
            },
        )
        .unwrap();

    let commit_ts_1 = client.get_ts();
    client
        .kv_commit(
            TxnMutations::from_normal(vec![
                new_lock_mutation(pk1.clone()),
                new_shared_lock_mutation(key.clone()),
            ]),
            start_ts_1,
            commit_ts_1,
            false,
        )
        .unwrap();

    let compact_count_before_noise = compact_count.load(Ordering::SeqCst);
    let large_value = vec![b'x'; 1024];
    client.put_kv(1000..3000, i_to_key, |_| large_value.clone());
    must_wait(
        || compact_count.load(Ordering::SeqCst) > compact_count_before_noise,
        30,
        || "wait for rfengine compaction after shared-lock traffic timeout".to_string(),
    );

    cluster.restart_node(node_id, Duration::from_secs(0), false, |_, _| {});
    thread::sleep(Duration::from_secs(1));
    cluster.wait_region_replicated(&key, 1);

    let mut restarted_client = cluster.new_client();
    let start_ts_3 = restarted_client.get_ts();
    let for_update_ts_3 = restarted_client.get_ts();
    let (_results, key_errors) = restarted_client
        .kv_pessimistic_lock(
            key.clone().into(),
            vec![key.clone().into()],
            start_ts_3,
            3000,
            for_update_ts_3,
            Some(1),
        )
        .unwrap();
    assert_eq!(key_errors.len(), 1);
    assert!(
        key_errors[0].has_locked(),
        "expected persisted shared lock after compaction and restart"
    );

    let commit_ts_2 = restarted_client.get_ts();
    restarted_client
        .kv_commit(
            TxnMutations::from_normal(vec![
                new_lock_mutation(pk2.clone()),
                new_shared_lock_mutation(key.clone()),
            ]),
            start_ts_2,
            commit_ts_2,
            false,
        )
        .unwrap();

    let start_ts_4 = restarted_client.get_ts();
    let for_update_ts_4 = restarted_client.get_ts();
    let (_results, key_errors) = restarted_client
        .kv_pessimistic_lock(
            key.clone().into(),
            vec![key.clone().into()],
            start_ts_4,
            3000,
            for_update_ts_4,
            None,
        )
        .unwrap();
    assert!(key_errors.is_empty());

    let mvcc_info = restarted_client
        .kv_get_mvcc_by_key(&key)
        .expect("kv_get_mvcc_by_key failed")
        .expect("mvcc info missing");
    let writes = mvcc_info.get_writes();
    assert!(
        writes
            .iter()
            .any(|w| { w.get_start_ts() == start_ts_1.into_inner() && w.get_type() == Op::Lock }),
        "missing first shared-lock write after restart"
    );
    assert!(
        writes
            .iter()
            .any(|w| { w.get_start_ts() == start_ts_2.into_inner() && w.get_type() == Op::Lock }),
        "missing second shared-lock write after restart"
    );
    cluster.stop();
}