// Copyright 2026 PingCAP, Inc.

use std::{
    io::Write,
    path::{Path, PathBuf},
    sync::atomic::Ordering,
    time::Duration,
};

use sqlx::ConnectOptions;
use test_cloud_server::{
    keyspace::KeyspaceManager,
    oss::prepare_dfs,
    sync_diff_inspector::SyncDiffer,
    ticdc::{CDC_BIN_ENV_KEY, CdcServer, CdcServerConfig, validate_cdc_binary},
    tidb::{ConnParams, LocalTidb, TidbCluster},
    tpc::Tpc,
};
use tikv_util::{info, time::Instant};

use crate::{Running, TPCC_COUNTER, test_tidb::*, test_tpc::check_tpc_binary, *};

const LOCAL_TIDB_HEALTHY_TIMEOUT: Duration = Duration::from_secs(60);
const WAIT_SYNC_TIMEOUT: Duration = Duration::from_secs(300);
const NO_PROGRESS_TIMEOUT: Duration = Duration::from_secs(180);
const SYNC_POINT_INTERVAL: Duration = Duration::from_secs(30);
const TPCC_WAREHOUSES: usize = 2;
const TPCC_MAX_PROCS: usize = 1;
const TPCC_THREADS: usize = 4;
const TPCC_RUN_DURATION: Duration = Duration::from_secs(10);
const TEST_DURATION: u64 = 120;

#[test]
fn test_random_cdc() {
    // Initialize the CDC random test workspace and runtime.
    init_logger();
    let _cdc_dir_guard = tempfile::Builder::new().prefix("cdc_").tempdir().unwrap();
    let cdc_dir = _cdc_dir_guard.path().to_path_buf();
    tikv_util::set_panic_hook(false, cdc_dir.to_str().unwrap());

    let prepare_time = Instant::now_coarse();
    let switches = normalized_cdc_switches();
    info!("cdc switches"; "switches" => ?switches);

    let keyspace_count = switches.cdc_keyspace_count;
    let test_duration = Duration::from_secs(env_param("TEST_DURATION", TEST_DURATION));
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .worker_threads(4)
        .thread_name("test_random_cdc")
        .build()
        .unwrap();
    let _guard = runtime.enter();

    // Validate the external binaries required by the CDC scenario.
    let cdc_bin = std::env::var(CDC_BIN_ENV_KEY).expect("env CDC_BIN is not set");
    validate_cdc_binary(&cdc_bin);
    let tidb_bin = PathBuf::from(
        std::env::var(crate::test_tidb::TIDB_BIN_ENV_KEY).expect("env TIDB_BIN is not set"),
    );
    let tpc_bin = std::env::var(crate::test_tidb::TPC_BIN_ENV_KEY).expect("env TPC_BIN is not set");
    check_tpc_binary(&tpc_bin);

    // Prepare DFS, PD, TiKV, and upstream TiDB state for all keyspaces.
    let (_temp_dir, _oss, dfs_config) = prepare_dfs("oss_");
    let security_conf = new_security_config();
    let tc = prepare_tidb_cluster(&security_conf, &switches, keyspace_count);
    let mut cluster = prepare_cluster(
        &dfs_config,
        &security_conf,
        NODES_COUNT,
        keyspace_count,
        &switches,
        &tc,
    );
    let pd_client = cluster.get_pd_client_ext();
    let pd_ctl = std::sync::Arc::new(cluster.get_pd_control().unwrap());
    let keyspace_manager = cluster.keyspace_manager().clone();

    // Start the CSE and upstream TiDB components before enabling CDC.
    let tikv_worker_addr = cluster.tikv_worker_endpoints().pop().unwrap();
    start_components(
        &tc,
        tikv_worker_addr,
        &switches,
        &dfs_config,
        &runtime,
        keyspace_count,
    );
    runtime.block_on(prepare_tidb_variables(&tc, &keyspace_manager, &switches));

    // Start a local downstream TiDB used as the CDC sink.
    let downstream = LocalTidb::start(
        tidb_bin,
        cdc_dir.join("downstream"),
        cdc_dir.clone(),
        switches.rep_log_level.clone(),
    );
    runtime.block_on(downstream.wait_healthy(LOCAL_TIDB_HEALTHY_TIMEOUT));
    runtime.block_on(downstream.init_gc_safe_point());

    // Start a standalone new-architecture TiCDC server for the test cluster.
    let cdc = CdcServer::start(
        cdc_dir.join("cdc"),
        cdc_dir.join("cdc.log"),
        cdc_bin,
        tc.pd.endpoints(),
        switches.rep_log_level.clone(),
        CdcServerConfig {
            enable_legacy_safepoint: switches.for_serverless,
        },
    );
    runtime.block_on(cdc.wait_healthy());

    // Build per-keyspace test cases and validate keyspace-scoped API behavior.
    let cases = build_keyspace_cases(&tc, &keyspace_manager, &cdc_dir, SYNC_POINT_INTERVAL);
    assert_eq!(
        cases.len(),
        keyspace_count,
        "CDC test expected {keyspace_count} keyspaces, got {:?}",
        cases.iter().map(|c| c.keyspace_id).collect::<Vec<_>>()
    );
    for case in &cases {
        let cdc_cli = cdc.cdc_cli(case.keyspace_name.clone());
        runtime.block_on(cdc_cli.validate_keyspace_query_contract());
    }

    let client = cluster.new_client();
    let downstream_conn = downstream.conn_params();
    let mut sync_differs = Vec::with_capacity(cases.len());
    for case in &cases {
        // Create each keyspace changefeed and prepare its TPCC and sync-diff inputs.
        let start_ts = client.get_ts().into_inner();
        let cdc_cli = cdc.cdc_cli(case.keyspace_name.clone());
        runtime.block_on(async {
            cdc_cli
                .create_changefeed(
                    &case.changefeed_id,
                    &downstream_conn,
                    start_ts,
                    &case.filter_rules,
                    case.sync_point_interval,
                )
                .await;
            cdc_cli.wait_changefeed_normal(&case.changefeed_id).await;
        });
        runtime.block_on(prepare_tpcc_for_keyspace(&keyspace_manager, &tpc_bin, case));
        sync_differs.push(SyncDiffer::new(
            case.sync_diff_dir.clone(),
            case.upstream_conn.clone(),
            downstream_conn.clone(),
            case.filter_rules.clone(),
            Some(case.syncpoint_changefeed()),
            case.sync_point_interval,
            switches.rep_log_level.clone(),
        ));
    }

    // Run TPCC workloads concurrently across all CDC keyspaces.
    info!("start CDC TPCC workloads"; "keyspace_count" => cases.len());
    let start_time = Instant::now();
    let running = Running::new_start();
    let handles = cases
        .iter()
        .cloned()
        .map(|case| {
            spawn_tpcc_for_keyspace(
                keyspace_manager.clone(),
                tpc_bin.clone(),
                case,
                running.clone(),
            )
        })
        .collect::<Vec<_>>();

    while start_time.saturating_elapsed() < test_duration {
        std::thread::sleep(Duration::from_secs(1));
    }

    // Stop workloads and verify TPCC data through the upstream TiDBs.
    info!("stop CDC TPCC workloads");
    running.stop();
    runtime.block_on(async {
        for handle in handles {
            handle.await.unwrap();
        }
    });
    for case in &cases {
        runtime.block_on(check_tpcc_for_keyspace(&keyspace_manager, &tpc_bin, case));
    }

    // Wait until every changefeed reaches the verification timestamp.
    let verify_ts = client.get_ts().into_inner();
    for case in &cases {
        let cdc_cli = cdc.cdc_cli(case.keyspace_name.clone());
        runtime.block_on(cdc_cli.wait_changefeed_checkpoint(
            &case.changefeed_id,
            verify_ts,
            WAIT_SYNC_TIMEOUT,
        ));
    }
    // Verify final downstream consistency with syncpoint-based sync-diff.
    for (case, sync_differ) in cases.iter().zip(sync_differs.iter()) {
        info!("wait sync-diff final consistency"; "keyspace_id" => case.keyspace_id, "verify_ts" => verify_ts);
        sync_differ.must_wait_sync_to(
            verify_ts,
            WAIT_SYNC_TIMEOUT,
            NO_PROGRESS_TIMEOUT,
            case.sync_point_interval,
        );
    }

    // Stop sync-diff workers and remove all CDC changefeeds.
    runtime.block_on(async {
        for sync_differ in &sync_differs {
            sync_differ.stop().await;
        }
    });
    for case in &cases {
        let cdc_cli = cdc.cdc_cli(case.keyspace_name.clone());
        runtime.block_on(async {
            cdc_cli.delete_changefeed(&case.changefeed_id).await;
            cdc_cli.wait_changefeed_removed(&case.changefeed_id).await;
        });
    }

    // Tear down CDC, TiDB, CSE, schedulers, and PD after final cluster
    // verification.
    cdc.stop();
    downstream.stop();
    runtime.block_on(async {
        check_and_stop_components(&tc).await;
        stop_schedulers(pd_ctl).await;
        verify_cluster(&mut cluster, &switches, &[]).await;
    });
    cluster.stop();
    let region_number = pd_client.get_regions_number();
    tc.pd.stop_all();

    // Print the random-test success summary.
    let stats = WorkloadStats::collect();
    let stdout = std::io::stdout();
    writeln!(
        stdout.lock(),
        "{} TEST SUCCEED: elapsed {:?},{:?}, keyspace_count {}, region_number {}, {:?}",
        test_id(),
        prepare_time.saturating_elapsed(),
        start_time.saturating_elapsed(),
        keyspace_count,
        region_number,
        stats,
    )
    .unwrap();
    stdout.lock().flush().unwrap();
}

#[derive(Clone)]
struct CdcKeyspaceCase {
    keyspace_id: u32,
    keyspace_name: String,
    upstream_conn: ConnParams,
    tpcc_database: String,
    changefeed_id: String,
    filter_rules: Vec<String>,
    sync_diff_dir: PathBuf,
    sync_point_interval: Duration,
}

impl CdcKeyspaceCase {
    fn syncpoint_changefeed(&self) -> String {
        format!("{}/{}", self.keyspace_name, self.changefeed_id)
    }
}

fn build_keyspace_cases(
    tc: &TidbCluster,
    keyspace_manager: &KeyspaceManager,
    cdc_dir: &Path,
    sync_point_interval: Duration,
) -> Vec<CdcKeyspaceCase> {
    let mut keyspace_ids = keyspace_manager.get_all_keyspaces();
    keyspace_ids.sort_unstable();
    keyspace_ids
        .into_iter()
        .map(|keyspace_id| {
            let keyspace_name = keyspace_manager
                .get_keyspace_meta(keyspace_id)
                .unwrap()
                .name();
            let tpcc_database = format!("tpcc_ks{keyspace_id}");
            let changefeed_id = changefeed_id_for_keyspace(keyspace_id);
            let filter_rules = vec![format!("`{tpcc_database}`.*")];
            CdcKeyspaceCase {
                keyspace_id,
                keyspace_name,
                upstream_conn: get_tidb_conn_params(tc, keyspace_manager, keyspace_id),
                tpcc_database,
                changefeed_id,
                filter_rules,
                sync_diff_dir: cdc_dir.join(format!("sync_diff_ks{keyspace_id}")),
                sync_point_interval,
            }
        })
        .collect()
}

fn normalized_cdc_switches() -> Switches {
    let mut switches = normalized_switches();
    switches.enable_tikv_cdc = true;
    assert!(
        switches.rep_cdc_newarch,
        "CDC random test requires TiCDC newarch"
    );
    assert!(
        switches.cdc_keyspace_count > 0,
        "CDC_KEYSPACE_COUNT must be greater than 0"
    );
    switches.tiflash_switch_on = false;
    switches.columnar_switch_on = false;
    switches.jepsen_switch_on = false;
    switches.unique_workload_switch_on = false;
    switches.ia_table_ratio = 0.0;
    switches.tidb_gc_lifetime = "600s".to_string();
    switches
}

fn changefeed_id_for_keyspace(keyspace_id: u32) -> String {
    let id = test_id();
    if id.is_empty() {
        format!("random-cdc-ks{keyspace_id}")
    } else {
        format!("random-cdc-{id}-ks{keyspace_id}")
    }
}

async fn prepare_tpcc_for_keyspace(
    keyspace_manager: &KeyspaceManager,
    tpc_bin: &str,
    case: &CdcKeyspaceCase,
) {
    let lock = keyspace_manager.get_keyspace_lock(case.keyspace_id);
    let _guard = lock.shared_lock().await;
    let pool = connect_upstream_pool(case).await;
    let sql = format!("CREATE DATABASE IF NOT EXISTS `{}`", case.tpcc_database);
    crate::sql_util::query(&sql).execute(&pool).await.unwrap();

    let tpc = new_tpcc(case, tpc_bin);
    tpc.prepare(TPCC_THREADS).await.unwrap();
    tpc.check().await.unwrap();
}

async fn check_tpcc_for_keyspace(
    keyspace_manager: &KeyspaceManager,
    tpc_bin: &str,
    case: &CdcKeyspaceCase,
) {
    let lock = keyspace_manager.get_keyspace_lock(case.keyspace_id);
    let _guard = lock.shared_lock().await;
    new_tpcc(case, tpc_bin).check().await.unwrap();
}

fn spawn_tpcc_for_keyspace(
    keyspace_manager: KeyspaceManager,
    tpc_bin: String,
    case: CdcKeyspaceCase,
    running: Running,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let start_time = Instant::now_coarse();
        while running.get() {
            let lock = keyspace_manager.get_keyspace_lock(case.keyspace_id);
            let Some(_guard) = lock.try_shared_lock() else {
                tokio::task::yield_now().await;
                continue;
            };
            let tpc = new_tpcc(&case, &tpc_bin);
            let txns = tpc
                .run(TPCC_THREADS, false, true, TPCC_RUN_DURATION)
                .await
                .unwrap();
            TPCC_COUNTER.fetch_add(txns, Ordering::Relaxed);
            tpc.check().await.unwrap();
        }
        info!("cdc tpcc workload exit"; "keyspace_id" => case.keyspace_id, "dur" => ?start_time.saturating_elapsed());
    })
}

async fn connect_upstream_pool(case: &CdcKeyspaceCase) -> sqlx::MySqlPool {
    let mut opts = sqlx::mysql::MySqlConnectOptions::new()
        .host(&case.upstream_conn.host)
        .port(case.upstream_conn.port)
        .username(&case.upstream_conn.user)
        .database("test");
    if !case.upstream_conn.password.is_empty() {
        opts = opts.password(&case.upstream_conn.password);
    }
    let opts = opts.log_statements(log::LevelFilter::Info);
    sqlx::mysql::MySqlPoolOptions::new()
        .max_connections(8)
        .connect_with(opts)
        .await
        .unwrap()
}

fn new_tpcc(case: &CdcKeyspaceCase, tpc_bin: &str) -> Tpc {
    let mut tpc = Tpc::new(
        format!("cdc-tpcc-{}[{}]", case.keyspace_id, case.keyspace_name),
        PathBuf::from(tpc_bin),
    );
    tpc.tpcc()
        .host(&case.upstream_conn.host)
        .port(case.upstream_conn.port)
        .user(&case.upstream_conn.user)
        .password(&case.upstream_conn.password)
        .db(&case.tpcc_database)
        .warehouses(TPCC_WAREHOUSES)
        .max_procs(TPCC_MAX_PROCS);
    tpc
}