// Copyright 2026 PingCAP, Inc.

use std::{collections::HashSet, ops::Range, sync::Arc, thread, time::Duration};

use futures::{SinkExt, TryStreamExt, executor::block_on};
use grpcio::{ChannelBuilder, ClientDuplexReceiver, EnvBuilder, WriteFlags};
use http::{Method, Request, StatusCode};
use hyper::Body;
use kvengine::{Engine, dfs::new_dfs_from_config};
use kvproto::cdcpb::{ChangeDataClient, ChangeDataEvent, ChangeDataRequest};
use pd_client::PdClient;
use replication_worker::{
    CdcMsg, ReplicationScheduler, ReplicationWorker, ReplicationWorkerConfig,
};
use rfengine::RfEngine;
use test_cloud_server::{
    ServerCluster, ServerClusterBuilder, alloc_node_id_vec, move_region_to_voter_stores,
    must_wait_result,
    oss::{ObjectStorageService, prepare_dfs},
    ticdc::KeyspaceTestService,
    util::RawRegion,
};
use test_util::{alloc_port, init_log_for_test};
use tikv::config::TikvConfig;
use tikv_util::{
    AsyncSafeRuntime, config::ReadableDuration, future::block_on_timeout,
    sys::thread::StdThreadBuildWrapper,
};

use super::util::{
    REGION_REPLICA_COUNT, WAIT_SECS, add_keyspace_via_scheduler, assert_regions_on_store_group,
    assert_shard_stats_eventually_differ, assert_shard_stats_eventually_match,
    create_keyspace_with_one_table, move_region_ids_to_group, move_regions_to_group, region_ids,
    region_versions, scan_keyspace_regions_upstream, split_table_rows, take_lightweight_backup,
    wait_keyspace_shard_metas_ready, wait_worker_store_registered,
};
use crate::new_security_config;

struct ManagedKeyspace {
    keyspace_id: u32,
    table_id: i64,
    regions: Vec<RawRegion>,
}

impl ManagedKeyspace {
    fn region_ids(&self) -> Vec<u64> {
        region_ids(&self.regions)
    }
}

struct UnsafeRecoverFixture {
    _tmp_dir: tempfile::TempDir,
    _oss: ObjectStorageService,
    _worker_dir: tempfile::TempDir,
    _runtime: Arc<AsyncSafeRuntime>,
    backup_rt: tokio::runtime::Runtime,
    cluster: ServerCluster,
    scheduler: ReplicationScheduler,
    rep_kv: Engine,
    rep_raft: RfEngine,
    worker_handle: Option<thread::JoinHandle<()>>,
    _keyspace_svcs: Vec<KeyspaceTestService>,
    worker_grpc_addr: String,
    merged_store_id: u64,
    primary: ManagedKeyspace,
    group0_store_ids: Vec<u64>,
    group0_leader_store_id: u64,
    group1_store_ids: Vec<u64>,
    group1_leader_store_id: u64,
}

impl UnsafeRecoverFixture {
    fn new(test_name: &str) -> Self {
        // Set up the shared storage backend and runtimes for the fixture.
        let (_tmp_dir, _oss, dfs_config) = prepare_dfs(test_name);
        let runtime = Arc::new(AsyncSafeRuntime::new(
            tokio::runtime::Builder::new_multi_thread()
                .worker_threads(4)
                .thread_name("unsafe-recover-clear")
                .enable_all()
                .build()
                .unwrap(),
        ));
        let backup_rt = tokio::runtime::Runtime::new().unwrap();

        // Start a six-node cluster with lightweight backup enabled.
        let nodes = alloc_node_id_vec(6);
        let cluster = ServerClusterBuilder::new(nodes, |_, conf: &mut TikvConfig| {
            conf.dfs = dfs_config.clone();
            conf.rfengine.lightweight_backup = true;
        })
        .pd_server_cnt(1)
        .build();
        cluster.wait_region_replicated(&[], REGION_REPLICA_COUNT);

        let pd_client = cluster.get_pd_client();
        pd_client.disable_default_operator();

        // Split the cluster into non-included and include-only store groups.
        let mut nodes = cluster.get_nodes();
        nodes.sort_unstable();
        let group0_nodes = nodes[..3].to_vec();
        let group1_nodes = nodes[3..].to_vec();
        let group0_store_ids = group0_nodes
            .iter()
            .map(|node_id| cluster.get_store_id(*node_id))
            .collect::<Vec<_>>();
        let group1_store_ids = group1_nodes
            .iter()
            .map(|node_id| cluster.get_store_id(*node_id))
            .collect::<Vec<_>>();
        let group1_store_addrs = group1_nodes
            .iter()
            .map(|node_id| cluster.get_node_config(*node_id).server.addr.clone())
            .collect::<Vec<_>>();
        let group0_leader_store_id = group0_store_ids[0];
        let group1_leader_store_id = group1_store_ids[0];

        // Move every existing region onto the non-included store group.
        let all_regions = block_on(pd_client.scan_regions(vec![], vec![], 1024)).unwrap();
        for region in all_regions {
            move_region_to_voter_stores(
                &cluster,
                pd_client.as_ref(),
                region.get_region().get_id(),
                &group0_store_ids,
                group0_leader_store_id,
            );
        }

        // Create the primary keyspace, split it, and write its initial rows.
        let mut client = cluster.new_client();
        let (keyspace_id, table_id) =
            create_keyspace_with_one_table(&backup_rt, &cluster, &mut client, test_name);
        split_table_rows(&cluster, &mut client, keyspace_id, table_id, &[10, 20]);
        Self::write_keyspace_rows_with_runtime(
            &backup_rt,
            &cluster,
            keyspace_id,
            table_id,
            0..30,
            "value",
        );

        // Move the primary keyspace to the include-only store group to make it
        // replicable.
        let mut regions = scan_keyspace_regions_upstream(&mut client, keyspace_id);
        assert_regions_on_store_group(&regions, &group0_store_ids);
        move_regions_to_group(
            &cluster,
            pd_client.as_ref(),
            &regions,
            &group1_store_ids,
            group1_leader_store_id,
        );
        regions = scan_keyspace_regions_upstream(&mut client, keyspace_id);
        assert_regions_on_store_group(&regions, &group1_store_ids);
        let expected_versions = region_versions(&regions);

        let primary = ManagedKeyspace {
            keyspace_id,
            table_id,
            regions,
        };

        // Configure the replication worker to include only the target store group.
        let worker_dir = tempfile::Builder::new()
            .prefix("unsafe-recover-clear-worker-")
            .tempdir()
            .unwrap();
        let mut worker_cfg = ReplicationWorkerConfig {
            enabled: true,
            grpc_addr: format!("127.0.0.1:{}", alloc_port()),
            advertise_addr: String::new(),
            sync_interval: ReadableDuration::millis(100),
            report_region_interval: ReadableDuration::millis(100),
            include_only_store_addr_keywords: group1_store_addrs,
            ..Default::default()
        };
        worker_cfg.advertise_addr = worker_cfg.grpc_addr.clone();
        worker_cfg.merged_engine.mem_table_size = cluster.get_mem_table_size();

        let worker_grpc_addr = worker_cfg.advertise_addr.clone();
        let merged_store_id = worker_cfg.merged_engine.merged_store_id;

        // Take a backup snapshot that the worker can bootstrap from.
        take_lightweight_backup(&backup_rt, &dfs_config, cluster.get_pure_pd_client());

        // Start the replication worker and capture its testing handles.
        let dfs = new_dfs_from_config(dfs_config);
        let encryption_key_manager = Arc::new(cloud_encryption::EncryptionKeyManager::new());
        let mut worker = ReplicationWorker::new(
            cluster.get_pure_pd_client(),
            cluster.pd_endpoints().to_vec(),
            dfs,
            worker_dir.path().display().to_string(),
            new_security_config(),
            worker_cfg,
            runtime.clone(),
            encryption_key_manager,
        )
        .unwrap();
        let scheduler = worker.scheduler();
        let rep_kv = worker.test_kv_engine();
        let rep_raft = worker.test_raft_engine();
        let worker_handle = thread::Builder::new()
            .name("unsafe-recover-clear-worker".to_owned())
            .spawn_wrapper(move || tikv_util::init_task_local_sync(|| worker.run()))
            .unwrap();

        // Register the primary keyspace and wait until downstream shards are ready.
        wait_keyspace_shard_metas_ready(&scheduler, primary.keyspace_id, &expected_versions);
        let keyspace_svc = KeyspaceTestService::new(
            primary.keyspace_id,
            merged_store_id,
            worker_grpc_addr.clone(),
        );
        add_keyspace_via_scheduler(&scheduler, primary.keyspace_id, &keyspace_svc);
        wait_worker_store_registered(&keyspace_svc, merged_store_id);
        assert_shard_stats_eventually_match(&cluster, &rep_kv, &primary.regions);

        Self {
            _tmp_dir,
            _oss,
            _worker_dir: worker_dir,
            _runtime: runtime,
            backup_rt,
            cluster,
            scheduler,
            rep_kv,
            rep_raft,
            worker_handle: Some(worker_handle),
            _keyspace_svcs: vec![keyspace_svc],
            worker_grpc_addr,
            merged_store_id,
            primary,
            group0_store_ids,
            group0_leader_store_id,
            group1_store_ids,
            group1_leader_store_id,
        }
    }

    fn scan_keyspace_regions(&self, keyspace_id: u32) -> Vec<RawRegion> {
        let mut client = self.cluster.new_client();
        scan_keyspace_regions_upstream(&mut client, keyspace_id)
    }

    fn scan_regions_by_ids(&self, region_ids: &[u64]) -> Vec<RawRegion> {
        let region_id_set = region_ids.iter().copied().collect::<HashSet<_>>();
        let all_regions = block_on(
            self.cluster
                .get_pd_client()
                .scan_regions(vec![], vec![], 2048),
        )
        .unwrap();
        all_regions
            .into_iter()
            .map(|region| region.get_region().clone().into())
            .filter(|region: &RawRegion| region_id_set.contains(&region.id))
            .collect()
    }

    fn create_replicated_keyspace(
        &mut self,
        keyspace_name: &str,
        split_points: &[u32],
        value_prefix: &str,
    ) -> ManagedKeyspace {
        let mut client = self.cluster.new_client();
        let (keyspace_id, table_id) = create_keyspace_with_one_table(
            &self.backup_rt,
            &self.cluster,
            &mut client,
            keyspace_name,
        );
        split_table_rows(
            &self.cluster,
            &mut client,
            keyspace_id,
            table_id,
            split_points,
        );
        self.write_keyspace_rows(keyspace_id, table_id, 0..30, value_prefix);

        let mut regions = scan_keyspace_regions_upstream(&mut client, keyspace_id);
        assert_regions_on_store_group(&regions, &self.group0_store_ids);
        move_regions_to_group(
            &self.cluster,
            self.cluster.get_pd_client().as_ref(),
            &regions,
            &self.group1_store_ids,
            self.group1_leader_store_id,
        );
        regions = scan_keyspace_regions_upstream(&mut client, keyspace_id);
        assert_regions_on_store_group(&regions, &self.group1_store_ids);
        let expected_versions = region_versions(&regions);

        wait_keyspace_shard_metas_ready(&self.scheduler, keyspace_id, &expected_versions);
        let keyspace_svc = KeyspaceTestService::new(
            keyspace_id,
            self.merged_store_id,
            self.worker_grpc_addr.clone(),
        );
        add_keyspace_via_scheduler(&self.scheduler, keyspace_id, &keyspace_svc);
        wait_worker_store_registered(&keyspace_svc, self.merged_store_id);
        assert_shard_stats_eventually_match(&self.cluster, &self.rep_kv, &regions);
        self._keyspace_svcs.push(keyspace_svc);

        ManagedKeyspace {
            keyspace_id,
            table_id,
            regions,
        }
    }

    fn write_keyspace_rows(
        &self,
        keyspace_id: u32,
        table_id: i64,
        rows: Range<usize>,
        value_prefix: &str,
    ) {
        Self::write_keyspace_rows_with_runtime(
            &self.backup_rt,
            &self.cluster,
            keyspace_id,
            table_id,
            rows,
            value_prefix,
        );
    }

    fn write_keyspace_rows_with_runtime(
        runtime: &tokio::runtime::Runtime,
        cluster: &ServerCluster,
        keyspace_id: u32,
        table_id: i64,
        rows: Range<usize>,
        value_prefix: &str,
    ) {
        let keyspace_client = runtime.block_on(cluster.new_keyspace_client());
        runtime
            .block_on(keyspace_client.keyspace_put_kv(
                keyspace_id,
                table_id,
                rows,
                |i| format!("k{i:08}").into_bytes(),
                |i| format!("{value_prefix}-{i:08}").into_bytes(),
            ))
            .unwrap();
    }

    fn move_region_ids_to_group0(&self, region_ids: &[u64]) {
        move_region_ids_to_group(
            &self.cluster,
            self.cluster.get_pd_client().as_ref(),
            region_ids,
            &self.group0_store_ids,
            self.group0_leader_store_id,
        );
        self.assert_region_ids_on_group(region_ids, &self.group0_store_ids);
    }

    fn move_region_ids_to_group1(&self, region_ids: &[u64]) {
        move_region_ids_to_group(
            &self.cluster,
            self.cluster.get_pd_client().as_ref(),
            region_ids,
            &self.group1_store_ids,
            self.group1_leader_store_id,
        );
        self.assert_region_ids_on_group(region_ids, &self.group1_store_ids);
    }

    fn inject_dependency(&self, parent_region_id: u64, child_region_id: u64) {
        self.rep_raft
            .add_dependent(parent_region_id, child_region_id);
    }

    fn assert_region_ids_on_group(&self, region_ids: &[u64], expected_store_ids: &[u64]) {
        let selected = self.scan_regions_by_ids(region_ids);
        assert_eq!(
            selected.len(),
            region_ids.len(),
            "missing regions after move: requested {:?}, actual {:?}",
            region_ids,
            selected.iter().map(|region| region.id).collect::<Vec<_>>(),
        );
        assert_regions_on_store_group(&selected, expected_store_ids);
    }

    fn post_clear_via_scheduler(
        &self,
        selector: &str,
        dry_run: bool,
    ) -> (StatusCode, serde_json::Value) {
        let uri = if dry_run {
            format!("/cdc/unsafe_recover/clear?{selector}&dry_run=true")
        } else {
            format!("/cdc/unsafe_recover/clear?{selector}")
        };
        let req = Request::builder()
            .method(Method::POST)
            .uri(uri)
            .body(Body::empty())
            .unwrap();
        let resp = block_on_timeout(
            self.scheduler.handle_http_request(req),
            Duration::from_secs(5),
        )
        .unwrap_or_else(|_| {
            panic!(
                "unsafe recover clear request timed out: selector={}, dry_run={}",
                selector, dry_run
            )
        })
        .unwrap_or_else(|err| {
            panic!(
                "unsafe recover clear request failed: selector={}, dry_run={}, err={:?}",
                selector, dry_run, err
            )
        });
        let status = resp.status();
        let body = self.backup_rt.block_on(read_json(resp));
        (status, body)
    }

    fn wait_local_shards_present(&self, region_ids: &[u64]) {
        must_wait_result(
            || {
                let missing = region_ids
                    .iter()
                    .copied()
                    .filter(|region_id| self.rep_kv.get_shard(*region_id).is_none())
                    .collect::<Vec<_>>();
                if missing.is_empty() {
                    Ok(())
                } else {
                    Err(format!("missing downstream shards {:?}", missing))
                }
            },
            WAIT_SECS,
            || format!("wait local shards present {:?}", region_ids),
        );
    }

    fn wait_local_shards_absent(&self, region_ids: &[u64]) {
        must_wait_result(
            || {
                let present = region_ids
                    .iter()
                    .copied()
                    .filter(|region_id| self.rep_kv.get_shard(*region_id).is_some())
                    .collect::<Vec<_>>();
                if present.is_empty() {
                    Ok(())
                } else {
                    Err(format!("unexpected downstream shards {:?}", present))
                }
            },
            WAIT_SECS,
            || format!("wait local shards absent {:?}", region_ids),
        );
    }
}

impl Drop for UnsafeRecoverFixture {
    fn drop(&mut self) {
        self.scheduler.schedule(CdcMsg::Stop);
        if let Some(handle) = self.worker_handle.take() {
            let _ = handle.join();
        }
        self.cluster.stop();
    }
}

#[test]
fn test_replication_worker_unsafe_recover_clear() {
    init_log_for_test();

    // All scenarios need the same six-node topology, backup, and replication
    // worker. Keep destructive scenarios isolated with dedicated keyspaces
    // instead of rebuilding that expensive fixture for every assertion group.
    let mut fixture = UnsafeRecoverFixture::new("test_unsafe_recover_clear_");
    check_region_and_dry_run(&fixture, &fixture.primary);

    // Active CDC cleanup needs an independently registered keyspace because a
    // preceding unsafe clear invalidates the original keyspace registration.
    // One region is sufficient to cover the stream-cleanup invariant.
    let active_cdc = fixture.create_replicated_keyspace(
        "test-unsafe-recover-clear-active-cdc",
        &[],
        "active-cdc",
    );
    check_active_cdc_request_cleanup(&fixture, &active_cdc);

    let atomic_target = fixture.create_replicated_keyspace(
        "test-unsafe-recover-clear-atomic-target",
        &[10],
        "atomic-target",
    );
    let reload =
        fixture.create_replicated_keyspace("test-unsafe-recover-clear-reload", &[], "reload");
    check_keyspace_atomicity_and_dependency_closure(&fixture, &atomic_target, &reload);
    check_keyspace_reload_after_moving_back(&fixture, &reload);
}

fn check_region_and_dry_run(fixture: &UnsafeRecoverFixture, target: &ManagedKeyspace) {
    let region_ids = target.region_ids();
    let target_region_id = region_ids[1];
    let other_region_ids = region_ids
        .iter()
        .copied()
        .filter(|region_id| *region_id != target_region_id)
        .collect::<Vec<_>>();

    // Verify clearing is rejected while the region is still on included stores.
    let (reject_status, reject_body) =
        fixture.post_clear_via_scheduler(&format!("region_id={target_region_id}"), false);
    assert_eq!(reject_status, StatusCode::BAD_REQUEST);
    assert!(!reject_body["success"].as_bool().unwrap());
    assert!(
        reject_body["reason"]
            .as_str()
            .is_some_and(|reason| reason.contains("non-skipped stores")),
        "unexpected body: {:?}",
        reject_body,
    );
    fixture.wait_local_shards_present(&region_ids);

    // Move the target region out of the included store group.
    fixture.move_region_ids_to_group0(&[target_region_id]);
    fixture.wait_local_shards_present(&region_ids);

    // Check that dry run reports the region without removing local shards.
    let (dry_run_status, dry_run_body) =
        fixture.post_clear_via_scheduler(&format!("region_id={target_region_id}"), true);
    assert_eq!(dry_run_status, StatusCode::OK);
    assert!(dry_run_body["success"].as_bool().unwrap());
    assert!(dry_run_body["dry_run"].as_bool().unwrap());
    assert_eq!(
        dry_run_body["region_ids"],
        serde_json::json!([target_region_id])
    );
    fixture.wait_local_shards_present(&region_ids);

    // Clear the target region and keep the remaining replicated shards intact.
    let (status, body) =
        fixture.post_clear_via_scheduler(&format!("region_id={target_region_id}"), false);
    assert_eq!(status, StatusCode::OK);
    assert!(body["success"].as_bool().unwrap());
    assert!(!body["dry_run"].as_bool().unwrap());
    assert_eq!(body["region_ids"], serde_json::json!([target_region_id]));

    fixture.wait_local_shards_absent(&[target_region_id]);
    fixture.wait_local_shards_present(&other_region_ids);
}

fn check_active_cdc_request_cleanup(fixture: &UnsafeRecoverFixture, target: &ManagedKeyspace) {
    let target_region = target.regions[0].clone();
    let request_id = 1;

    // Open a CDC stream and register an active feed for the target region.
    let client = new_cdc_client(&fixture.worker_grpc_addr);
    let (mut tx, mut rx) = client.event_feed_v2().unwrap();
    let req = new_cdc_register_request(&target_region, request_id);
    block_on(tx.send((req, WriteFlags::default()))).unwrap();
    wait_for_cdc_stream_activity(&mut rx);

    // Move the region out and clear it through the unsafe recover endpoint.
    fixture.move_region_ids_to_group0(&[target_region.id]);
    let (status, body) =
        fixture.post_clear_via_scheduler(&format!("region_id={}", target_region.id), false);
    assert_eq!(status, StatusCode::OK);
    assert!(body["success"].as_bool().unwrap());

    // Verify the downstream shard is removed and the CDC stream observes the
    // cleanup.
    fixture.wait_local_shards_absent(&[target_region.id]);
    wait_for_cdc_region_not_found(&mut rx, target_region.id, request_id);
}

fn check_keyspace_atomicity_and_dependency_closure(
    fixture: &UnsafeRecoverFixture,
    target: &ManagedKeyspace,
    untouched: &ManagedKeyspace,
) {
    let region_ids = target.region_ids();
    assert_eq!(region_ids.len(), 2, "atomic target should have two regions");
    let initially_blocked_region_id = *region_ids.last().unwrap();
    let initially_clearable_region_ids = region_ids[..region_ids.len() - 1].to_vec();

    // Move only part of the keyspace out of included stores.
    fixture.move_region_ids_to_group0(&initially_clearable_region_ids);
    fixture.wait_local_shards_present(&region_ids);
    fixture.wait_local_shards_present(&untouched.region_ids());

    // Verify keyspace clear is rejected until every region becomes clearable.
    let (reject_status, reject_body) =
        fixture.post_clear_via_scheduler(&format!("keyspace_id={}", target.keyspace_id), false);
    assert_eq!(reject_status, StatusCode::BAD_REQUEST);
    assert!(!reject_body["success"].as_bool().unwrap());
    assert!(
        reject_body["reason"]
            .as_str()
            .is_some_and(|reason| reason.contains("non-skipped stores")),
        "unexpected body: {:?}",
        reject_body,
    );
    fixture.wait_local_shards_present(&region_ids);
    fixture.wait_local_shards_present(&untouched.region_ids());

    // Move the remaining blocked region out. Before clearing the keyspace,
    // verify that a single-region request cannot omit its dependent shard.
    fixture.move_region_ids_to_group0(&[initially_blocked_region_id]);
    let parent_region_id = region_ids[0];
    let child_region_id = region_ids[1];
    fixture.inject_dependency(parent_region_id, child_region_id);
    let (dependency_status, dependency_body) =
        fixture.post_clear_via_scheduler(&format!("region_id={parent_region_id}"), false);
    assert_eq!(dependency_status, StatusCode::BAD_REQUEST);
    assert!(!dependency_body["success"].as_bool().unwrap());
    assert!(
        dependency_body["reason"]
            .as_str()
            .is_some_and(|reason| reason.contains("dependent regions outside request")),
        "unexpected body: {:?}",
        dependency_body,
    );
    fixture.wait_local_shards_present(&region_ids);

    // The full-keyspace request includes the dependency closure and clears
    // both regions atomically.
    let (status, body) =
        fixture.post_clear_via_scheduler(&format!("keyspace_id={}", target.keyspace_id), false);
    assert_eq!(status, StatusCode::OK);
    assert!(body["success"].as_bool().unwrap());
    assert_eq!(body["region_ids"], serde_json::json!(region_ids));

    // Confirm only the target keyspace is removed from downstream state.
    fixture.wait_local_shards_absent(&region_ids);
    fixture.wait_local_shards_present(&untouched.region_ids());
    assert_shard_stats_eventually_match(&fixture.cluster, &fixture.rep_kv, &untouched.regions);
}

fn check_keyspace_reload_after_moving_back(
    fixture: &UnsafeRecoverFixture,
    target: &ManagedKeyspace,
) {
    let region_ids = target.region_ids();

    fixture.move_region_ids_to_group0(&region_ids);

    // Write new upstream data so downstream shard stats become stale.
    fixture.write_keyspace_rows(
        target.keyspace_id,
        target.table_id,
        30..45,
        "value-after-move-out",
    );
    assert_shard_stats_eventually_differ(&fixture.cluster, &fixture.rep_kv, &target.regions);

    // Clear the moved-out keyspace from downstream state.
    let (status, body) =
        fixture.post_clear_via_scheduler(&format!("keyspace_id={}", target.keyspace_id), false);
    assert_eq!(status, StatusCode::OK);
    assert!(body["success"].as_bool().unwrap());

    fixture.wait_local_shards_absent(&region_ids);

    // Move the keyspace back and verify the worker reloads fresh shard state.
    fixture.move_region_ids_to_group1(&region_ids);
    let refreshed_regions = fixture.scan_keyspace_regions(target.keyspace_id);
    assert_shard_stats_eventually_match(&fixture.cluster, &fixture.rep_kv, &refreshed_regions);
}

fn new_cdc_client(addr: &str) -> ChangeDataClient {
    let env = Arc::new(EnvBuilder::new().build());
    let channel = ChannelBuilder::new(env).connect(addr);
    ChangeDataClient::new(channel)
}

fn new_cdc_register_request(region: &RawRegion, request_id: u64) -> ChangeDataRequest {
    let mut req = ChangeDataRequest {
        region_id: region.id,
        request_id,
        checkpoint_ts: 0,
        ..Default::default()
    };
    req.set_region_epoch(region.epoch.clone());
    req.mut_header().set_ticdc_version("8.5.0".into());
    req
}

fn wait_for_cdc_stream_activity(rx: &mut ClientDuplexReceiver<ChangeDataEvent>) {
    must_wait_result(
        || match block_on_timeout(rx.try_next(), Duration::from_secs(1)) {
            Ok(Ok(Some(_))) => Ok(()),
            Ok(Ok(None)) => Err("cdc stream closed unexpectedly".to_owned()),
            Ok(Err(err)) => Err(format!("receive cdc event failed: {:?}", err)),
            Err(_) => Err("cdc stream still idle".to_owned()),
        },
        WAIT_SECS,
        || "wait cdc stream activity".to_owned(),
    );
}

fn wait_for_cdc_region_not_found(
    rx: &mut ClientDuplexReceiver<ChangeDataEvent>,
    region_id: u64,
    request_id: u64,
) {
    must_wait_result(
        || match block_on_timeout(rx.try_next(), Duration::from_secs(1)) {
            Ok(Ok(Some(event))) => {
                if change_data_event_has_region_not_found(&event, region_id, request_id) {
                    Ok(())
                } else {
                    Err(format!(
                        "unexpected cdc event while waiting for region_not_found: {:?}",
                        event
                    ))
                }
            }
            Ok(Ok(None)) => Err("cdc stream closed before region_not_found".to_owned()),
            Ok(Err(err)) => Err(format!("receive cdc event failed: {:?}", err)),
            Err(_) => Err("cdc stream has no region_not_found yet".to_owned()),
        },
        WAIT_SECS,
        || {
            format!(
                "wait cdc region_not_found for region {} request {}",
                region_id, request_id
            )
        },
    );
}

fn change_data_event_has_region_not_found(
    event: &ChangeDataEvent,
    region_id: u64,
    request_id: u64,
) -> bool {
    event.get_events().iter().any(|inner| {
        inner.get_region_id() == region_id
            && inner.get_request_id() == request_id
            && inner.get_error().has_region_not_found()
            && inner.get_error().get_region_not_found().get_region_id() == region_id
    })
}

async fn read_json(resp: http::Response<Body>) -> serde_json::Value {
    let body = hyper::body::to_bytes(resp.into_body()).await.unwrap();
    serde_json::from_slice(&body).unwrap()
}