// Copyright 2025 PingCAP, Inc.

use std::{
    collections::HashMap,
    fs,
    net::TcpListener,
    path::PathBuf,
    process::Command,
    sync::{Arc, Mutex},
    thread,
    time::Duration,
};

use async_trait::async_trait;
use grpcio::{ChannelBuilder, EnvBuilder, Environment};
use hyper::{Body, Method, Request, Response, http::StatusCode, service::service_fn};
use pd_client::PdClient;
use replication_worker::{KeyspaceService, KeyspaceStates};
use security::{HttpClient, SecurityConfig, SecurityManager};
use serde_derive::{Deserialize, Serialize};
use test_pd_client::{PdClientExt, PdWrapper, TestPdClient};
use tikv_util::sys::process::{
    ChildSupervisorHandle, ProcessSpawnSpec, StopReason, spawn_supervisor,
};
use tokio::sync::oneshot;
use url::Url;

use crate::{
    client::{ClusterClient, ClusterClientOptions, RefStore},
    try_wait_result_async,
    util::unused_local_port,
};

const ERR_API_INVALID_PARAM: &str = "CDC:ErrAPIInvalidParam";
const ERR_CHANGEFEED_ALREADY_EXISTS: &str = "CDC:ErrChangeFeedAlreadyExists";
const ERR_CHANGEFEED_NOT_EXISTS: &str = "CDC:ErrChangeFeedNotExists";
const ERR_OWNER_NOT_FOUND: &str = "CDC:ErrOwnerNotFound";
const MSG_COORDINATOR_NOT_INITIALIZED: &str = "coordinator is not fully initialized, wait a moment";

/// Environment variable that points to the TiCDC binary used by integration
/// tests.
pub const CDC_BIN_ENV_KEY: &str = "CDC_BIN";

const CDC_HEALTHY_TIMEOUT: Duration = Duration::from_secs(60);
const CHANGEFEED_WAIT_TIMEOUT: Duration = Duration::from_secs(60);

// Ref: https://docs.pingcap.com/tidb/stable/ticdc-open-api-v2/#parameter-descriptions.
#[rustfmt::skip]
/*
{
  "changefeed_id":"rep-task",
  "sink_uri":"mysql://root@127.0.0.1:4455",
  "start_ts": 460426866192809988,
  "replica_config":{
    "enable_sync_point":true,
    "sync_point_interval":"30s",
    "filter":{
      "rules":["sbtest.*"]
    }
  }
}
 */
#[derive(Serialize, Default)]
pub struct ChangefeedParams {
    pub changefeed_id: String,
    pub sink_uri: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub start_ts: Option<u64>,
    pub replica_config: ChangefeedReplicaConfig,
}

#[derive(Serialize, Default)]
pub struct ChangefeedReplicaConfig {
    pub enable_sync_point: bool,
    pub sync_point_interval: String, // Minimum is 30s.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub ignore_ineligible_table: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub filter: Option<ChangefeedFilter>,
}

#[derive(Serialize, Default)]
pub struct ChangefeedFilter {
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub rules: Vec<String>,
}

/// Validates that a TiCDC binary exists and supports the required server flags.
pub fn validate_cdc_binary(cdc_bin: &str) {
    let meta =
        fs::metadata(cdc_bin).unwrap_or_else(|e| panic!("CDC_BIN {} is invalid: {}", cdc_bin, e));
    assert!(meta.is_file(), "CDC_BIN {} is not a file", cdc_bin);

    let output = Command::new(cdc_bin)
        .arg("server")
        .arg("--help")
        .output()
        .unwrap_or_else(|e| panic!("failed to execute {} server --help: {}", cdc_bin, e));
    let stdout = String::from_utf8_lossy(&output.stdout);
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        output.status.success(),
        "cdc binary help failed: status {:?}, stdout {}, stderr {}",
        output.status,
        stdout,
        stderr
    );
    let help = format!("{stdout}\n{stderr}");
    for flag in ["--pd", "--addr", "--data-dir", "--newarch"] {
        assert!(
            help.contains(flag),
            "cdc binary does not advertise required flag {}: {}",
            flag,
            help
        );
    }
}

/// Supervised TiCDC server process used by integration tests.
pub struct CdcServer {
    addr: String,
    http_client: HttpClient,
    handle: ChildSupervisorHandle,
}

/// Configuration for the TiCDC server process used by integration tests.
#[derive(Default, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct CdcServerConfig {
    pub enable_legacy_safepoint: bool,
}

impl CdcServer {
    /// Starts a TiCDC server process bound to a random local port.
    pub fn start(
        data_dir: PathBuf,
        log_file: PathBuf,
        cdc_bin: String,
        pd_endpoints: Vec<String>,
        log_level: String,
        config: CdcServerConfig,
    ) -> Self {
        fs::create_dir_all(&data_dir).unwrap();
        let port = unused_local_port();
        let addr = format!("127.0.0.1:{port}");
        let pd_endpoints = pd_endpoints
            .into_iter()
            .map(|endpoint| {
                if endpoint.contains("://") {
                    endpoint
                } else {
                    format!("http://{endpoint}")
                }
            })
            .collect::<Vec<_>>()
            .join(",");
        let mut args = vec![
            "server".to_string(),
            format!("--addr={addr}"),
            format!("--log-file={}", log_file.display()),
            format!("--log-level={log_level}"),
            format!("--data-dir={}", data_dir.join("data").display()),
            format!("--pd={pd_endpoints}"),
            "--newarch".to_string(),
        ];
        let config_file = data_dir.join("ticdc.toml");
        fs::write(&config_file, toml::to_string(&config).unwrap())
            .unwrap_or_else(|e| panic!("write CDC config {} failed: {}", config_file.display(), e));
        args.push(format!("--config={}", config_file.display()));
        let spec = ProcessSpawnSpec::new(
            "random cdc",
            "cdc",
            "random-cdc-proc",
            cdc_bin,
            args,
            log_file,
        );
        let handle = spawn_supervisor(spec).unwrap_or_else(|e| panic!("start CDC failed: {}", e));
        let http_client = SecurityManager::new(&SecurityConfig::default())
            .unwrap()
            .http_client(hyper::Client::builder())
            .unwrap();
        Self {
            addr,
            http_client,
            handle,
        }
    }

    /// Returns the TiCDC HTTP API address.
    pub fn addr(&self) -> &str {
        &self.addr
    }

    /// Creates a keyspace-scoped TiCDC OpenAPI helper bound to this server
    /// address.
    pub fn cdc_cli(&self, keyspace_name: String) -> CdcCli {
        CdcCli::new(self.http_client.clone(), self.addr.clone(), keyspace_name)
    }

    /// Waits until the TiCDC status endpoint becomes healthy.
    pub async fn wait_healthy(&self) {
        let client = self.http_client.clone();
        let url = format!("http://{}/api/v2/status", self.addr);
        try_wait_result_async(
            || {
                let client = client.clone();
                let url = url.clone();
                Box::pin(async move {
                    dispatch_http(&client, Method::GET, &url, None)
                        .await
                        .map(|_| ())
                })
            },
            CDC_HEALTHY_TIMEOUT.as_secs() as usize,
        )
        .await
        .unwrap_or_else(|e| panic!("wait TiCDC healthy at {url} failed: {}", e));
    }

    /// Stops the TiCDC process and waits for supervisor shutdown.
    pub fn stop(self) {
        self.handle
            .stop_and_join(StopReason::Destroy)
            .unwrap_or_else(|e| panic!("stop CDC failed: {}", e));
    }
}

/// TiCDC OpenAPI helper for integration tests.
pub struct CdcCli {
    client: HttpClient,
    cdc_addr: String,
    keyspace_name: String,
}

impl CdcCli {
    fn new(client: HttpClient, cdc_addr: String, keyspace_name: String) -> Self {
        Self {
            client,
            cdc_addr,
            keyspace_name,
        }
    }

    /// Validates the keyspace-scoped changefeed list contract.
    pub async fn validate_keyspace_query_contract(&self) {
        let url = self.changefeed_list_url();
        let display_url = url.clone();
        let client = self.client.clone();
        let body = try_wait_result_async(
            move || {
                let client = client.clone();
                let url = url.clone();
                Box::pin(async move {
                    let resp = dispatch_http_raw(&client, Method::GET, &url, None).await?;
                    if resp.status.is_success() {
                        Ok(resp.body)
                    } else {
                        assert!(
                            is_retryable_keyspace_query_error(&resp.body),
                            "standalone TiCDC keyspace query validation failed: status {}, body {}",
                            resp.status,
                            resp.body,
                        );
                        Err(format!(
                            "GET {url}: status {}, body {}",
                            resp.status, resp.body
                        ))
                    }
                })
            },
            CHANGEFEED_WAIT_TIMEOUT.as_secs() as usize,
        )
        .await
        .unwrap_or_else(|e| panic!("validate keyspace query contract failed: {}", e));
        let list = serde_json::from_str::<ChangefeedListResp>(&body).unwrap_or_else(|e| {
            panic!(
                "invalid TiCDC list response from {}: {}, body {}",
                display_url, e, body
            )
        });
        assert_eq!(
            list.total,
            list.items.len(),
            "invalid TiCDC list total from {display_url}: {body}"
        );
    }

    /// Creates a keyspace-scoped changefeed and retries while TiCDC is
    /// initializing.
    pub async fn create_changefeed(
        &self,
        changefeed_id: &str,
        downstream: &crate::tidb::ConnParams,
        start_ts: u64,
        filter_rules: &[String],
        sync_point_interval: Duration,
    ) {
        let url = self.changefeed_list_url();
        let params = ChangefeedParams {
            changefeed_id: changefeed_id.to_string(),
            sink_uri: mysql_sink_uri(downstream),
            start_ts: Some(start_ts),
            replica_config: ChangefeedReplicaConfig {
                enable_sync_point: true,
                sync_point_interval: format!("{}s", sync_point_interval.as_secs()),
                ignore_ineligible_table: Some(true),
                filter: Some(ChangefeedFilter {
                    rules: filter_rules.to_vec(),
                }),
            },
        };
        let body = serde_json::to_string(&params).unwrap();
        let client = self.client.clone();
        let keyspace_name = self.keyspace_name.clone();
        try_wait_result_async(
            move || {
                let client = client.clone();
                let url = url.clone();
                let body = body.clone();
                let keyspace_name = keyspace_name.clone();
                Box::pin(async move {
                    let resp = dispatch_http_raw(&client, Method::POST, &url, Some(body.clone())).await?;
                    if resp.status.is_success() || is_create_changefeed_already_exists_error(&resp.body) {
                        Ok(resp.body)
                    } else {
                        assert!(
                            is_retryable_create_changefeed_error(&resp.body),
                            "create changefeed failed: url {url}, keyspace_name {}, body {}, status {}, response {}",
                            keyspace_name,
                            body,
                            resp.status,
                            resp.body,
                        );
                        Err(format!(
                            "POST {url}: status {}, body {}",
                            resp.status, resp.body
                        ))
                    }
                })
            },
            CHANGEFEED_WAIT_TIMEOUT.as_secs() as usize,
        )
        .await
        .unwrap_or_else(|e| panic!("create changefeed {changefeed_id} failed: {}", e));
    }

    /// Waits until a changefeed reaches the normal state.
    pub async fn wait_changefeed_normal(&self, changefeed_id: &str) {
        let client = self.client.clone();
        let url = self.changefeed_query_url(changefeed_id);
        let changefeed_id = changefeed_id.to_string();
        let display_changefeed_id = changefeed_id.clone();
        try_wait_result_async(
            move || {
                let client = client.clone();
                let url = url.clone();
                Box::pin(async move {
                    let info = query_changefeed(&client, &url).await?;
                    if info.state == "normal" {
                        Ok(())
                    } else {
                        Err(format!(
                            "changefeed {} is not normal: state {}, checkpoint {}, error {:?}",
                            info.id, info.state, info.checkpoint_ts, info.error
                        ))
                    }
                })
            },
            CHANGEFEED_WAIT_TIMEOUT.as_secs() as usize,
        )
        .await
        .unwrap_or_else(|e| {
            panic!(
                "wait changefeed {display_changefeed_id} normal failed: {}",
                e
            )
        });
    }

    /// Waits until a changefeed checkpoint reaches the specified timestamp.
    pub async fn wait_changefeed_checkpoint(
        &self,
        changefeed_id: &str,
        verify_ts: u64,
        timeout: Duration,
    ) {
        let client = self.client.clone();
        let url = self.changefeed_query_url(changefeed_id);
        let changefeed_id = changefeed_id.to_string();
        let display_changefeed_id = changefeed_id.clone();
        try_wait_result_async(
            move || {
                let client = client.clone();
                let url = url.clone();
                Box::pin(async move {
                    let info = query_changefeed(&client, &url).await?;
                    if info.state == "normal" && info.checkpoint_ts >= verify_ts {
                        Ok(())
                    } else {
                        Err(format!(
                            "changefeed {} checkpoint not reached: state {}, checkpoint {}, verify_ts {}, error {:?}",
                            info.id, info.state, info.checkpoint_ts, verify_ts, info.error
                        ))
                    }
                })
            },
            timeout.as_secs() as usize,
        )
        .await
        .unwrap_or_else(|e| {
            panic!(
                "wait changefeed {display_changefeed_id} checkpoint failed: {}",
                e
            )
        });
    }

    /// Deletes a keyspace-scoped changefeed.
    pub async fn delete_changefeed(&self, changefeed_id: &str) {
        let client = self.client.clone();
        let url = self.changefeed_query_url(changefeed_id);
        let changefeed_id = changefeed_id.to_string();
        let display_changefeed_id = changefeed_id.clone();
        try_wait_result_async(
            move || {
                let client = client.clone();
                let url = url.clone();
                let changefeed_id = changefeed_id.clone();
                Box::pin(async move {
                    let resp = dispatch_http_raw(&client, Method::DELETE, &url, None).await?;
                    if resp.status.is_success()
                        || is_delete_changefeed_not_exists_error(&resp.body, &changefeed_id)
                    {
                        Ok(())
                    } else {
                        assert!(
                            is_retryable_delete_changefeed_error(&resp.body),
                            "delete changefeed failed: url {url}, changefeed {}, status {}, response {}",
                            changefeed_id,
                            resp.status,
                            resp.body,
                        );
                        Err(format!(
                            "DELETE {url}: status {}, body {}",
                            resp.status, resp.body
                        ))
                    }
                })
            },
            CHANGEFEED_WAIT_TIMEOUT.as_secs() as usize,
        )
        .await
        .unwrap_or_else(|e| panic!("delete changefeed {display_changefeed_id} failed: {}", e));
    }

    /// Waits until a deleted changefeed is no longer returned by TiCDC.
    pub async fn wait_changefeed_removed(&self, changefeed_id: &str) {
        let client = self.client.clone();
        let url = self.changefeed_query_url(changefeed_id);
        let changefeed_id = changefeed_id.to_string();
        let display_changefeed_id = changefeed_id.clone();
        try_wait_result_async(
            move || {
                let client = client.clone();
                let url = url.clone();
                let changefeed_id = changefeed_id.clone();
                Box::pin(async move {
                    let resp = dispatch_http_raw(&client, Method::GET, &url, None).await?;
                    if is_changefeed_removed_response(&resp, &changefeed_id) {
                        Ok(())
                    } else {
                        Err(format!(
                            "changefeed {} still exists: status {}, body {}",
                            changefeed_id, resp.status, resp.body
                        ))
                    }
                })
            },
            CHANGEFEED_WAIT_TIMEOUT.as_secs() as usize,
        )
        .await
        .unwrap_or_else(|e| {
            panic!(
                "wait changefeed {display_changefeed_id} removed failed: {}",
                e
            )
        });
    }

    fn changefeed_list_url(&self) -> String {
        let mut url = Url::parse(&format!("http://{}/api/v2/changefeeds", self.cdc_addr)).unwrap();
        url.query_pairs_mut()
            .append_pair("keyspace", &self.keyspace_name);
        url.into()
    }

    fn changefeed_query_url(&self, changefeed_id: &str) -> String {
        let mut url = Url::parse(&format!("http://{}/api/v2/changefeeds", self.cdc_addr)).unwrap();
        url.path_segments_mut().unwrap().push(changefeed_id);
        url.query_pairs_mut()
            .append_pair("keyspace", &self.keyspace_name);
        url.into()
    }
}

async fn query_changefeed(
    client: &HttpClient,
    url: &str,
) -> std::result::Result<ChangefeedDetailInfo, String> {
    let body = dispatch_http(client, Method::GET, url, None).await?;
    serde_json::from_str::<ChangefeedDetailInfo>(&body)
        .map_err(|e| format!("parse changefeed response failed: url {url}, body {body}, err {e}"))
}

#[derive(Debug)]
struct HttpResponseText {
    status: StatusCode,
    body: String,
}

async fn dispatch_http(
    client: &HttpClient,
    method: Method,
    url: &str,
    body: Option<String>,
) -> std::result::Result<String, String> {
    let resp = dispatch_http_raw(client, method.clone(), url, body).await?;
    if resp.status.is_success() {
        Ok(resp.body)
    } else {
        Err(format!(
            "{method} {url}: status {}, body {}",
            resp.status, resp.body
        ))
    }
}

async fn dispatch_http_raw(
    client: &HttpClient,
    method: Method,
    url: &str,
    body: Option<String>,
) -> std::result::Result<HttpResponseText, String> {
    let req = Request::builder()
        .method(method.as_str())
        .uri(url)
        .body(body.unwrap_or_default().into())
        .unwrap();
    let resp = client
        .request(req)
        .await
        .map_err(|e| format!("{method} {url}: failed to send request: {e}"))?;
    let status = resp.status();
    let body = hyper::body::to_bytes(resp.into_body())
        .await
        .map_err(|e| format!("{method} {url}: failed to read response body: {e}"))?;
    let body = String::from_utf8(body.to_vec())
        .map_err(|e| format!("{method} {url}: invalid utf-8 response: {e}"))?;
    Ok(HttpResponseText { status, body })
}

#[derive(Debug, Deserialize)]
struct ChangefeedListResp {
    total: usize,
    items: Vec<serde_json::Value>,
}

#[derive(Debug, Deserialize)]
struct ChangefeedDetailInfo {
    id: String,
    state: String,
    checkpoint_ts: u64,
    error: Option<serde_json::Value>,
}

#[derive(Debug, Deserialize)]
struct TiCdcOpenApiError {
    error_msg: String,
    error_code: String,
}

impl TiCdcOpenApiError {
    fn is_initializing_error(&self) -> bool {
        match self.error_code.as_str() {
            ERR_OWNER_NOT_FOUND => true,
            _ => self.error_msg == MSG_COORDINATOR_NOT_INITIALIZED,
        }
    }

    fn is_start_ts_larger_than_current_tso_error(&self) -> bool {
        self.error_code == ERR_API_INVALID_PARAM
            && self.error_msg.contains("invalid start-ts")
            && self.error_msg.contains("larger than current tso")
    }

    fn is_changefeed_already_exists_error(&self) -> bool {
        self.error_code == ERR_CHANGEFEED_ALREADY_EXISTS
    }

    fn is_changefeed_not_exists_error(&self, changefeed_id: &str) -> bool {
        self.error_code == ERR_CHANGEFEED_NOT_EXISTS
            && self
                .error_msg
                .contains(&format!("changefeed not exists, {changefeed_id}"))
    }
}

pub fn is_retryable_create_changefeed_error(body: &str) -> bool {
    let Some(openapi_error) = parse_ticdc_openapi_error(body) else {
        return false;
    };
    openapi_error.is_initializing_error()
        || openapi_error.is_start_ts_larger_than_current_tso_error()
}

pub fn is_create_changefeed_already_exists_error(body: &str) -> bool {
    parse_ticdc_openapi_error(body)
        .map(|openapi_error| openapi_error.is_changefeed_already_exists_error())
        .unwrap_or(false)
}

fn is_retryable_keyspace_query_error(body: &str) -> bool {
    parse_ticdc_openapi_error(body)
        .map(|openapi_error| openapi_error.is_initializing_error())
        .unwrap_or(false)
}

fn is_retryable_delete_changefeed_error(body: &str) -> bool {
    parse_ticdc_openapi_error(body)
        .map(|openapi_error| openapi_error.is_initializing_error())
        .unwrap_or(false)
}

fn is_delete_changefeed_not_exists_error(body: &str, changefeed_id: &str) -> bool {
    parse_ticdc_openapi_error(body)
        .map(|openapi_error| openapi_error.is_changefeed_not_exists_error(changefeed_id))
        .unwrap_or(false)
}

fn is_changefeed_removed_response(resp: &HttpResponseText, changefeed_id: &str) -> bool {
    if resp.status != StatusCode::BAD_REQUEST {
        return false;
    }
    let Some(openapi_error) = serde_json::from_str::<TiCdcOpenApiError>(&resp.body).ok() else {
        return false;
    };
    openapi_error.is_changefeed_not_exists_error(changefeed_id)
}

fn parse_ticdc_openapi_error(body: &str) -> Option<TiCdcOpenApiError> {
    serde_json::from_str::<TiCdcOpenApiError>(body).ok()
}

fn mysql_sink_uri(downstream: &crate::tidb::ConnParams) -> String {
    let mut url = Url::parse("mysql://localhost").unwrap();
    url.set_username(&downstream.user)
        .expect("mysql sink uri must accept username");
    if downstream.password.is_empty() {
        url.set_password(None)
            .expect("mysql sink uri must clear password");
    } else {
        url.set_password(Some(&downstream.password))
            .expect("mysql sink uri must accept password");
    }
    url.set_host(Some(&downstream.host))
        .expect("mysql sink uri must accept host");
    url.set_port(Some(downstream.port))
        .expect("mysql sink uri must accept port");
    url.into()
}

pub struct TestTiCdc {
    addr: String,
    shutdown_tx: Option<oneshot::Sender<()>>,
    handle: Option<thread::JoinHandle<()>>,
}

impl TestTiCdc {
    pub fn start() -> Self {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap().to_string();
        let (shutdown_tx, shutdown_rx) = oneshot::channel();

        let handle = thread::spawn(move || {
            let rt = tokio::runtime::Builder::new_multi_thread()
                .thread_name("test_ticdc")
                .worker_threads(2)
                .enable_all()
                .build()
                .unwrap();
            rt.block_on(async move {
                let make_svc = hyper::service::make_service_fn(|_| async {
                    Ok::<_, hyper::Error>(service_fn(|req: Request<Body>| async move {
                        let (parts, _body) = req.into_parts();
                        match (parts.method, parts.uri.path()) {
                            (Method::GET, "/api/v2/status") => Ok::<_, hyper::Error>(
                                Response::new(Body::from(r#"{"liveness":0}"#)),
                            ),
                            _ => Ok::<_, hyper::Error>(
                                Response::builder()
                                    .status(StatusCode::NOT_FOUND)
                                    .body(Body::from("Not Found"))
                                    .unwrap(),
                            ),
                        }
                    }))
                });

                let server = hyper::Server::from_tcp(listener).unwrap().serve(make_svc);

                tokio::select! {
                    _ = server => {}
                    _ = shutdown_rx => {}
                }
            })
        });

        Self {
            addr,
            shutdown_tx: Some(shutdown_tx),
            handle: Some(handle),
        }
    }

    pub fn addr(&self) -> &str {
        &self.addr
    }
}

impl Drop for TestTiCdc {
    fn drop(&mut self) {
        if let Some(tx) = self.shutdown_tx.take() {
            let _ = tx.send(());
        }
        if let Some(handle) = self.handle.take() {
            let _ = handle.join();
        }
    }
}

pub struct KeyspaceTestService {
    keyspace_id: u32,
    merged_store_id: u64,
    advertise_addr: String,
    pd_url: String,
    cdc_addr: String,

    pd: Option<PdWrapper>,
    cdc: Option<TestTiCdc>,

    states: KeyspaceStates,
    pd_client: Arc<TestPdClient>,
}

impl KeyspaceTestService {
    pub fn new(keyspace_id: u32, merged_store_id: u64, advertise_addr: String) -> Self {
        let pd = PdWrapper::new_test(1, &SecurityConfig::default(), None);
        let pd_client = pd.test_client().unwrap();
        let cdc = TestTiCdc::start();
        let pd_url = pd
            .endpoints()
            .and_then(|e| e.first())
            .expect("test PD must have endpoints")
            .clone();

        let cdc_addr = cdc.addr().to_string();
        let states = KeyspaceStates::new_provisioned(pd_url.clone(), cdc_addr.clone());

        Self {
            keyspace_id,
            merged_store_id,
            advertise_addr,
            pd_url,
            cdc_addr,
            pd: Some(pd),
            cdc: Some(cdc),
            states,
            pd_client,
        }
    }

    pub fn pd_url(&self) -> &str {
        &self.pd_url
    }

    pub fn cdc_addr(&self) -> &str {
        &self.cdc_addr
    }

    pub fn pd_client_ext(&self) -> Arc<dyn PdClientExt> {
        self.pd_client.clone()
    }

    pub fn new_cluster_client(&self) -> ClusterClient {
        let env: Arc<Environment> = Arc::new(EnvBuilder::new().cq_count(1).build());
        let mut channels = HashMap::default();
        let stores = self.pd_client.get_all_stores(true).expect("get_all_stores");
        for store in stores {
            let channel = ChannelBuilder::new(env.clone())
                .keepalive_time(Duration::from_secs(1))
                .keepalive_timeout(Duration::from_secs(1))
                .max_reconnect_backoff(Duration::from_secs(1))
                .initial_reconnect_backoff(Duration::from_secs(1))
                .connect(store.get_address());
            channels.insert(store.get_id(), channel);
        }

        ClusterClient {
            opts: ClusterClientOptions::default(),
            pd_client: self.pd_client.clone(),
            channels,
            region_ranges: Default::default(),
            regions: Default::default(),
            ref_store: Arc::new(Mutex::new(RefStore::default())),
            max_ts: Default::default(),
            async_commit: false,
            lock_resolver: None,
            txn_file_helper: None,
        }
    }

    pub async fn new_txn_client(&self) -> tikv_client::TransactionClient {
        tikv_client::TransactionClient::new(vec![self.pd_url.clone()])
            .await
            .unwrap()
    }
}

#[async_trait]
impl KeyspaceService for KeyspaceTestService {
    fn keyspace_id(&self) -> u32 {
        self.keyspace_id
    }

    async fn start(&mut self) -> replication_worker::Result<()> {
        replication_worker::bootstrap(
            self.pd_client.clone(),
            self.merged_store_id,
            self.advertise_addr.clone(),
        )
        .await
    }

    async fn destroy(&mut self) -> replication_worker::Result<()> {
        let _ = self.cdc.take();
        let _ = self.pd.take();
        Ok(())
    }

    fn get_states(&self) -> &KeyspaceStates {
        &self.states
    }

    fn get_states_mut(&mut self) -> &mut KeyspaceStates {
        &mut self.states
    }

    fn get_pd_client(&self) -> Arc<dyn PdClient> {
        self.pd_client.clone()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_changefeed_urls_escape_path_and_query_components() {
        let cli = CdcCli::new(
            HttpClient::Http(hyper::Client::new()),
            "127.0.0.1:8300".to_string(),
            "ks 1/a".to_string(),
        );

        assert_eq!(
            cli.changefeed_list_url(),
            "http://127.0.0.1:8300/api/v2/changefeeds?keyspace=ks+1%2Fa"
        );
        assert_eq!(
            cli.changefeed_query_url("cf 1/a"),
            "http://127.0.0.1:8300/api/v2/changefeeds/cf%201%2Fa?keyspace=ks+1%2Fa"
        );
    }

    #[test]
    fn test_changefeed_detail_response_uses_checkpoint_ts() {
        let info = serde_json::from_str::<ChangefeedDetailInfo>(
            r#"{
                "id": "cf-1",
                "state": "normal",
                "checkpoint_ts": 42,
                "error": null
            }"#,
        )
        .unwrap();

        assert_eq!(info.id, "cf-1");
        assert_eq!(info.state, "normal");
        assert_eq!(info.checkpoint_ts, 42);
        assert_eq!(info.error, None);
    }

    #[test]
    fn test_changefeed_detail_response_rejects_missing_checkpoint_ts() {
        let err = serde_json::from_str::<ChangefeedDetailInfo>(
            r#"{
                "id": "cf-1",
                "state": "normal",
                "checkpoint_tso": 42,
                "error": null
            }"#,
        )
        .unwrap_err();

        assert!(
            err.to_string().contains("checkpoint_ts"),
            "unexpected parse error: {}",
            err
        );
    }

    #[test]
    fn test_changefeed_removed_response_cases() {
        let cases = vec![
            (
                "removed error json",
                HttpResponseText {
                    status: StatusCode::BAD_REQUEST,
                    body: r#"{
    "error_msg": "[CDC:ErrChangeFeedNotExists]changefeed not exists, random-cdc-000001-ks1",
    "error_code": "CDC:ErrChangeFeedNotExists"
}"#
                    .to_string(),
                },
                "random-cdc-000001-ks1",
                true,
            ),
            (
                "non-json not found",
                HttpResponseText {
                    status: StatusCode::NOT_FOUND,
                    body: "Not Found".to_string(),
                },
                "cf-1",
                false,
            ),
            (
                "success list body",
                HttpResponseText {
                    status: StatusCode::OK,
                    body: r#"{"items":[],"total":0}"#.to_string(),
                },
                "cf-1",
                false,
            ),
            (
                "different changefeed id",
                HttpResponseText {
                    status: StatusCode::BAD_REQUEST,
                    body: r#"{
    "error_msg": "[CDC:ErrChangeFeedNotExists]changefeed not exists, other-cf",
    "error_code": "CDC:ErrChangeFeedNotExists"
}"#
                    .to_string(),
                },
                "cf-1",
                false,
            ),
        ];

        for (name, resp, changefeed_id, expected) in cases {
            assert_eq!(
                is_changefeed_removed_response(&resp, changefeed_id),
                expected,
                "{name}"
            );
        }
    }

    #[test]
    fn test_retryable_create_changefeed_error_cases() {
        let cases = vec![
            (
                "owner not found",
                r#"{
    "error_msg": "[CDC:ErrOwnerNotFound]owner not found",
    "error_code": "CDC:ErrOwnerNotFound"
}"#,
                true,
                false,
            ),
            (
                "start ts race",
                r#"{
    "error_msg": "[CDC:ErrAPIInvalidParam]invalid start-ts 10, larger than current tso 9",
    "error_code": "CDC:ErrAPIInvalidParam"
}"#,
                true,
                false,
            ),
            (
                "json initializing error",
                r#"{
    "error_msg": "coordinator is not fully initialized, wait a moment",
    "error_code": ""
}"#,
                true,
                false,
            ),
            (
                "already exists succeeds but is not retryable",
                r#"{
    "error_msg": "[CDC:ErrChangeFeedAlreadyExists]changefeed already exists, cf-1",
    "error_code": "CDC:ErrChangeFeedAlreadyExists"
}"#,
                false,
                true,
            ),
            (
                "non-json server not ready",
                "server is not ready",
                false,
                false,
            ),
            (
                "non-json initializing error",
                MSG_COORDINATOR_NOT_INITIALIZED,
                false,
                false,
            ),
            (
                "formatted dispatch error is not parsed",
                r#"POST http://127.0.0.1:8300/api/v2/changefeeds: status 500 Internal Server Error, body {
    "error_msg": "[CDC:ErrOwnerNotFound]owner not found",
    "error_code": "CDC:ErrOwnerNotFound"
}"#,
                false,
                false,
            ),
        ];

        for (name, err, retryable, already_exists) in cases {
            assert_eq!(
                is_retryable_create_changefeed_error(err),
                retryable,
                "retryable: {name}"
            );
            assert_eq!(
                is_create_changefeed_already_exists_error(err),
                already_exists,
                "already exists: {name}"
            );
        }
    }

    #[test]
    fn test_delete_changefeed_error_cases() {
        let cases = vec![
            (
                "owner not found",
                r#"{
    "error_msg": "[CDC:ErrOwnerNotFound]owner not found",
    "error_code": "CDC:ErrOwnerNotFound"
}"#,
                "cf-1",
                true,
                false,
            ),
            (
                "json initializing error",
                r#"{
    "error_msg": "coordinator is not fully initialized, wait a moment",
    "error_code": ""
}"#,
                "cf-1",
                true,
                false,
            ),
            (
                "already deleted",
                r#"{
    "error_msg": "[CDC:ErrChangeFeedNotExists]changefeed not exists, cf-1",
    "error_code": "CDC:ErrChangeFeedNotExists"
}"#,
                "cf-1",
                false,
                true,
            ),
            (
                "different changefeed id",
                r#"{
    "error_msg": "[CDC:ErrChangeFeedNotExists]changefeed not exists, other-cf",
    "error_code": "CDC:ErrChangeFeedNotExists"
}"#,
                "cf-1",
                false,
                false,
            ),
            (
                "non-json server not ready",
                "server is not ready",
                "cf-1",
                false,
                false,
            ),
        ];

        for (name, err, changefeed_id, retryable, not_exists) in cases {
            assert_eq!(
                is_retryable_delete_changefeed_error(err),
                retryable,
                "retryable: {name}"
            );
            assert_eq!(
                is_delete_changefeed_not_exists_error(err, changefeed_id),
                not_exists,
                "not exists: {name}"
            );
        }
    }

    #[test]
    fn test_mysql_sink_uri_encodes_credentials() {
        let params = crate::tidb::ConnParams {
            host: "127.0.0.1".to_string(),
            port: 4000,
            user: "user:name".to_string(),
            password: "p@ss:word".to_string(),
        };

        assert_eq!(
            mysql_sink_uri(&params),
            "mysql://user%3Aname:p%40ss%3Aword@127.0.0.1:4000"
        );
    }
}