// Copyright 2025 PingCAP, Inc.

use std::time::Duration;

use http::Uri;
use security::HttpClient;
use serde_derive::Deserialize;
use thiserror::Error;
use tikv_util::{box_err, box_try, info};

use crate::{Error, Result, util::read_from_ticdc};

#[derive(Error, Debug, Clone)]
pub enum TiCdcError {
    #[error("Server is not ready: {0}")]
    ServerIsNotReady(String),
    #[error("Changefeed already exists: {0}")]
    ChangeFeedAlreadyExists(String),
    #[error("Invalid format: {0}")]
    InvalidFormat(String),
    #[error("{error_msg}")]
    OtherError {
        error_code: String,
        error_msg: String,
    },
}

#[derive(Deserialize, Clone, Debug, Default)]
pub(crate) struct ErrorResponse {
    pub error_msg: String,
    pub error_code: String,
}

pub(crate) fn parse_ticdc_response(resp: &[u8]) -> TiCdcError {
    let Ok(js_resp) = serde_json::from_slice::<ErrorResponse>(resp) else {
        return TiCdcError::InvalidFormat(String::from_utf8_lossy(resp).to_string());
    };

    match js_resp.error_code.as_str() {
        "CDC:ErrChangeFeedAlreadyExists" => TiCdcError::ChangeFeedAlreadyExists(js_resp.error_msg),
        "CDC:ErrServerIsNotReady" => TiCdcError::ServerIsNotReady(js_resp.error_msg),
        _ => TiCdcError::OtherError {
            error_code: js_resp.error_code,
            error_msg: js_resp.error_msg,
        },
    }
}

// Ref: https://docs.pingcap.com/tidb/stable/ticdc-open-api-v2/#query-the-replication-task-list
#[derive(Deserialize, Clone, Debug, Default)]
pub(crate) struct ReplicationTaskList {
    pub total: usize,
    pub items: Vec<ReplicationTaskItem>,
}

#[allow(unused)]
#[derive(Deserialize, Clone, Debug, Default)]
pub(crate) struct RunningError {
    pub time: String,
    pub addr: String,
    pub code: String,
    pub message: String,
}

#[allow(unused)]
#[derive(Deserialize, Clone, Debug, Default)]
pub(crate) struct ReplicationTaskItem {
    pub id: String,
    pub state: String, // normal, stopped, error, failed, finished
    pub checkpoint_tso: u64,
    pub checkpoint_time: String,
    pub error: Option<RunningError>,
}

pub(crate) fn format_changefeed_summary(changefeeds: &[ReplicationTaskItem]) -> String {
    const MAX_FEEDS: usize = 5;

    let mut summary = changefeeds
        .iter()
        .take(MAX_FEEDS)
        .map(|feed| format!("{}({})", feed.id, feed.state))
        .collect::<Vec<_>>()
        .join(", ");
    if changefeeds.len() > MAX_FEEDS {
        summary.push_str(&format!(", ... +{} more", changefeeds.len() - MAX_FEEDS));
    }
    summary
}

pub(crate) async fn list_changefeeds_from_ticdc(
    keyspace_id: u32,
    cdc_addr: &str,
    http_client: &HttpClient,
    timeout: Duration,
    tag: &str,
) -> Result<ReplicationTaskList> {
    let scheme = if matches!(http_client, HttpClient::Http(_)) {
        "http"
    } else {
        "https"
    };
    let uri: Uri = box_try!(format!("{scheme}://{cdc_addr}/api/v2/changefeeds").parse());
    let resp = read_from_ticdc(
        tag,
        http_client,
        uri,
        timeout,
        Duration::from_millis(200),
        |_| false,
    )
    .await?;
    let changefeed_list =
        serde_json::from_slice::<ReplicationTaskList>(&resp).map_err(|e| -> Error {
            let err_msg = format!("failed to parse changefeeds response: {:?}", e);
            debug_assert!(false, "{}", err_msg);
            box_err!("{}", err_msg)
        })?;
    info!("TiCDC changefeed list: {:?}", changefeed_list; "keyspace" => keyspace_id);
    debug_assert_eq!(changefeed_list.items.len(), changefeed_list.total);
    Ok(changefeed_list)
}