use anyhow::Result;
use lazy_static::lazy_static;
use tici_common::index::IndexKey;
use tici_grpc::{
    ColumnInfo, FullTextIndexInfo,
    tidb_def::{
        BINARY_FLAG, COLTYPE_BLOB, COLTYPE_DATE, COLTYPE_DATETIME, COLTYPE_DOUBLE, COLTYPE_FLOAT,
        COLTYPE_LONG, COLTYPE_LONGLONG, COLTYPE_NEWDECIMAL, COLTYPE_TINY, COLTYPE_VARCHAR,
    },
};

use crate::{
    analysis::analyzer::DEFAULT_ANALYZER_NAME,
    utils::schema::{
        FulltextField, IndexMeta, IndexMetaStorage, IndexMetaVersion, InvertedColumn,
        ShardingColumn, SortColumn,
    },
};

pub const TEST_INDEX_KEY: IndexKey = IndexKey {
    keyspace_id: 1,
    // The table_id in test data ci/cdc_test_data.zip was 123
    table_id: 123,
    index_id: 3,
};

pub const TEST_FT_INDEX_INFO_STD: FullTextIndexInfo = FullTextIndexInfo {
    parser_type: tici_grpc::FullTextParserType::StandardV1,
};

pub const TEST_FT_INDEX_INFO_STD_STR: &str = r#"{ "parser_type": "STANDARD_V1" }"#;

lazy_static! {
    pub static ref TEST_INDEX_META_OVERLAP: IndexMeta = IndexMeta::try_from(IndexMetaStorage {
        version: IndexMetaVersion::LATEST,
        index_key: TEST_INDEX_KEY.clone(),
        created_ts: 0,
        columns: vec![
            ColumnInfo {
                column_id: 1,
                column_name: "name".to_string(),
                r#type: COLTYPE_VARCHAR,
                is_primary_key: true,
                collation: "utf8mb4_bin".to_string(),
                ..Default::default()
            },
            ColumnInfo {
                column_id: 2,
                column_name: "description".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 3,
                column_name: "pk1".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: true,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 4,
                column_name: "pk2".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: true,
                ..Default::default()
            },
        ],
        index_column_count: 2,
        sort_columns: vec![
            SortColumn {
                col_id: 1,
                is_asc: true,
            },
            SortColumn {
                col_id: 3,
                is_asc: true,
            },
            SortColumn {
                col_id: 4,
                is_asc: true,
            },
        ],
        inverted_columns: vec![],
        fulltext_fields: vec![
            FulltextField::new(vec![1], DEFAULT_ANALYZER_NAME.to_string()),
            FulltextField::new(vec![2], DEFAULT_ANALYZER_NAME.to_string()),
        ],
        sharding_columns: vec![],
        analyzer_defs: vec![],
        is_partition: false,
    })
    .unwrap();
    pub static ref TEST_INDEX_META_NORMAL: IndexMeta = IndexMeta::try_from(IndexMetaStorage {
        version: IndexMetaVersion::LATEST,
        index_key: TEST_INDEX_KEY.clone(),
        created_ts: 0,
        columns: vec![
            ColumnInfo {
                column_id: 1,
                column_name: "name".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 2,
                column_name: "description".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 3,
                column_name: "pk1".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: true,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 4,
                column_name: "pk2".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: true,
                ..Default::default()
            },
        ],
        index_column_count: 2,
        sort_columns: vec![
            SortColumn {
                col_id: 1,
                is_asc: true,
            },
            SortColumn {
                col_id: 3,
                is_asc: true,
            },
            SortColumn {
                col_id: 4,
                is_asc: true,
            },
        ],
        inverted_columns: vec![],
        fulltext_fields: vec![
            FulltextField::new(vec![1], DEFAULT_ANALYZER_NAME.to_string()),
            FulltextField::new(vec![2], DEFAULT_ANALYZER_NAME.to_string()),
        ],
        sharding_columns: vec![],
        analyzer_defs: vec![],
        is_partition: false,
    })
    .unwrap();
    pub static ref TEST_HDFSLOGS_INDEX_META: IndexMeta = IndexMeta::try_from(IndexMetaStorage {
        version: IndexMetaVersion::LATEST,
        index_key: TEST_INDEX_KEY.clone(),
        created_ts: 0,
        columns: vec![
            ColumnInfo {
                column_id: 1,
                column_name: "severity_text".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 2,
                column_name: "body".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 3,
                column_name: "id".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: true,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 4,
                column_name: "tenant_id".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: true,
                ..Default::default()
            },
        ],
        index_column_count: 4,
        sort_columns: vec![
            SortColumn {
                col_id: 1,
                is_asc: true,
            },
            SortColumn {
                col_id: 3,
                is_asc: true,
            },
            SortColumn {
                col_id: 4,
                is_asc: true,
            },
        ],
        inverted_columns: vec![InvertedColumn { col_id: 3 }, InvertedColumn { col_id: 4 }],
        fulltext_fields: vec![
            FulltextField::new(vec![1], DEFAULT_ANALYZER_NAME.to_string()),
            FulltextField::new(vec![2], DEFAULT_ANALYZER_NAME.to_string()),
        ],
        sharding_columns: vec![],
        analyzer_defs: vec![],
        is_partition: false,
    })
    .unwrap();
    pub static ref TEST_FLOAT_INDEX_META: IndexMeta = IndexMeta::try_from(IndexMetaStorage {
        version: IndexMetaVersion::LATEST,
        index_key: TEST_INDEX_KEY.clone(),
        created_ts: 0,
        columns: vec![
            ColumnInfo {
                column_id: 1,
                column_name: "id".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: true,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 2,
                column_name: "score1".to_string(),
                r#type: COLTYPE_FLOAT,
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 3,
                column_name: "score2".to_string(),
                r#type: COLTYPE_DOUBLE,
                is_primary_key: false,
                ..Default::default()
            },
        ],
        index_column_count: 3,
        sort_columns: vec![
            SortColumn {
                col_id: 1,
                is_asc: true,
            },
            SortColumn {
                col_id: 2,
                is_asc: true,
            },
            SortColumn {
                col_id: 3,
                is_asc: true,
            },
        ],
        inverted_columns: vec![
            InvertedColumn { col_id: 1 },
            InvertedColumn { col_id: 2 },
            InvertedColumn { col_id: 3 }
        ],
        fulltext_fields: vec![],
        sharding_columns: vec![],
        analyzer_defs: vec![],
        is_partition: false,
    })
    .unwrap();
    pub static ref TEST_DECIMAL_INDEX_META: IndexMeta = IndexMeta::try_from(IndexMetaStorage {
        version: IndexMetaVersion::LATEST,
        index_key: TEST_INDEX_KEY.clone(),
        created_ts: 0,
        columns: vec![
            ColumnInfo {
                column_id: 1,
                column_name: "id".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: true,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 2,
                column_name: "score1".to_string(),
                r#type: COLTYPE_NEWDECIMAL,
                is_primary_key: false,
                column_length: 18,
                decimal: 9,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 3,
                column_name: "score2".to_string(),
                r#type: COLTYPE_NEWDECIMAL,
                is_primary_key: false,
                column_length: 20,
                decimal: 6,
                ..Default::default()
            },
        ],
        index_column_count: 3,
        sort_columns: vec![
            SortColumn {
                col_id: 1,
                is_asc: true,
            },
            SortColumn {
                col_id: 2,
                is_asc: true,
            },
            SortColumn {
                col_id: 3,
                is_asc: true,
            },
        ],
        inverted_columns: vec![
            InvertedColumn { col_id: 1 },
            InvertedColumn { col_id: 2 },
            InvertedColumn { col_id: 3 }
        ],
        fulltext_fields: vec![],
        sharding_columns: vec![],
        analyzer_defs: vec![],
        is_partition: false,
    })
    .unwrap();
    static ref TEST_HDFSLOGS_INDEX_META_TEMPLATE: IndexMetaStorage = IndexMetaStorage {
        version: IndexMetaVersion::LATEST,
        index_key: TEST_INDEX_KEY.clone(),
        created_ts: 0,
        columns: vec![
            ColumnInfo {
                column_id: 1,
                column_name: "severity_text".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 2,
                column_name: "body".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 3,
                column_name: "id".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: true,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 4,
                column_name: "tenant_id".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: true,
                ..Default::default()
            },
        ],
        index_column_count: 4,
        sort_columns: vec![
            SortColumn {
                col_id: 1,
                is_asc: true,
            },
            SortColumn {
                col_id: 3,
                is_asc: true,
            },
            SortColumn {
                col_id: 4,
                is_asc: true,
            },
        ],
        inverted_columns: vec![InvertedColumn { col_id: 3 }, InvertedColumn { col_id: 4 }],
        fulltext_fields: vec![
            FulltextField::new(vec![1], DEFAULT_ANALYZER_NAME.to_string()),
            FulltextField::new(vec![2], DEFAULT_ANALYZER_NAME.to_string()),
        ],
        sharding_columns: vec![],
        analyzer_defs: vec![],
        is_partition: false,
    };
    static ref TEST_HDFSLOGS_INDEX_META_INT_HANDLE_TEMPLATE: IndexMetaStorage = {
        let mut index_meta = TEST_HDFSLOGS_INDEX_META_TEMPLATE.clone();
        // For the int-handle case, only the `id` column (column_id = 3) should remain
        // as the primary key. This ensures the table has a single int handle PK.
        index_meta.columns[3].is_primary_key = false;

        // Validate that there is exactly one primary key column and that it is `id`.
        let pk_columns: Vec<&ColumnInfo> = index_meta
            .columns
            .iter()
            .filter(|c| c.is_primary_key)
            .collect();
        assert_eq!(
            1,
            pk_columns.len(),
            "TEST_HDFSLOGS_INDEX_META_INT_HANDLE_TEMPLATE must have exactly one primary key column"
        );
        assert!(
            pk_columns[0].column_name == "id" && pk_columns[0].column_id == 3,
            "The single primary key column for int handle must be the `id` column (column_id = 3)"
        );
        index_meta
    };
    pub static ref TEST_MULTI_PK_META: IndexMeta = IndexMeta::try_from(IndexMetaStorage {
        version: IndexMetaVersion::LATEST,
        index_key: TEST_INDEX_KEY.clone(),
        created_ts: 0,
        columns: vec![
            ColumnInfo {
                column_id: 1,
                column_name: "id1".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: true,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 2,
                column_name: "id2".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: true,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 3,
                column_name: "id3".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: true,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 4,
                column_name: "id4".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: true,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 5,
                column_name: "id5".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: true,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 6,
                column_name: "v".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: false,
                column_length: 20,
                decimal: 6,
                ..Default::default()
            },
        ],
        index_column_count: 6,
        sort_columns: vec![],
        inverted_columns: vec![
            InvertedColumn { col_id: 1 },
            InvertedColumn { col_id: 2 },
            InvertedColumn { col_id: 3 },
            InvertedColumn { col_id: 4 },
            InvertedColumn { col_id: 5 },
            InvertedColumn { col_id: 6 },
        ],
        fulltext_fields: vec![],
        sharding_columns: vec![],
        analyzer_defs: vec![],
        is_partition: false,
    })
    .unwrap();

    pub static ref TEST_HDFSLOGS_INDEX_META_FULLTEXT: IndexMeta = IndexMeta::try_from(IndexMetaStorage {
        version: IndexMetaVersion::LATEST,
        index_key: TEST_INDEX_KEY.clone(),
        created_ts: 0,
        columns: vec![
            ColumnInfo {
                column_id: 1,
                column_name: "severity_text".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 2,
                column_name: "body".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 3,
                column_name: "id".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: true,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 4,
                column_name: "tenant_id".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: true,
                ..Default::default()
            },
        ],
        index_column_count: 2,
        sort_columns: vec![],
        inverted_columns: vec![],
        fulltext_fields: vec![
            FulltextField::new(vec![1], DEFAULT_ANALYZER_NAME.to_string()),
            FulltextField::new(vec![2], DEFAULT_ANALYZER_NAME.to_string()),
        ],
        sharding_columns: vec![],
        analyzer_defs: vec![],
        is_partition: false,
    })
    .unwrap();

    pub static ref TEST_ALL_TYPES_META: IndexMeta = IndexMeta::try_from(IndexMetaStorage {
        version: IndexMetaVersion::LATEST,
        index_key: TEST_INDEX_KEY.clone(),
        created_ts: 0,
        columns: vec![
            ColumnInfo {
                column_id: 1,
                column_name: "id".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: true,
                ..Default::default()
            },

            ColumnInfo {
                column_id: 2,
                column_name: "col_tinyint".to_string(),
                r#type: COLTYPE_TINY,
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 3,
                column_name: "col_bigint".to_string(),
                r#type: COLTYPE_LONGLONG,
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 4,
                column_name: "col_float".to_string(),
                r#type: COLTYPE_FLOAT,
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 5,
                column_name: "col_double".to_string(),
                r#type: COLTYPE_DOUBLE,
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 6,
                column_name: "col_date".to_string(),
                r#type: COLTYPE_DATE,
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 7,
                column_name: "col_datetime".to_string(),
                r#type: COLTYPE_DATETIME,
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 8,
                column_name: "col_varchar".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 9,
                column_name: "col_blob".to_string(),
                r#type: COLTYPE_BLOB,
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 10,
                column_name: "col_decimal".to_string(),
                r#type: COLTYPE_NEWDECIMAL,
                is_primary_key: false,
                column_length: 18,
                decimal: 6,
                ..Default::default()
            },
        ],
        index_column_count: 10,
        sort_columns: vec![],
        inverted_columns: vec![
            InvertedColumn { col_id: 2 },
            InvertedColumn { col_id: 3 },
            InvertedColumn { col_id: 4 },
            InvertedColumn { col_id: 5 },
            InvertedColumn { col_id: 6 },
            InvertedColumn { col_id: 7 },
            InvertedColumn { col_id: 8 },
            InvertedColumn { col_id: 9 },
            InvertedColumn { col_id: 10 },
        ],
        fulltext_fields: vec![],
        sharding_columns: vec![],
        analyzer_defs: vec![],
        is_partition: false,
    })
    .unwrap();

    pub static ref TEST_EMPLOYEES_PARTITION_INDEX_META: IndexMetaStorage = IndexMetaStorage {
        version: IndexMetaVersion::LATEST,
        index_key: TEST_INDEX_KEY.clone(),
        created_ts: 0,
        columns: vec![
            ColumnInfo {
                column_id: 1,
                column_name: "id".to_string(),
                r#type: COLTYPE_LONG,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: true,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 2,
                column_name: "fname".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 3,
                column_name: "lname".to_string(),
                r#type: COLTYPE_VARCHAR,
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 4,
                column_name: "store_id".to_string(),
                r#type: COLTYPE_LONG,
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 5,
                column_name: "department_id".to_string(),
                r#type: COLTYPE_LONG,
                is_primary_key: true,
                ..Default::default()
            },
        ],
        index_column_count: 5,
        sort_columns: vec![
            SortColumn {
                col_id: 1,
                is_asc: true,
            },
            SortColumn {
                col_id: 2,
                is_asc: false,
            },
            SortColumn {
                col_id: 4,
                is_asc: true,
            },
        ],
        inverted_columns: vec![
            InvertedColumn { col_id: 1 },
            InvertedColumn { col_id: 2 },
            InvertedColumn { col_id: 4 },
        ],
        fulltext_fields: vec![],
        sharding_columns: vec![
            ShardingColumn { col_id: 1 },
            ShardingColumn { col_id: 2 },
            ShardingColumn { col_id: 4 },
        ],
        analyzer_defs: vec![],
        is_partition: true,
    };

    pub static ref TEST_INDEX_META_BINARY_PK: IndexMeta = IndexMeta::try_from(IndexMetaStorage {
        version: IndexMetaVersion::LATEST,
        index_key: TEST_INDEX_KEY.clone(),
        created_ts: 0,
        columns: vec![
            ColumnInfo {
                column_id: 1,
                column_name: "description".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: false,
                ..Default::default()
            },
            ColumnInfo {
                column_id: 2,
                column_name: "binary_pk".to_string(),
                r#type: COLTYPE_VARCHAR,
                collation: "utf8mb4_bin".to_string(),
                is_primary_key: true,
                flag: BINARY_FLAG,
                ..Default::default()
            },
        ],
        index_column_count: 1,
        sort_columns: vec![],
        inverted_columns: vec![],
        fulltext_fields: vec![FulltextField::new(
            vec![1],
            DEFAULT_ANALYZER_NAME.to_string(),
        )],
        sharding_columns: vec![],
        analyzer_defs: vec![],
        is_partition: false,
    })
    .unwrap();
}

pub fn mock_custom_employees_partition_index_meta(index_key: IndexKey) -> Result<IndexMeta> {
    let mut index_meta = TEST_EMPLOYEES_PARTITION_INDEX_META.clone();
    index_meta.index_key = index_key;
    IndexMeta::try_from(index_meta)
}

pub fn mock_custom_hdfs_logs_index_meta(
    index_key: IndexKey,
    sharding_column_ids: &[i64],
) -> Result<IndexMeta> {
    let mut index_meta = TEST_HDFSLOGS_INDEX_META_TEMPLATE.clone();
    index_meta.index_key = index_key;
    let mut sharding_columns = vec![];
    for sharding_column_id in sharding_column_ids {
        sharding_columns.push(ShardingColumn {
            col_id: *sharding_column_id,
        });
    }
    index_meta.sharding_columns = sharding_columns;
    IndexMeta::try_from(index_meta)
}

pub fn mock_custom_hdfs_logs_index_meta_int_handle(
    index_key: IndexKey,
    sharding_column_ids: &[i64],
) -> Result<IndexMeta> {
    let mut index_meta = TEST_HDFSLOGS_INDEX_META_INT_HANDLE_TEMPLATE.clone();
    index_meta.index_key = index_key;
    let mut sharding_columns = vec![];
    for sharding_column_id in sharding_column_ids {
        sharding_columns.push(ShardingColumn {
            col_id: *sharding_column_id,
        });
    }
    index_meta.sharding_columns = sharding_columns;
    IndexMeta::try_from(index_meta)
}