use std::{
    env,
    path::{Path, PathBuf},
};

use anyhow::{Context, Result, bail, ensure};
use mysql_async::{Pool, TxOpts, prelude::Queryable};
use serde_json;
use tici_common::index::IndexKey;
use tici_grpc::{
    CIStr,
    tidb_def::{self, COLTYPE_LONG, COLTYPE_TIMESTAMP, COLTYPE_VARCHAR},
};
use tici_shard::test_utils::index_meta::TEST_FT_INDEX_INFO_STD;
use tokio::process::Command;
use url::Url;

const DEFAULT_TIDB_PORT: u16 = 4000;
const HDFS_DATASET_FILE: &str = "hdfs-logs-multitenants.json";

struct TiDBEndpoint {
    host: String,
    port: u16,
}

/// Drop (if exists) and create an empty `hdfs_10w`-style table via
/// `ci/read_hdfs.py` (`max_rows=0`, without `--insert_only`).
pub async fn bootstrap_empty_hdfs_table(tidb_addr: &str, table_name: &str) -> Result<()> {
    validate_table_name(table_name)?;
    let tidb_endpoint = parse_tidb_endpoint(tidb_addr)?;
    let project_root = project_root()?;
    let script = project_root.join("ci/read_hdfs.py");
    ensure!(
        script.is_file(),
        "read_hdfs.py not found: {}",
        script.display()
    );

    let asset_dir = env::var_os("ASSET_DIR")
        .map(PathBuf::from)
        .unwrap_or_else(|| project_root.clone());
    let input = asset_dir.join(HDFS_DATASET_FILE);
    ensure!(
        input.is_file(),
        "HDFS log dataset not found: {}",
        input.display()
    );

    let output = Command::new("python3")
        .arg(&script)
        .arg(table_name)
        .arg("--input")
        .arg(&input)
        .arg("--max_rows")
        .arg("0")
        .arg("--tidb_host")
        .arg(tidb_endpoint.host)
        .arg("--tidb_port")
        .arg(tidb_endpoint.port.to_string())
        .output()
        .await
        .with_context(|| format!("failed to run {}", script.display()))?;

    if !output.status.success() {
        bail!(
            "failed to bootstrap empty HDFS table {table_name}, status={}, stdout={}, stderr={}",
            output.status,
            String::from_utf8_lossy(&output.stdout),
            String::from_utf8_lossy(&output.stderr)
        );
    }
    Ok(())
}

pub async fn get_table_id(tidb_addr: &str) -> Result<i64> {
    let pool = Pool::new(tidb_addr);
    let mut conn = pool.get_conn().await?;
    let mut tx = conn.start_transaction(TxOpts::default()).await?;
    let q = tx
        .query(r"SELECT tidb_table_id FROM information_schema.tables WHERE table_name='hdfs_10w';")
        .await?;
    if q.is_empty() {
        return Err(anyhow::anyhow!("Table hdfs_10w not found in tidb"));
    }
    Ok(q[0])
}

pub async fn insert_hdfs_logs(tidb_addr: &str, table_name: &str, rows: u64) -> Result<()> {
    validate_table_name(table_name)?;
    let tidb_endpoint = parse_tidb_endpoint(tidb_addr)?;
    let project_root = project_root()?;
    let script = project_root.join("ci/read_hdfs.py");
    ensure!(
        script.is_file(),
        "read_hdfs.py not found: {}",
        script.display()
    );

    let asset_dir = env::var_os("ASSET_DIR")
        .map(PathBuf::from)
        .unwrap_or_else(|| project_root.clone());
    let input = asset_dir.join(HDFS_DATASET_FILE);
    ensure!(
        input.is_file(),
        "HDFS log dataset not found: {}",
        input.display()
    );

    let output = Command::new("python3")
        .arg(&script)
        .arg(table_name)
        .arg("--input")
        .arg(&input)
        .arg("--max_rows")
        .arg(rows.to_string())
        .arg("--insert_only")
        .arg("--tidb_host")
        .arg(tidb_endpoint.host)
        .arg("--tidb_port")
        .arg(tidb_endpoint.port.to_string())
        .output()
        .await
        .with_context(|| format!("failed to run {}", script.display()))?;

    if !output.status.success() {
        bail!(
            "failed to insert HDFS logs into {table_name}, status={}, stdout={}, stderr={}",
            output.status,
            String::from_utf8_lossy(&output.stdout),
            String::from_utf8_lossy(&output.stderr)
        );
    }
    Ok(())
}

fn parse_tidb_endpoint(tidb_addr: &str) -> Result<TiDBEndpoint> {
    let url =
        Url::parse(tidb_addr).with_context(|| format!("failed to parse TiDB DSN: {tidb_addr}"))?;
    let host = url
        .host_str()
        .with_context(|| format!("missing host in TiDB DSN: {tidb_addr}"))?
        .to_string();
    Ok(TiDBEndpoint {
        host,
        port: url.port().unwrap_or(DEFAULT_TIDB_PORT),
    })
}

fn project_root() -> Result<PathBuf> {
    if let Some(project_root) = env::var_os("PROJECT_ROOT") {
        return Ok(PathBuf::from(project_root));
    }

    Path::new(env!("CARGO_MANIFEST_DIR"))
        .join("../..")
        .canonicalize()
        .context("failed to resolve project root")
}

fn validate_table_name(table_name: &str) -> Result<()> {
    let mut chars = table_name.chars();
    let Some(first) = chars.next() else {
        bail!("table name cannot be empty");
    };
    ensure!(
        first == '_' || first.is_ascii_alphabetic(),
        "invalid table name: {table_name}"
    );
    ensure!(
        chars.all(|c| c == '_' || c.is_ascii_alphanumeric()),
        "invalid table name: {table_name}"
    );
    Ok(())
}

fn get_tidb_column_info(
    id: i64,
    name: &str,
    tp: i32,
    is_pk: bool,
    offset: i32,
) -> tici_grpc::TiDBColumnInfo {
    tici_grpc::TiDBColumnInfo {
        id,
        name: CIStr {
            original: name.to_string(),
            lower_case: name.to_string(),
        },
        offset,
        field_type: tici_grpc::TiDBFieldType {
            tp: tp as u8,
            flag: if is_pk { tidb_def::PRI_KEY_FLAG } else { 0 },
            collate: "utf8mb4_bin".to_string(),
            ..Default::default()
        },
    }
}

fn get_tidb_index_column(
    column_infos: &[tici_grpc::TiDBColumnInfo],
    col_id: i64,
) -> tici_grpc::TiDBIndexColumn {
    let col_index = column_infos.iter().position(|c| c.id == col_id).unwrap();
    let col = column_infos.get(col_index).unwrap();
    tici_grpc::TiDBIndexColumn {
        name: col.name.clone(),
        offset: col.offset,
        length: 0,
    }
}

fn get_tidb_index_info(
    table: CIStr,
    column_infos: &[tici_grpc::TiDBColumnInfo],
    index_id: i64,
) -> tici_grpc::TiDBIndexInfo {
    tici_grpc::TiDBIndexInfo {
        id: index_id,
        name: CIStr {
            original: "fulltext_severity_body_idx".to_string(),
            lower_case: "fulltext_severity_body_idx".to_string(),
        },
        table: Some(table),
        columns: vec![
            get_tidb_index_column(column_infos, 3),
            get_tidb_index_column(column_infos, 4),
        ],
        tp: tici_grpc::TiDBIndexType::Fulltext,
        unique: false,
        primary: false,
        invisible: false,
        global: false,
        mv_index: false,
        full_text_info: Some(TEST_FT_INDEX_INFO_STD.clone()),
        hybrid_index: None,
    }
}

fn get_tidb_table_info(table_id: i64, index_id: i64) -> tici_grpc::TiDBTableInfo {
    let columns = vec![
        get_tidb_column_info(1, "id", COLTYPE_LONG, true, 0),
        get_tidb_column_info(2, "timestamp", COLTYPE_TIMESTAMP, false, 1),
        get_tidb_column_info(3, "severity_text", COLTYPE_VARCHAR, false, 2),
        get_tidb_column_info(4, "body", COLTYPE_VARCHAR, false, 3),
        get_tidb_column_info(5, "tenant_id", COLTYPE_LONG, true, 4),
    ];
    let name = CIStr {
        original: "hdfs_10w".to_string(),
        lower_case: "hdfs_10w".to_string(),
    };
    let indexes = vec![get_tidb_index_info(name.clone(), &columns, index_id)];
    tici_grpc::TiDBTableInfo {
        id: table_id,
        name,
        columns,
        // charset: String,
        // collate: String,
        indexes,
        // pk_is_handle: bool,
        is_common_handle: true,
        // update_ts: u64,
        // version: u16,
        ..Default::default()
    }
}

fn get_database_name() -> String {
    "test".to_string()
}

pub fn make_create_index_req(index_key: &IndexKey) -> tici_grpc::CreateIndexRequest {
    let table_info = get_tidb_table_info(index_key.table_id, index_key.index_id);
    let json_string = serde_json::to_string(&table_info).unwrap();
    tici_grpc::CreateIndexRequest {
        database_name: get_database_name(),
        table_info: json_string.into_bytes(),
        index_id: index_key.index_id,
        keyspace_id: index_key.keyspace_id,
        parser_info: Some(tici_grpc::DEFAULT_PARSER_INFO.clone()),
    }
}