use anyhow::Context as _;
use log::info;
use mysql_async::{Conn, TxOpts, prelude::*};
use serde::{Deserialize, Serialize};
use tici_common::index::IndexKey;

use crate::meta::store::metastore::META_DATABASE_NAME;

pub const META_TABLE_NAMES: [&str; 1] = ["tici_index_meta"];

#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct IndexMetaProperties {
    #[serde(default)]
    pub finished_append_base: bool,
}

fn serialize_index_meta_properties(props: &IndexMetaProperties) -> Result<String, anyhow::Error> {
    Ok(serde_json::to_string(props)?)
}

fn parse_index_meta_properties(
    keyspace_id: u32,
    table_id: i64,
    index_id: i64,
    properties_str: Option<&str>,
) -> Result<IndexMetaProperties, anyhow::Error> {
    match properties_str {
        Some(s) if !s.trim().is_empty() => serde_json::from_str(s).with_context(|| {
            format!(
                "restore_index_meta: deserialize properties, keyspace_id={keyspace_id}, table_id={table_id}, index_id={index_id}"
            )
        }),
        _ => Ok(IndexMetaProperties::default()),
    }
}

pub async fn ensure_schema(conn: &mut Conn) -> Result<(), anyhow::Error> {
    info!("Ensure ShardStore table schema");
    let _s: Vec<String> = conn
        .query(format!(
            r"
            CREATE DATABASE IF NOT EXISTS {0};
            CREATE TABLE IF NOT EXISTS {0}.tici_index_meta (
                keyspace_id INT UNSIGNED NOT NULL,
                table_id BIGINT UNSIGNED NOT NULL,
                index_id BIGINT UNSIGNED NOT NULL,
                index_info LONGTEXT NOT NULL,
                table_info LONGTEXT NOT NULL,
                cdc_s3_bucket VARCHAR(16383),
                cdc_s3_prefix VARCHAR(16383),
                properties TEXT,
                PRIMARY KEY (keyspace_id, table_id, index_id) CLUSTERED
            );",
            META_DATABASE_NAME
        ))
        .await?;
    info!("Ensure IndexStore table schema finished");
    Ok(())
}

pub async fn init(
    conn: &mut Conn,
    table_info: &tici_grpc::TableInfo,
    index_info: &tici_grpc::IndexInfo,
    cdc_s3_bucket: String,
    cdc_s3_prefix: String,
) -> Result<(), anyhow::Error> {
    let mut tx = conn.start_transaction(TxOpts::default()).await?;
    let table_str = serde_json::to_string(&table_info)?;
    let index_str = serde_json::to_string(&index_info)?;
    let properties = serialize_index_meta_properties(&IndexMetaProperties::default())?;
    let q = tx
            .exec_iter(
                format!(r"INSERT IGNORE INTO {}.tici_index_meta (keyspace_id,table_id,index_id,index_info,table_info,cdc_s3_bucket,cdc_s3_prefix,properties)
                VALUES (:keyspace_id,:table_id,:index_id,:index_info,:table_info,:cdc_s3_bucket,:cdc_s3_prefix,:properties);", META_DATABASE_NAME),
                params! {
                    "keyspace_id" => index_info.keyspace_id,
                    "table_id" => index_info.table_id,
                    "index_id" => index_info.index_id,
                    "table_info" => table_str,
                    "index_info" => index_str,
                    "cdc_s3_bucket" => cdc_s3_bucket,
                    "cdc_s3_prefix" => cdc_s3_prefix,
                    "properties" => properties,
                },
            )
            .await?;

    if q.affected_rows() != 1 {
        let affect_rows = q.affected_rows();
        tx.rollback().await?;
        return Err(anyhow::format_err!(
            "init_index: detected inconsistency, keyspace_id={}, table_id={}, index_id={}, affect_rows={}",
            index_info.keyspace_id,
            index_info.table_id,
            index_info.index_id,
            affect_rows,
        ));
    }
    tx.commit().await?;
    Ok(())
}

pub async fn drop(conn: &mut Conn, index_key: &IndexKey) -> Result<(), anyhow::Error> {
    let mut tx = conn.start_transaction(TxOpts::default()).await?;
    tx.exec_iter(
        format!(
            r"DELETE FROM {}.tici_index_meta WHERE keyspace_id=:keyspace_id AND table_id=:table_id AND index_id=:index_id;",
            META_DATABASE_NAME
        ),
        params! {
            "keyspace_id" => index_key.keyspace_id,
            "table_id" => index_key.table_id,
            "index_id" => index_key.index_id,
        },
    )
    .await?;
    // We neglect if there isn't any.
    tx.commit().await?;
    Ok(())
}

pub async fn finish_append_base(
    conn: &mut Conn,
    index_key: &IndexKey,
) -> Result<(), anyhow::Error> {
    let mut tx = conn.start_transaction(TxOpts::default()).await?;
    let properties_str: Option<String> = tx
        .exec_first(
            format!(
                r"SELECT properties FROM {}.tici_index_meta WHERE keyspace_id=:keyspace_id AND table_id=:table_id AND index_id=:index_id FOR UPDATE;",
                META_DATABASE_NAME
            ),
            params! {
                "keyspace_id" => index_key.keyspace_id,
                "table_id" => index_key.table_id,
                "index_id" => index_key.index_id,
            },
        )
        .await?
        .ok_or_else(|| {
            anyhow::format_err!(
                "finish_append_base: index not found, keyspace_id={}, table_id={}, index_id={}",
                index_key.keyspace_id,
                index_key.table_id,
                index_key.index_id,
            )
        })?;
    let mut properties = parse_index_meta_properties(
        index_key.keyspace_id,
        index_key.table_id,
        index_key.index_id,
        properties_str.as_deref(),
    )?;
    if properties.finished_append_base {
        tx.rollback().await?;
        return Ok(());
    }
    properties.finished_append_base = true;
    let properties = serialize_index_meta_properties(&properties)?;
    let q = tx
        .exec_iter(
            format!(
                r"UPDATE {}.tici_index_meta SET properties=:properties WHERE keyspace_id=:keyspace_id AND table_id=:table_id AND index_id=:index_id;",
                META_DATABASE_NAME
            ),
            params! {
                "keyspace_id" => index_key.keyspace_id,
                "table_id" => index_key.table_id,
                "index_id" => index_key.index_id,
                "properties" => properties,
            },
        )
        .await?;
    if q.affected_rows() != 1 {
        let affect_rows = q.affected_rows();
        tx.rollback().await?;
        return Err(anyhow::format_err!(
            "finish_append_base: update inconsistency, keyspace_id={}, table_id={}, index_id={}, affect_rows={}",
            index_key.keyspace_id,
            index_key.table_id,
            index_key.index_id,
            affect_rows,
        ));
    }
    tx.commit().await?;
    Ok(())
}

fn parse_index_meta_row(
    keyspace_id: u32,
    table_id: i64,
    index_id: i64,
    index_str: &str,
    table_str: &str,
    cdc_s3_bucket: String,
    cdc_s3_prefix: String,
    properties_str: Option<&str>,
) -> Result<
    (
        tici_grpc::IndexInfo,
        tici_grpc::TableInfo,
        String,
        String,
        IndexMetaProperties,
    ),
    anyhow::Error,
> {
    let mut index: tici_grpc::IndexInfo = serde_json::from_str(index_str).with_context(|| {
        format!(
            "restore_index_meta: deserialize index_info, keyspace_id={keyspace_id}, table_id={table_id}, index_id={index_id}"
        )
    })?;
    let mut table: tici_grpc::TableInfo = serde_json::from_str(table_str).with_context(|| {
        format!(
            "restore_index_meta: deserialize table_info, keyspace_id={keyspace_id}, table_id={table_id}, index_id={index_id}"
        )
    })?;
    index.keyspace_id = keyspace_id;
    table.keyspace_id = keyspace_id;
    let properties = parse_index_meta_properties(keyspace_id, table_id, index_id, properties_str)?;
    Ok((index, table, cdc_s3_bucket, cdc_s3_prefix, properties))
}

pub async fn restore(
    conn: &mut Conn,
) -> Result<
    Vec<(
        tici_grpc::IndexInfo,
        tici_grpc::TableInfo,
        String,
        String,
        IndexMetaProperties,
    )>,
    anyhow::Error,
> {
    let indexes = conn
        .query_map(
            format!(
                r"SELECT keyspace_id,table_id,index_id,index_info,table_info,cdc_s3_bucket,cdc_s3_prefix,properties FROM {}.tici_index_meta",
                META_DATABASE_NAME
            ),
            |(
                keyspace_id,
                table_id,
                index_id,
                index_str,
                table_str,
                cdc_s3_bucket,
                cdc_s3_prefix,
                properties,
            ): (u32, i64, i64, String, String, String, String, Option<String>)| {
                parse_index_meta_row(
                    keyspace_id,
                    table_id,
                    index_id,
                    index_str.as_str(),
                    table_str.as_str(),
                    cdc_s3_bucket,
                    cdc_s3_prefix,
                    properties.as_deref(),
                )
            },
        )
        .await?;
    let indexes = indexes
        .into_iter()
        .collect::<Result<Vec<_>, anyhow::Error>>()?;

    info!("Restored {} indexes from meta table", indexes.len());
    Ok(indexes)
}

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

    use super::{IndexMetaProperties, parse_index_meta_properties, parse_index_meta_row};

    #[test]
    fn parse_index_meta_row_overrides_keyspace_id() -> Result<(), anyhow::Error> {
        let index = tici_grpc::IndexInfo {
            keyspace_id: 0,
            table_id: 1,
            index_id: 2,
            index_name: "idx".to_string(),
            index_type: 0,
            columns: Vec::new(),
            is_unique: false,
            parser_info: Some(tici_grpc::DEFAULT_PARSER_INFO.clone()),
            other_params: HashMap::new(),
            fulltext_index: String::new(),
            hybrid_index: String::new(),
            created_ts: 0,
        };
        let table = tici_grpc::TableInfo {
            keyspace_id: 0,
            table_id: 1,
            table_name: "t".to_string(),
            database_name: "db".to_string(),
            version: 1,
            columns: Vec::new(),
            is_clustered: false,
            is_partition: false,
            logical_table_id: 0,
        };
        let index_str = serde_json::to_string(&index)?;
        let table_str = serde_json::to_string(&table)?;

        let (restored_index, restored_table, bucket, prefix, properties) = parse_index_meta_row(
            42,
            1,
            2,
            index_str.as_str(),
            table_str.as_str(),
            "b".to_string(),
            "p".to_string(),
            Some(r#"{"finished_append_base":true}"#),
        )?;

        assert_eq!(restored_index.keyspace_id, 42);
        assert_eq!(restored_table.keyspace_id, 42);
        assert_eq!(bucket, "b");
        assert_eq!(prefix, "p");
        assert!(properties.finished_append_base);
        Ok(())
    }

    #[test]
    fn parse_index_meta_row_invalid_index_json_returns_err() -> Result<(), anyhow::Error> {
        let table = tici_grpc::TableInfo {
            keyspace_id: 0,
            table_id: 1,
            table_name: "t".to_string(),
            database_name: "db".to_string(),
            version: 1,
            columns: Vec::new(),
            is_clustered: false,
            is_partition: false,
            logical_table_id: 0,
        };
        let table_str = serde_json::to_string(&table)?;

        let res = parse_index_meta_row(
            42,
            1,
            2,
            "not_json",
            table_str.as_str(),
            "b".to_string(),
            "p".to_string(),
            None,
        );
        assert!(res.is_err());
        Ok(())
    }

    #[test]
    fn parse_index_meta_properties_defaults_when_missing() -> Result<(), anyhow::Error> {
        let props = parse_index_meta_properties(42, 1, 2, None)?;
        assert_eq!(props, IndexMetaProperties::default());
        Ok(())
    }

    #[test]
    fn parse_index_meta_properties_invalid_json_returns_err() {
        let res = parse_index_meta_properties(42, 1, 2, Some("not_json"));
        assert!(res.is_err());
    }
}