// Copyright 2025 PingCAP, Inc.

use std::{borrow::Cow, collections::HashMap, sync::Arc, time::Duration};

use bytes::{Buf, Bytes};
use http::{HeaderMap, HeaderValue};
use hyper::{Body, Method, Request, Response, Result, StatusCode, Uri};
use merged_engine::ForceStop;
use security::HttpClient;
use serde::{Deserialize, Serialize};
use tikv_util::{box_err, error, future::paired_future_callback, info, warn};

use crate::{
    CdcMsg, Error, UnsafeRecoverClearTarget,
    ticdc_util::{format_changefeed_summary, list_changefeeds_from_ticdc},
};

#[derive(Clone)]
pub struct ReplicationScheduler {
    sender: tikv_util::mpsc::Sender<CdcMsg>,
    cdc_addrs: Arc<dashmap::DashMap<u32, String>>,
    add_keyspace_inflight: Arc<dashmap::DashSet<u32>>,
    http_client: Arc<HttpClient>,
    scheme: String,
    sync_ticdc_timeout: Duration,
    #[allow(dead_code)]
    force_stop: ForceStop,
}

impl ReplicationScheduler {
    pub fn new(
        sender: tikv_util::mpsc::Sender<CdcMsg>,
        cdc_addrs: Arc<dashmap::DashMap<u32, String>>,
        http_client: Arc<HttpClient>,
        sync_ticdc_timeout: Duration,
        force_stop: ForceStop,
    ) -> Self {
        let scheme = if matches!(http_client.as_ref(), HttpClient::Http(_)) {
            "http".to_string()
        } else {
            "https".to_string()
        };
        ReplicationScheduler {
            sender,
            cdc_addrs,
            add_keyspace_inflight: Arc::new(dashmap::DashSet::new()),
            http_client,
            scheme,
            sync_ticdc_timeout,
            force_stop,
        }
    }

    pub fn schedule(&self, msg: CdcMsg) {
        let _ = self.sender.send(msg);
    }

    pub fn get_cdc_addr(&self, keyspace_id: u32) -> Option<String> {
        self.cdc_addrs
            .get(&keyspace_id)
            .map(|addr| addr.value().clone())
    }

    pub fn force_stop(&self) {
        #[cfg(feature = "testexport")]
        self.force_stop.set();
    }
}

#[derive(Serialize)]
struct ErrorResponse {
    error_msg: String,
    error_code: String,
}

#[derive(Serialize, Deserialize, Clone, Debug, Default)]
pub struct ChangefeedRequest {
    pub changefeed_id: String,
    pub sink_uri: String,
    #[serde(default)]
    pub start_ts: u64, // `0`: use current time.
}

#[derive(Serialize)]
#[allow(dead_code)]
struct ErrorInfo {
    addr: String,
    code: String,
    message: String,
}

#[derive(Serialize)]
#[allow(dead_code)]
struct TaskStatus {
    capture_id: String,
    table_ids: Vec<u64>,
}

#[derive(Serialize)]
struct UnsafeRecoverClearResponse {
    success: bool,
    dry_run: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    region_ids: Option<Vec<u64>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    reason: Option<String>,
}

pub async fn handle_cdc_request(
    scheduler: Option<&ReplicationScheduler>,
    req: Request<Body>,
) -> Result<Response<Body>> {
    if let Some(scheduler) = scheduler {
        scheduler.handle_http_request(req).await
    } else {
        Ok(Response::builder()
            .status(StatusCode::SERVICE_UNAVAILABLE)
            .body("replication worker is disabled".into())
            .unwrap())
    }
}

impl ReplicationScheduler {
    fn error_response(status: StatusCode, msg: &str, code: &str) -> Response<Body> {
        let error = ErrorResponse {
            error_msg: msg.to_string(),
            error_code: code.to_string(),
        };
        Response::builder()
            .status(status)
            .body(Body::from(serde_json::to_string(&error).unwrap()))
            .unwrap()
    }

    pub async fn handle_http_request(&self, req: Request<Body>) -> Result<Response<Body>> {
        let (parts, body) = req.into_parts();
        let method = &parts.method;
        let uri = &parts.uri;
        let path = uri.path();
        let headers = parts.headers;

        if *method == Method::GET && path == "/cdc/keyspace" {
            return Ok(self.handle_get_keyspaces().await);
        }

        let query = uri.query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();

        if path == "/cdc/unsafe_recover/clear" {
            return Ok(self.handle_unsafe_recover_clear(method, &query_pairs).await);
        }

        let keyspace_id = {
            if let Some(keyspace_str) = query_pairs.get("keyspace_id").map(|v| v.as_ref()) {
                match keyspace_str.parse::<u32>() {
                    Ok(keyspace_id) => keyspace_id,
                    Err(_) => {
                        return Ok(Self::error_response(
                            StatusCode::BAD_REQUEST,
                            "Invalid keyspace_id",
                            "CDC:ErrInvalidParam",
                        ));
                    }
                }
            } else {
                return Ok(Self::error_response(
                    StatusCode::BAD_REQUEST,
                    "keyspace_id is required",
                    "CDC:ErrInvalidParam",
                ));
            }
        };

        let response = match (method, path) {
            (&Method::POST, "/cdc/api/v2/changefeeds") => {
                let body_bytes = hyper::body::to_bytes(body).await?;
                self.handle_create_changefeed(keyspace_id, body_bytes).await
            }
            (&Method::POST, "/cdc/keyspace") => {
                let body_bytes = hyper::body::to_bytes(body).await?;
                self.handle_add_keyspace(keyspace_id, &body_bytes).await
            }
            (&Method::DELETE, "/cdc/keyspace") => {
                let force = match extract_bool_param(&query_pairs, "force") {
                    Ok(force) => force,
                    Err(err) => return Ok(err),
                };
                self.handle_remove_keyspace(keyspace_id, force).await
            }
            (&Method::DELETE, path) if path.starts_with("/cdc/api/v2/changefeeds/") => {
                self.handle_delete_changefeed(keyspace_id, path).await?
            }
            (method, path) if path.starts_with("/cdc/") => {
                self.forward_request_to_cdc(keyspace_id, method, uri, headers, body)
                    .await?
            }
            _ => {
                info!("Invalid request {} {}", method, uri);
                Self::error_response(
                    StatusCode::NOT_FOUND,
                    "Route not found",
                    "CDC:ErrAPIRouteNotFound",
                )
            }
        };

        Ok(response)
    }

    async fn handle_delete_changefeed(
        &self,
        keyspace_id: u32,
        path: &str,
    ) -> Result<Response<Body>> {
        if let Some(changefeed_id) = path.strip_prefix("/cdc/api/v2/changefeeds/") {
            let cdc_addr = self.get_cdc_addr(keyspace_id);
            if cdc_addr.is_none() {
                return Ok(Self::error_response(
                    StatusCode::NOT_FOUND,
                    "No CDC server found for the keyspace",
                    "CDC:ErrNoCDCServer",
                ));
            }
            let cdc_addr = cdc_addr.unwrap();
            let remove_cdc_task_uri = format!(
                "{}://{}/api/v2/changefeeds/{}",
                &self.scheme, &cdc_addr, changefeed_id
            );

            let req = http::Request::delete(remove_cdc_task_uri)
                .body(hyper::Body::empty())
                .unwrap();
            if let Err(err) = self.http_client.request(req).await {
                error!("delete changefeed from TiCDC failed"; "err" => ?err);
                return Ok(Self::error_response(
                    StatusCode::BAD_GATEWAY,
                    &format!("forward request to TiCDC failed: {err}"),
                    "CDC:ErrForwardTiCDC",
                ));
            }
            let (cb, fut) = paired_future_callback();
            self.schedule(CdcMsg::RemoveTask {
                keyspace_id,
                changefeed_id: changefeed_id.to_string(),
                cb,
            });
            let res = fut.await.unwrap();
            if let Err(err) = res {
                return Ok(Self::error_response(
                    StatusCode::BAD_REQUEST,
                    &err.to_string(),
                    "CDC:ErrRemoveChangefeed",
                ));
            }
            Ok(Response::builder()
                .status(StatusCode::OK)
                .body(Body::empty())
                .unwrap())
        } else {
            Ok(Self::error_response(
                StatusCode::BAD_REQUEST,
                "Invalid changefeed ID",
                "CDC:ErrInvalidChangefeedID",
            ))
        }
    }

    async fn forward_request_to_cdc(
        &self,
        keyspace_id: u32,
        method: &Method,
        uri: &Uri,
        headers: HeaderMap<HeaderValue>,
        body: Body,
    ) -> Result<Response<Body>> {
        let cdc_addr = self.cdc_addrs.get(&keyspace_id).map(|r| r.value().clone());
        if cdc_addr.is_none() {
            return Ok(Self::error_response(
                StatusCode::NOT_FOUND,
                "No CDC server found for the keyspace",
                "CDC:ErrNoCDCServer",
            ));
        }
        let new_path_query = uri
            .path_and_query()
            .unwrap()
            .to_string()
            .replace("/cdc/", "/");
        let cdc_addr = cdc_addr.unwrap();
        let new_uri = Uri::builder()
            .scheme(self.scheme.as_str())
            .authority(cdc_addr.as_str())
            .path_and_query(new_path_query)
            .build()
            .unwrap();

        let mut req_builder = Request::builder().method(method).uri(new_uri);

        for (key, value) in headers.iter() {
            req_builder = req_builder.header(key, value);
        }

        let req = req_builder.body(body).unwrap();

        match self.http_client.request(req).await {
            Ok(resp) => Ok(resp),
            Err(err) => {
                error!("forward request to TiCDC failed"; "err" => ?err);
                Ok(Self::error_response(
                    StatusCode::BAD_GATEWAY,
                    &format!("forward request to TiCDC failed: {err}"),
                    "CDC:ErrForwardTiCDC",
                ))
            }
        }
    }

    async fn handle_create_changefeed(&self, keyspace_id: u32, body: Bytes) -> Response<Body> {
        info!("handle create change feed");
        match serde_json::from_slice::<ChangefeedRequest>(body.chunk()) {
            Ok(request) => {
                if request.changefeed_id.is_empty() {
                    return Self::error_response(
                        StatusCode::BAD_REQUEST,
                        "changefeed_id is required",
                        "CDC:ErrInvalidRequestBody",
                    );
                }
                if request.sink_uri.is_empty() {
                    return Self::error_response(
                        StatusCode::BAD_REQUEST,
                        "sink_uri is required",
                        "CDC:ErrInvalidRequestBody",
                    );
                }
                let (cb, fut) = paired_future_callback();

                self.schedule(CdcMsg::NewTask {
                    keyspace_id,
                    changefeed_id: request.changefeed_id,
                    start_ts: request.start_ts,
                    body,
                    cb,
                });
                let res = fut.await.unwrap();
                if let Err(err) = res {
                    return Self::error_response(
                        StatusCode::SERVICE_UNAVAILABLE,
                        &err.to_string(),
                        "CDC:ErrCreateChangefeed",
                    );
                }
                let (status, body) = res.unwrap();
                Response::builder()
                    .status(status)
                    .body(body.into())
                    .unwrap()
            }
            Err(e) => Self::error_response(
                StatusCode::BAD_REQUEST,
                &e.to_string(),
                "CDC:ErrInvalidRequestBody",
            ),
        }
    }

    async fn handle_add_keyspace(&self, keyspace_id: u32, body_bytes: &[u8]) -> Response<Body> {
        if !self.add_keyspace_inflight.insert(keyspace_id) {
            info!("handle add keyspace: reject duplicated"; "keyspace" => keyspace_id);
            return Self::error_response(
                StatusCode::CONFLICT,
                &format!("keyspace {keyspace_id} add already in progress"),
                "CDC:ErrAddKeyspaceInProgress",
            );
        }

        info!("handle add keyspace"; "keyspace" => keyspace_id);
        let (orig_cb, fut) = paired_future_callback();
        let inflight = self.add_keyspace_inflight.clone();
        let cb = Box::new(move |res| {
            inflight.remove(&keyspace_id);
            orig_cb(res);
        });
        let msg = match serde_json::from_slice::<crate::scheduler::ProvisionedKeyspace>(body_bytes)
        {
            Ok(provisioned) => CdcMsg::AddKeyspace {
                keyspace_id,
                pd_url: provisioned.pd_url,
                cdc_addr: provisioned.cdc_addr,
                cb,
            },
            Err(_err) => CdcMsg::AddKeyspace {
                keyspace_id,
                pd_url: "".into(),
                cdc_addr: "".into(),
                cb,
            },
        };
        self.schedule(msg);
        let res = fut.await.unwrap();
        if let Err(err) = res {
            return Self::error_response(
                StatusCode::BAD_REQUEST,
                &err.to_string(),
                "CDC:ErrAddKeyspace",
            );
        }
        Response::builder()
            .status(StatusCode::OK)
            .body(Body::empty())
            .unwrap()
    }

    async fn handle_remove_keyspace(&self, keyspace_id: u32, force: bool) -> Response<Body> {
        info!("handle remove keyspace"; "keyspace" => keyspace_id, "force" => force);

        if let Some(cdc_addr) = self.get_cdc_addr(keyspace_id) {
            match list_changefeeds_from_ticdc(
                keyspace_id,
                &cdc_addr,
                self.http_client.as_ref(),
                self.sync_ticdc_timeout,
                "remove_keyspace_changefeed_check",
            )
            .await
            {
                Ok(changefeeds) if !changefeeds.items.is_empty() => {
                    let summary = format_changefeed_summary(&changefeeds.items);
                    warn!("remove_keyspace: TiCDC changefeeds not empty";
                        "keyspace" => keyspace_id,
                        "changefeeds" => summary.as_str(),
                        "force" => force);
                    if !force {
                        return Self::error_response(
                            StatusCode::BAD_REQUEST,
                            &format!("changefeeds not empty in TiCDC: {summary}"),
                            "CDC:ErrRemoveKeyspace",
                        );
                    }
                }
                Ok(_) => {}
                Err(err) => {
                    warn!("remove_keyspace: TiCDC changefeed check failed: {:?}", err;
                        "keyspace" => keyspace_id,
                        "force" => force);
                    if !force {
                        return Self::error_response(
                            StatusCode::BAD_REQUEST,
                            &format!("failed to query TiCDC changefeeds: {err}"),
                            "CDC:ErrRemoveKeyspace",
                        );
                    }
                }
            }
        }

        // This TiCDC precheck is intentionally best-effort. A concurrent
        // create-changefeed request can still win after the list call and before
        // local teardown; that remove/add race is acceptable for now.
        let (cb, fut) = paired_future_callback();
        self.schedule(CdcMsg::RemoveKeyspace { keyspace_id, cb });
        let res = fut.await.unwrap();
        if let Err(err) = res {
            return Self::error_response(
                StatusCode::BAD_REQUEST,
                &err.to_string(),
                "CDC:ErrRemoveKeyspace",
            );
        }
        Response::builder()
            .status(StatusCode::OK)
            .body(Body::empty())
            .unwrap()
    }

    async fn handle_get_keyspaces(&self) -> Response<Body> {
        info!("handle get keyspaces");
        let (cb, fut) = paired_future_callback();
        self.schedule(CdcMsg::GetKeyspaces { cb });
        let res = fut.await.unwrap();
        let keyspaces_resp = KeyspacesResp { keyspace_ids: res };
        let res_str = serde_json::to_string(&keyspaces_resp).unwrap();
        Response::builder()
            .status(StatusCode::OK)
            .body(res_str.into())
            .unwrap()
    }

    async fn handle_unsafe_recover_clear(
        &self,
        method: &Method,
        query_pairs: &HashMap<Cow<'_, str>, Cow<'_, str>>,
    ) -> Response<Body> {
        let dry_run = match parse_bool_param(query_pairs, "dry_run") {
            Ok(dry_run) => dry_run,
            Err(err) => {
                return unsafe_recover_clear_response(
                    StatusCode::BAD_REQUEST,
                    false,
                    false,
                    None,
                    Some(err),
                );
            }
        };

        if *method != Method::POST {
            return unsafe_recover_clear_response(
                StatusCode::BAD_REQUEST,
                dry_run,
                false,
                None,
                Some("POST is required".to_string()),
            );
        }

        let target = match (
            query_pairs.get("region_id").map(|v| v.as_ref()),
            query_pairs.get("keyspace_id").map(|v| v.as_ref()),
        ) {
            (Some(_), Some(_)) | (None, None) => {
                return unsafe_recover_clear_response(
                    StatusCode::BAD_REQUEST,
                    dry_run,
                    false,
                    None,
                    Some("exactly one of region_id or keyspace_id is required".to_string()),
                );
            }
            (Some(region_id), None) => match region_id.parse::<u64>() {
                Ok(region_id) => UnsafeRecoverClearTarget::Region(region_id),
                Err(_) => {
                    return unsafe_recover_clear_response(
                        StatusCode::BAD_REQUEST,
                        dry_run,
                        false,
                        None,
                        Some("Invalid region_id".to_string()),
                    );
                }
            },
            (None, Some(keyspace_id)) => match keyspace_id.parse::<u32>() {
                Ok(keyspace_id) => UnsafeRecoverClearTarget::Keyspace(keyspace_id),
                Err(_) => {
                    return unsafe_recover_clear_response(
                        StatusCode::BAD_REQUEST,
                        dry_run,
                        false,
                        None,
                        Some("Invalid keyspace_id".to_string()),
                    );
                }
            },
        };

        let (cb, fut) = paired_future_callback();
        self.schedule(CdcMsg::UnsafeRecoverClear {
            target,
            dry_run,
            cb,
        });
        let res = match fut.await {
            Ok(res) => res,
            Err(err) => {
                error!(
                    "unsafe recover clear response callback dropped";
                    "target" => ?target,
                    "dry_run" => dry_run,
                    "err" => ?err,
                );
                return unsafe_recover_clear_response(
                    StatusCode::SERVICE_UNAVAILABLE,
                    dry_run,
                    false,
                    None,
                    Some(
                        "replication worker unavailable while handling unsafe recover clear"
                            .to_string(),
                    ),
                );
            }
        };
        match res {
            Ok(region_ids) => {
                unsafe_recover_clear_response(StatusCode::OK, dry_run, true, Some(region_ids), None)
            }
            Err(Error::PdTimeout(reason)) => unsafe_recover_clear_response(
                StatusCode::SERVICE_UNAVAILABLE,
                dry_run,
                false,
                None,
                Some(reason),
            ),
            Err(Error::UnsafeRecoverClearNotFound(reason)) => unsafe_recover_clear_response(
                StatusCode::NOT_FOUND,
                dry_run,
                false,
                None,
                Some(reason),
            ),
            Err(Error::UnsafeRecoverClearRejected(reason)) => unsafe_recover_clear_response(
                StatusCode::BAD_REQUEST,
                dry_run,
                false,
                None,
                Some(reason),
            ),
            Err(err) => unsafe_recover_clear_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                dry_run,
                false,
                None,
                Some(err.to_string()),
            ),
        }
    }
}

#[derive(Clone, Serialize, Deserialize, PartialEq, Debug, Default)]
#[serde(default)]
pub struct ProvisionedKeyspace {
    pub pd_url: String,
    pub cdc_addr: String,
}

#[derive(Clone, Serialize, Deserialize, PartialEq, Debug, Default)]
#[serde(default)]
pub struct KeyspacesResp {
    pub keyspace_ids: Vec<u32>,
}

fn parse_bool_param(
    query_pairs: &HashMap<Cow<'_, str>, Cow<'_, str>>,
    key: &str,
) -> std::result::Result<bool, String> {
    match query_pairs.get(key).map(|v| v.as_ref()) {
        None => Ok(false),
        Some("") | Some("0") | Some("false") => Ok(false),
        Some("1") | Some("true") => Ok(true),
        Some(_) => Err(format!("Invalid boolean param \"{key}\"")),
    }
}

fn extract_bool_param(
    query_pairs: &HashMap<Cow<'_, str>, Cow<'_, str>>,
    key: &str,
) -> std::result::Result<bool, Response<Body>> {
    parse_bool_param(query_pairs, key).map_err(|msg| {
        ReplicationScheduler::error_response(StatusCode::BAD_REQUEST, &msg, "CDC:ErrInvalidParam")
    })
}

#[derive(Clone, Serialize, Deserialize, PartialEq, Debug, Default)]
#[serde(default)]
pub struct CdcStatus {
    pub liveness: usize,
}

pub(crate) async fn get_cdc_status(
    client: &HttpClient,
    cdc_addr: &str,
    timeout: Duration,
) -> crate::Result<()> {
    let scheme = if matches!(client, HttpClient::Http(_)) {
        "http"
    } else {
        "https"
    };
    let url = format!("{}://{}/api/v2/status", scheme, cdc_addr);
    let start = tikv_util::time::Instant::now();
    while start.saturating_elapsed() < timeout {
        let uri = Uri::try_from(&url).unwrap();
        let resp = client.get(uri).await;
        let Ok(resp) = resp else {
            tokio::time::sleep(Duration::from_secs(1)).await;
            continue;
        };
        let resp_data = hyper::body::to_bytes(resp.into_body()).await?;
        let status = serde_json::from_slice::<CdcStatus>(&resp_data)
            .map_err(|e| -> Error { box_err!("Failed to parse CDC status: {}", e) })?;
        if status.liveness == 0 {
            return Ok(());
        }
    }
    Err(box_err!("CDC status check timed out"))
}

fn unsafe_recover_clear_response(
    status: StatusCode,
    dry_run: bool,
    success: bool,
    region_ids: Option<Vec<u64>>,
    reason: Option<String>,
) -> Response<Body> {
    let body = UnsafeRecoverClearResponse {
        success,
        dry_run,
        region_ids,
        reason,
    };
    Response::builder()
        .status(status)
        .body(Body::from(serde_json::to_vec(&body).unwrap()))
        .unwrap()
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;

    use hyper::http::request;

    use super::*;

    const VALID_CHANGEFEED_STATES: [&str; 6] =
        ["all", "normal", "stopped", "error", "failed", "finished"];

    fn parse_query(uri: &Uri) -> HashMap<Cow<'_, str>, Cow<'_, str>> {
        let query = uri.query().unwrap_or("");
        url::form_urlencoded::parse(query.as_bytes()).collect()
    }

    fn test_cdc_addr(changefeeds_status: StatusCode, changefeeds_body: &'static str) -> String {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let make_svc = hyper::service::make_service_fn(move |_| async move {
            Ok::<_, hyper::Error>(hyper::service::service_fn(move |req| async move {
                let path = req.uri().path();
                let query_pairs = parse_query(req.uri());
                if let Some(suffix) = path.strip_prefix("/api/v2/changefeeds/") {
                    let status = if suffix.is_empty() {
                        StatusCode::BAD_REQUEST
                    } else {
                        StatusCode::OK
                    };
                    return Ok::<_, hyper::Error>(
                        Response::builder()
                            .status(status)
                            .body(Body::empty())
                            .unwrap(),
                    );
                }

                if path == "/api/v2/changefeeds" {
                    if let Some(state) = query_pairs.get("state") {
                        if !VALID_CHANGEFEED_STATES.contains(&state.as_ref()) {
                            return Ok::<_, hyper::Error>(
                                Response::builder()
                                    .status(StatusCode::BAD_REQUEST)
                                    .body(Body::empty())
                                    .unwrap(),
                            );
                        }
                    }
                    return Ok::<_, hyper::Error>(
                        Response::builder()
                            .status(changefeeds_status)
                            .body(Body::from(changefeeds_body))
                            .unwrap(),
                    );
                }

                Ok::<_, hyper::Error>(
                    Response::builder()
                        .status(StatusCode::NOT_FOUND)
                        .body(Body::empty())
                        .unwrap(),
                )
            }))
        });
        let server = hyper::Server::from_tcp(listener).unwrap().serve(make_svc);
        tokio::spawn(async move {
            let _ = server.await;
        });
        addr.to_string()
    }

    impl ReplicationScheduler {
        fn new_test_with_changefeeds(
            changefeeds_status: StatusCode,
            changefeeds_body: &'static str,
        ) -> Self {
            let (sender, receiver) = tikv_util::mpsc::unbounded();
            std::thread::spawn(move || {
                while let Ok(msg) = receiver.recv() {
                    match msg {
                        CdcMsg::NewTask { cb, .. } => {
                            cb(Ok((StatusCode::OK, Bytes::from_static(b"{}"))));
                        }
                        CdcMsg::RemoveTask { cb, .. }
                        | CdcMsg::RemoveKeyspace { cb, .. }
                        | CdcMsg::AddKeyspace { cb, .. }
                        | CdcMsg::LoadKeyspaceShards { cb, .. } => {
                            cb(Ok(()));
                        }
                        CdcMsg::GetKeyspaces { cb } => {
                            cb(Vec::new());
                        }
                        CdcMsg::LoadKeyspaceShardMetas { cb, .. } => {
                            cb(Ok(merged_engine::KeyspaceShardMetas {
                                scope: merged_engine::KeyspaceShardMetaScope::Pinned,
                                metas: std::collections::HashMap::new(),
                            }));
                        }
                        CdcMsg::UnsafeRecoverClear { target, cb, .. } => {
                            let region_ids = match target {
                                UnsafeRecoverClearTarget::Region(region_id) => vec![region_id],
                                UnsafeRecoverClearTarget::Keyspace(keyspace_id) => {
                                    let fake_region_id = u64::from(keyspace_id) * 100;
                                    (fake_region_id..fake_region_id + 10).collect()
                                }
                            };
                            cb(Ok(region_ids));
                        }
                        _ => {}
                    }
                }
            });
            let cdc_addrs = Arc::new(dashmap::DashMap::new());
            cdc_addrs.insert(1, test_cdc_addr(changefeeds_status, changefeeds_body));
            ReplicationScheduler::new(
                sender,
                cdc_addrs,
                Arc::new(HttpClient::Http(hyper::Client::new())),
                Duration::from_secs(1),
                ForceStop::default(),
            )
        }

        fn new_test() -> Self {
            Self::new_test_with_changefeeds(StatusCode::OK, r#"{"total":0,"items":[]}"#)
        }
    }

    #[tokio::test]
    async fn test_create_changefeed() {
        let worker = ReplicationScheduler::new_test();
        let req = request::Builder::new()
            .method(Method::POST)
            .uri("/cdc/api/v2/changefeeds?keyspace_id=1")
            .body(Body::from(
                r#"{
                    "changefeed_id": "test1",
                    "sink_uri": "blackhole://"
                }"#,
            ))
            .unwrap();

        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn test_list_changefeeds() {
        let worker = ReplicationScheduler::new_test();
        let req = request::Builder::new()
            .method(Method::GET)
            .uri("/cdc/api/v2/changefeeds?keyspace_id=1&state=normal")
            .body(Body::empty())
            .unwrap();

        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn test_get_changefeed() {
        let worker = ReplicationScheduler::new_test();
        let req = request::Builder::new()
            .method(Method::GET)
            .uri("/cdc/api/v2/changefeeds/test1?keyspace_id=1")
            .body(Body::empty())
            .unwrap();

        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn test_delete_changefeed() {
        let worker = ReplicationScheduler::new_test();
        let req = request::Builder::new()
            .method(Method::DELETE)
            .uri("/cdc/api/v2/changefeeds/test1?keyspace_id=1")
            .body(Body::empty())
            .unwrap();

        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn test_remove_keyspace_rejects_non_empty_changefeeds_without_force() {
        let worker = ReplicationScheduler::new_test_with_changefeeds(
            StatusCode::OK,
            r#"{"total":1,"items":[{"id":"cf-1","state":"normal","checkpoint_tso":100,"checkpoint_time":"","error":null}]}"#,
        );
        let req = request::Builder::new()
            .method(Method::DELETE)
            .uri("/cdc/keyspace?keyspace_id=1")
            .body(Body::empty())
            .unwrap();

        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let body = read_json(resp).await;
        assert_eq!(body["error_code"], "CDC:ErrRemoveKeyspace");
        assert_eq!(
            body["error_msg"],
            "changefeeds not empty in TiCDC: cf-1(normal)"
        );
    }

    #[tokio::test]
    async fn test_remove_keyspace_force_bypasses_non_empty_changefeeds() {
        let worker = ReplicationScheduler::new_test_with_changefeeds(
            StatusCode::OK,
            r#"{"total":1,"items":[{"id":"cf-1","state":"normal","checkpoint_tso":100,"checkpoint_time":"","error":null}]}"#,
        );
        let req = request::Builder::new()
            .method(Method::DELETE)
            .uri("/cdc/keyspace?keyspace_id=1&force=true")
            .body(Body::empty())
            .unwrap();

        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn test_remove_keyspace_rejects_ticdc_query_failure_without_force() {
        let worker = ReplicationScheduler::new_test_with_changefeeds(
            StatusCode::SERVICE_UNAVAILABLE,
            r#"{"error_code":"CDC:ErrServerIsNotReady","error_msg":"not ready"}"#,
        );
        let req = request::Builder::new()
            .method(Method::DELETE)
            .uri("/cdc/keyspace?keyspace_id=1")
            .body(Body::empty())
            .unwrap();

        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let body = read_json(resp).await;
        assert_eq!(body["error_code"], "CDC:ErrRemoveKeyspace");
        assert_eq!(
            body["error_msg"],
            "failed to query TiCDC changefeeds: Server is not ready: not ready"
        );
    }

    #[tokio::test]
    async fn test_remove_keyspace_force_bypasses_ticdc_query_failure() {
        let worker = ReplicationScheduler::new_test_with_changefeeds(
            StatusCode::SERVICE_UNAVAILABLE,
            r#"{"error_code":"CDC:ErrServerIsNotReady","error_msg":"not ready"}"#,
        );
        let req = request::Builder::new()
            .method(Method::DELETE)
            .uri("/cdc/keyspace?keyspace_id=1&force=true")
            .body(Body::empty())
            .unwrap();

        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn test_remove_keyspace_without_cdc_addr_skips_precheck() {
        let worker = ReplicationScheduler::new_test();
        worker.cdc_addrs.remove(&1);
        let req = request::Builder::new()
            .method(Method::DELETE)
            .uri("/cdc/keyspace?keyspace_id=1")
            .body(Body::empty())
            .unwrap();

        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn test_invalid_request() {
        let worker = ReplicationScheduler::new_test();
        // Missing keyspace_id for list changefeeds
        let req = request::Builder::new()
            .method(Method::GET)
            .uri("/cdc/api/v2/changefeeds")
            .body(Body::empty())
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);

        // Missing keyspace_id for get changefeed
        let req = request::Builder::new()
            .method(Method::GET)
            .uri("/cdc/api/v2/changefeeds/test1")
            .body(Body::empty())
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);

        // Missing keyspace_id for create changefeed
        let req = request::Builder::new()
            .method(Method::POST)
            .uri("/cdc/api/v2/changefeeds")
            .body(Body::from(
                r#"{
                "changefeed_id": "test1",
                "sink_uri": "blackhole://"
            }"#,
            ))
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);

        // Missing keyspace_id for delete changefeed
        let req = request::Builder::new()
            .method(Method::DELETE)
            .uri("/cdc/api/v2/changefeeds/test1")
            .body(Body::empty())
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        // Invalid state parameter
        let req = request::Builder::new()
            .method(Method::GET)
            .uri("/cdc/api/v2/changefeeds?keyspace_id=1&state=invalid")
            .body(Body::empty())
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);

        // Invalid changefeed ID format
        let req = request::Builder::new()
            .method(Method::GET)
            .uri("/cdc/api/v2/changefeeds/?keyspace_id=1")
            .body(Body::empty())
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);

        // Missing required field in create request
        let req = request::Builder::new()
            .method(Method::POST)
            .uri("/cdc/api/v2/changefeeds?keyspace_id=1")
            .body(Body::from("{}"))
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn test_add_keyspace_deduplicator() {
        let (sender, receiver) = tikv_util::mpsc::unbounded();
        let (gate_tx, gate_rx) = std::sync::mpsc::channel::<()>();
        let (first_received_tx, first_received_rx) = tokio::sync::oneshot::channel::<()>();

        std::thread::spawn(move || {
            let mut first_received_tx = Some(first_received_tx);
            while let Ok(msg) = receiver.recv() {
                if let CdcMsg::AddKeyspace { cb, .. } = msg {
                    if let Some(tx) = first_received_tx.take() {
                        let _ = tx.send(());
                        let _ = gate_rx.recv();
                    }
                    cb(Ok(()));
                }
            }
        });

        let scheduler = ReplicationScheduler::new(
            sender,
            Arc::new(dashmap::DashMap::new()),
            Arc::new(HttpClient::Http(hyper::Client::new())),
            Duration::from_secs(1),
            ForceStop::default(),
        );

        let req1 = request::Builder::new()
            .method(Method::POST)
            .uri("/cdc/keyspace?keyspace_id=1")
            .body(Body::from(r#"{"pd_url":"pd","cdc_addr":"cdc"}"#))
            .unwrap();
        let scheduler1 = scheduler.clone();
        let task1 =
            tokio::spawn(async move { scheduler1.handle_http_request(req1).await.unwrap() });

        // Ensure the first AddKeyspace request has been dispatched and is blocked in
        // the worker.
        let _ = first_received_rx.await;

        let req2 = request::Builder::new()
            .method(Method::POST)
            .uri("/cdc/keyspace?keyspace_id=1")
            .body(Body::from(r#"{"pd_url":"pd","cdc_addr":"cdc"}"#))
            .unwrap();
        let resp2 = scheduler.handle_http_request(req2).await.unwrap();
        assert_eq!(resp2.status(), StatusCode::CONFLICT);
        let body_bytes = hyper::body::to_bytes(resp2.into_body()).await.unwrap();
        let v: serde_json::Value = serde_json::from_slice(&body_bytes).unwrap();
        assert_eq!(v["error_code"], "CDC:ErrAddKeyspaceInProgress");

        // Unblock the first request and ensure it succeeds.
        gate_tx.send(()).unwrap();
        let resp1 = task1.await.unwrap();
        assert_eq!(resp1.status(), StatusCode::OK);

        // After the first request completes, the deduplicator should be cleared.
        let req3 = request::Builder::new()
            .method(Method::POST)
            .uri("/cdc/keyspace?keyspace_id=1")
            .body(Body::from(r#"{"pd_url":"pd","cdc_addr":"cdc"}"#))
            .unwrap();
        let resp3 = scheduler.handle_http_request(req3).await.unwrap();
        assert_eq!(resp3.status(), StatusCode::OK);
    }

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

    #[tokio::test]
    async fn test_unsafe_recover_clear_region() {
        let worker = ReplicationScheduler::new_test();
        let req = request::Builder::new()
            .method(Method::POST)
            .uri("/cdc/unsafe_recover/clear?region_id=1")
            .body(Body::empty())
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let body = read_json(resp).await;
        assert!(body["success"].as_bool().unwrap());
        assert!(!body["dry_run"].as_bool().unwrap());
        assert_eq!(body["region_ids"], serde_json::json!([1]));
    }

    #[tokio::test]
    async fn test_unsafe_recover_clear_keyspace() {
        let worker = ReplicationScheduler::new_test();
        let req = request::Builder::new()
            .method(Method::POST)
            .uri("/cdc/unsafe_recover/clear?keyspace_id=42")
            .body(Body::empty())
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let body = read_json(resp).await;
        assert!(body["success"].as_bool().unwrap());
        assert!(!body["dry_run"].as_bool().unwrap());
        assert_eq!(
            body["region_ids"],
            serde_json::json!([4200, 4201, 4202, 4203, 4204, 4205, 4206, 4207, 4208, 4209])
        );
    }

    #[tokio::test]
    async fn test_unsafe_recover_clear_dry_run() {
        let worker = ReplicationScheduler::new_test();
        let req = request::Builder::new()
            .method(Method::POST)
            .uri("/cdc/unsafe_recover/clear?region_id=1&dry_run=true")
            .body(Body::empty())
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let body = read_json(resp).await;
        assert!(body["success"].as_bool().unwrap());
        assert!(body["dry_run"].as_bool().unwrap());
        assert_eq!(body["region_ids"], serde_json::json!([1]));
    }

    #[tokio::test]
    async fn test_unsafe_recover_clear_pd_timeout_returns_service_unavailable() {
        let (sender, receiver) = tikv_util::mpsc::unbounded();
        std::thread::spawn(move || {
            while let Ok(msg) = receiver.recv() {
                if let CdcMsg::UnsafeRecoverClear { cb, .. } = msg {
                    cb(Err(Error::PdTimeout("timeout".to_string())));
                }
            }
        });
        let worker = ReplicationScheduler::new(
            sender,
            Arc::new(dashmap::DashMap::new()),
            Arc::new(HttpClient::Http(hyper::Client::new())),
            Duration::from_secs(1),
            ForceStop::default(),
        );
        let req = request::Builder::new()
            .method(Method::POST)
            .uri("/cdc/unsafe_recover/clear?region_id=1")
            .body(Body::empty())
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
        let body = read_json(resp).await;
        assert!(!body["success"].as_bool().unwrap());
        assert_eq!(body["reason"], "timeout");
    }

    #[tokio::test]
    async fn test_unsafe_recover_clear_rejects_missing_selector() {
        let worker = ReplicationScheduler::new_test();
        let req = request::Builder::new()
            .method(Method::POST)
            .uri("/cdc/unsafe_recover/clear")
            .body(Body::empty())
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let body = read_json(resp).await;
        assert!(!body["success"].as_bool().unwrap());
        assert!(!body["dry_run"].as_bool().unwrap());
    }

    #[tokio::test]
    async fn test_unsafe_recover_clear_rejects_both_selectors() {
        let worker = ReplicationScheduler::new_test();
        let req = request::Builder::new()
            .method(Method::POST)
            .uri("/cdc/unsafe_recover/clear?region_id=1&keyspace_id=42")
            .body(Body::empty())
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let body = read_json(resp).await;
        assert!(!body["success"].as_bool().unwrap());
        assert!(!body["dry_run"].as_bool().unwrap());
    }

    #[tokio::test]
    async fn test_unsafe_recover_clear_rejects_bad_selector() {
        let worker = ReplicationScheduler::new_test();
        let req = request::Builder::new()
            .method(Method::POST)
            .uri("/cdc/unsafe_recover/clear?region_id=abc")
            .body(Body::empty())
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let body = read_json(resp).await;
        assert!(!body["success"].as_bool().unwrap());
        assert!(!body["dry_run"].as_bool().unwrap());
    }

    #[tokio::test]
    async fn test_unsafe_recover_clear_rejects_bad_dry_run() {
        let worker = ReplicationScheduler::new_test();
        let req = request::Builder::new()
            .method(Method::POST)
            .uri("/cdc/unsafe_recover/clear?region_id=1&dry_run=maybe")
            .body(Body::empty())
            .unwrap();
        let resp = worker.handle_http_request(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let body = read_json(resp).await;
        assert!(!body["success"].as_bool().unwrap());
        assert!(!body["dry_run"].as_bool().unwrap());
    }

    #[test]
    fn test_extract_bool_param() {
        let cases = vec![
            ("/cdc/keyspace?keyspace_id=1", Ok(false)),
            ("/cdc/keyspace?keyspace_id=1&force=1", Ok(true)),
            ("/cdc/keyspace?keyspace_id=1&force=true", Ok(true)),
            ("/cdc/keyspace?keyspace_id=1&force=0", Ok(false)),
            ("/cdc/keyspace?keyspace_id=1&force=false", Ok(false)),
            (
                "/cdc/keyspace?keyspace_id=1&force=maybe",
                Err(StatusCode::BAD_REQUEST),
            ),
        ];

        for (uri, expected) in cases {
            let uri = Uri::from_static(uri);
            let query_pairs = parse_query(&uri);
            match expected {
                Ok(value) => {
                    assert_eq!(parse_bool_param(&query_pairs, "force").unwrap(), value);
                    assert_eq!(extract_bool_param(&query_pairs, "force").unwrap(), value);
                }
                Err(status) => {
                    parse_bool_param(&query_pairs, "force").unwrap_err();
                    let err_resp = extract_bool_param(&query_pairs, "force").unwrap_err();
                    assert_eq!(err_resp.status(), status);
                }
            }
        }
    }
}