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

use api_version::ApiV2;
use rand::Rng;
use security::SecurityConfig;
use tikv_util::info;
use tokio::sync::oneshot::{Receiver, Sender};

mod gc;
mod test_async_fetch;
mod test_async_io;
mod test_builtin_dfs;
mod test_busy_apply_peers;
mod test_cdc;
mod test_cold_region;
mod test_compaction;
mod test_disk_usage;
mod test_load_data;
mod test_merge;
mod test_native_br;
mod test_raft_router_liveness;
mod test_read_pool_task_details;
mod test_remote_coprocessor;
mod test_rfengine;
mod test_rfstore_online_config;
mod test_split_region;
mod test_transaction;
mod test_trim_over_bound;

mod test_local_file_gc;
mod test_sst_io_error;

mod test_in_memory_pessimistic_lock;

pub use test_cloud_server::{alloc_node_id, alloc_node_id_vec};
use tidb_query_datatype::codec::table;

fn i_to_key(i: usize) -> Vec<u8> {
    format!("xkey{:08}", i).into_bytes()
}

fn i_to_keyspace_key(keyspace_id: u32) -> impl Fn(usize) -> Vec<u8> {
    move |i: usize| -> Vec<u8> {
        let mut key = ApiV2::get_keyspace_prefix_by_id(keyspace_id);
        key.extend(format!("key{:08}", i).into_bytes());
        key
    }
}

fn i_to_val(i: usize) -> Vec<u8> {
    format!("val{:08}", i).into_bytes().repeat(10)
}

fn random_value<const N: usize>(_: usize) -> Vec<u8>
where
    [u8; N]: rand::Fill,
{
    let mut bytes = [0u8; N];
    rand::thread_rng().fill(&mut bytes);
    bytes.to_vec()
}

fn random_value_1kb(i: usize) -> Vec<u8> {
    random_value::<1024>(i)
}

fn i_to_row_key(keyspace_prefix: &[u8], table_id: i64, i: usize) -> Vec<u8> {
    let mut key = keyspace_prefix.to_vec();
    key.extend(table::encode_row_key(table_id, i as i64));
    key
}

fn new_security_config() -> SecurityConfig {
    let mut conf = SecurityConfig::default();
    conf.master_key.vendor = "test".to_string();
    conf.master_key.key_id = "random".to_string();
    conf
}

pub(crate) struct FailpointGuard {
    names: Vec<&'static str>,
}

impl FailpointGuard {
    fn new() -> Self {
        Self { names: Vec::new() }
    }

    #[must_use]
    fn cfg(mut self, name: &'static str, actions: &str) -> Self {
        fail::cfg(name, actions).unwrap();
        self.names.push(name);
        self
    }

    #[must_use]
    fn cfg_callback<F>(mut self, name: &'static str, f: F) -> Self
    where
        F: Fn() + Send + Sync + 'static,
    {
        fail::cfg_callback(name, f).unwrap();
        self.names.push(name);
        self
    }
}

impl Drop for FailpointGuard {
    fn drop(&mut self) {
        for name in self.names.drain(..) {
            fail::remove(name);
        }
    }
}

pub(crate) fn cfg_notify_once(name: &'static str) -> (Receiver<Sender<()>>, FailpointGuard) {
    let name_clone = name.to_string();
    let (tx, rx) = tokio::sync::oneshot::channel();
    let pack = std::sync::Mutex::new(Some(tx));
    let guard = FailpointGuard::new().cfg_callback(name, move || {
        info!("injecting failpoint: {}", name_clone);
        let mut v = pack.lock().unwrap();
        if v.is_none() {
            return;
        }
        let tx = v.take().unwrap();
        drop(v);
        let (tx2, rx) = tokio::sync::oneshot::channel();

        let _ = tx.send(tx2);
        tokio::task::block_in_place(|| {
            let _ = rx.blocking_recv();
        })
    });

    (rx, guard)
}

pub(crate) fn cfg_notify_after(
    name: &'static str,
    skipped_hits: usize,
) -> (Receiver<Sender<()>>, FailpointGuard) {
    let name_clone = name.to_string();
    let (tx, rx) = tokio::sync::oneshot::channel();
    let pack = std::sync::Mutex::new(Some(tx));
    let hit_count = std::sync::atomic::AtomicUsize::new(0);
    let guard = FailpointGuard::new().cfg_callback(name, move || {
        if hit_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst) < skipped_hits {
            return;
        }
        info!(
            "injecting failpoint after {} hits: {}",
            skipped_hits, name_clone
        );
        let mut v = pack.lock().unwrap();
        let Some(tx) = v.take() else {
            return;
        };
        drop(v);
        let (resume_tx, resume_rx) = tokio::sync::oneshot::channel();
        let _ = tx.send(resume_tx);
        tokio::task::block_in_place(|| {
            let _ = resume_rx.blocking_recv();
        })
    });

    (rx, guard)
}