use std::sync::Arc;

use crate::error::{column_not_found, no_such_table, Error, ErrorKind, Result};
use lazy_static::lazy_static;
use log::*;
use regex::Regex;
use serde::{Deserialize, Serialize};
use sqlx::types::chrono::NaiveDateTime;
use sqlx::{FromRow, MySql, MySqlConnection, MySqlPool};

#[derive(Clone, Debug, Deserialize, Serialize)]
pub enum IndexType {
    TinyInt,
    BigInt,
    Varchar,
    Datetime,
    Double,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct PayloadIndex {
    pub path: String,
    pub length: Option<u32>,
    pub index_type: IndexType,
}

impl PayloadIndex {
    pub fn create(&self, collection: &str) -> Result<(String, String)> {
        let index_type = match self.index_type {
            IndexType::TinyInt => "TINYINT",
            IndexType::BigInt => "BIGINT",
            IndexType::Varchar => &format!(
                "VARCHAR({})",
                self.length.expect("VARCHAR must have length")
            ),
            IndexType::Datetime => "DATETIME",
            IndexType::Double => "DOUBLE",
        };

        let (column_name, idx_name) = path_to_column_and_idx_name(&self.path)?;
        let add_column = format!(
            r#"
            ALTER TABLE `{}` ADD COLUMN `{}` {} AS (JSON_UNQUOTE(JSON_EXTRACT(payload, '$.{}')));
            "#,
            collection, column_name, index_type, &self.path
        );

        let add_idx = format!(
            r#"
            CREATE INDEX `{}` ON `{}`(`{}`);
            "#,
            idx_name, collection, column_name
        );

        Ok((add_column, add_idx))
    }

    pub fn drop(path: &str, collection: &str) -> Result<String> {
        let (column_name, _) = path_to_column_and_idx_name(path)?;
        let drop_column = format!(
            r#"
            ALTER TABLE `{}` DROP COLUMN `{}`;
            "#,
            collection, column_name
        );

        Ok(drop_column)
    }
}

lazy_static! {
    static ref PAYLOAD_PATH_PATTERN: Regex = Regex::new(r#"^[a-z0-9A-Z$_.]+$"#).unwrap();
}

fn is_valid_payload_path(key: &str) -> bool {
    PAYLOAD_PATH_PATTERN.is_match(key)
}

pub fn path_to_column_and_idx_name(path: &str) -> Result<(String, String)> {
    if !is_valid_payload_path(path) {
        return Err(Error::new(
            ErrorKind::BadRequest,
            format!("Invalid payload path: '{}'", path),
        ));
    }

    Ok((
        format!("__payload__{}", path),
        format!("__idx_payload__{}", path),
    ))
}

#[derive(FromRow)]
pub struct TenantCollectionMeta {
    pub collection_name: String,
    pub collection_id: u64,
    pub payload_schema: Option<String>,
    pub created_at: NaiveDateTime,
    pub updated_at: NaiveDateTime,
    pub deleted_at: Option<NaiveDateTime>,
}

pub(crate) const TENANT_COLLECTION_META_TABLE_NAME: &str = "_ai_collections";
impl TenantCollectionMeta {
    pub async fn create_table(db: &MySqlPool) -> Result<()> {
        let query = format!(
            r#"
        CREATE TABLE IF NOT EXISTS `{}`(
          `collection_name` VARCHAR(64) NOT NULL,
          `collection_id` BIGINT UNSIGNED NOT NULL,
          `payload_schema` JSON,
          `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
          `updated_at` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
          `deleted_at` DATETIME DEFAULT NULL,
          PRIMARY KEY `idx_collection_name`(`collection_name`)
        );
        "#,
            TENANT_COLLECTION_META_TABLE_NAME
        );
        sqlx::query(&query).execute(db).await?;
        Ok(())
    }

    pub async fn add_payload_idx(
        ctx: Arc<context::Context>,
        db: &MySqlPool,
        collection_name: &str,
        payload_index: PayloadIndex,
    ) -> Result<()> {
        let (add_column, add_idx) = payload_index
            .create(collection_name)
            .map_err(|e| e.ctx(ctx.clone()))?;
        let mut tx = db
            .begin()
            .await
            .map_err(|e| Error::from(e).ctx(ctx.clone()))?;
        sqlx::query(&add_column)
            .execute(&mut *tx)
            .await
            .map_err(|e| Error::from(e).ctx(ctx.clone()))?;
        sqlx::query(&add_idx)
            .execute(&mut *tx)
            .await
            .map_err(|e| Error::from(e).ctx(ctx.clone()))?;
        tx.commit()
            .await
            .map_err(|e| Error::from(e).ctx(ctx.clone()))?;
        Ok(())
    }

    pub async fn update_payload_index(
        ctx: Arc<context::Context>,
        db: &MySqlPool,
        collection_name: &str,
        payload_index: String,
    ) -> Result<()> {
        let query = format!(
            r#"
            UPDATE `{}` SET `payload_schema` = ? WHERE `collection_name` = ?;
            "#,
            TENANT_COLLECTION_META_TABLE_NAME
        );
        sqlx::query(&query)
            .bind(payload_index)
            .bind(collection_name)
            .execute(db)
            .await
            .map_err(|e| Error::from(e).ctx(ctx))?;
        Ok(())
    }

    pub async fn drop_payload_idx(
        ctx: Arc<context::Context>,
        db: &MySqlPool,
        collection_name: &str,
        path: &str,
    ) -> Result<()> {
        let drop_column =
            PayloadIndex::drop(path, collection_name).map_err(|e| e.ctx(ctx.clone()))?;
        sqlx::query(&drop_column)
            .execute(db)
            .await
            .map_err(|e| Error::from(e).ctx(ctx))?;
        Ok(())
    }

    pub async fn get_by_collection_name(
        ctx: Arc<context::Context>,
        db: &MySqlPool,
        collection_name: impl AsRef<str>,
        ignore_deleted: bool,
    ) -> Result<Option<TenantCollectionMeta>> {
        let collection_name = collection_name.as_ref();
        let query = if ignore_deleted {
            format!(
                r#"
                SELECT 
                    `collection_name`, 
                    `collection_id`, 
                    CAST(`payload_schema` AS CHAR) AS `payload_schema`,
                    `created_at`, `updated_at`, `deleted_at`
                FROM `{}` WHERE `collection_name` = ?;
                "#,
                TENANT_COLLECTION_META_TABLE_NAME
            )
        } else {
            format!(
                r#"
                SELECT 
                    `collection_name`, 
                    `collection_id`, 
                    CAST(`payload_schema` AS CHAR) AS `payload_schema`,
                    `created_at`, `updated_at`, `deleted_at`
                FROM `{}` WHERE `collection_name` = ? AND `deleted_at` IS NULL;
                "#,
                TENANT_COLLECTION_META_TABLE_NAME
            )
        };

        let mut retry = true;
        let record;
        loop {
            record = match sqlx::query_as::<MySql, TenantCollectionMeta>(&query)
                .bind(collection_name)
                .fetch_optional(db)
                .await
            {
                Err(e) => {
                    if retry && column_not_found(&e) {
                        warn!(
                            "[{}] Try add field `payload_schema` to table '{}'. {}",
                            ctx.id(),
                            TENANT_COLLECTION_META_TABLE_NAME,
                            e
                        );
                        let mut tx = db.begin().await?;
                        Self::add_field_payload_schema(&mut tx).await;
                        tx.commit().await?;
                        retry = false;
                        continue;
                    }
                    return Err(Error::from(e).ctx(ctx));
                }
                Ok(record) => {
                    if record.is_none() {
                        warn!(
                            "[{}] Collection {} not found in tenant-db",
                            ctx.id(),
                            collection_name
                        );
                    }
                    record
                }
            };
            break;
        }
        Ok(record)
    }

    pub async fn add_field_payload_schema(db: &mut MySqlConnection) {
        let query = format!(
            r#"
        ALTER TABLE `{}` ADD COLUMN `payload_schema` JSON;
        "#,
            TENANT_COLLECTION_META_TABLE_NAME
        );
        if let Err(e) = sqlx::query(&query).execute(db).await {
            error!(
                "Failed to add field `payload_schema` to table '{}'. {}",
                TENANT_COLLECTION_META_TABLE_NAME, e
            );
        }
    }

    pub async fn rename(
        db: &MySqlPool,
        from_collection_name: impl AsRef<str>,
        to_collection_name: impl AsRef<str>,
    ) -> Result<()> {
        let mut tx = db.begin().await?;
        sqlx::query(&format!(
            r#"DELETE FROM `{}` WHERE `collection_name` = ?"#,
            TENANT_COLLECTION_META_TABLE_NAME
        ))
        .bind(to_collection_name.as_ref())
        .execute(&mut *tx)
        .await?;
        sqlx::query(&format!(
            r#"UPDATE `{}` SET `collection_name` = ? WHERE `collection_name` = ?"#,
            TENANT_COLLECTION_META_TABLE_NAME
        ))
        .bind(to_collection_name.as_ref())
        .bind(from_collection_name.as_ref())
        .execute(&mut *tx)
        .await?;

        sqlx::query(&format!(
            r#"ALTER TABLE `{}` RENAME `{}`"#,
            from_collection_name.as_ref(),
            to_collection_name.as_ref(),
        ))
        .execute(&mut *tx)
        .await?;

        tx.commit().await?;

        Ok(())
    }

    pub async fn create(
        db: &MySqlPool,
        collection_name: impl AsRef<str>,
        collection_id: u64,
    ) -> Result<()> {
        let mut retry = true;
        loop {
            return match sqlx::query(&format!(
                "REPLACE INTO `{}`(`collection_name`, `collection_id`) VALUES (?, ?)",
                TENANT_COLLECTION_META_TABLE_NAME
            ))
            .bind(collection_name.as_ref())
            .bind(collection_id)
            .execute(db)
            .await
            {
                Err(e) => {
                    if retry && no_such_table(&e) {
                        warn!(
                            "Try create collection meta table '{}' with id '{}'/'{}'. {}",
                            TENANT_COLLECTION_META_TABLE_NAME,
                            collection_name.as_ref(),
                            collection_id,
                            e
                        );
                        Self::create_table(db).await?;
                        retry = false;
                        continue;
                    }
                    warn!(
                        "Failed to create collection '{}' with id '{}'. {}",
                        collection_name.as_ref(),
                        collection_id,
                        e
                    );
                    Err(e.into())
                }
                Ok(_) => Ok(()),
            };
        }
    }

    pub async fn delete(
        db: &MySqlPool,
        collection_name: impl AsRef<str>,
        collection_id: u64,
    ) -> Result<()> {
        let mut retry = true;
        loop {
            return match sqlx::query(&format!(
                r#"UPDATE `{}` SET `deleted_at` = NOW()
                WHERE `collection_name` = ? AND `collection_id` = ? AND `deleted_at` IS NULL"#,
                TENANT_COLLECTION_META_TABLE_NAME
            ))
            .bind(collection_name.as_ref())
            .bind(collection_id)
            .execute(db)
            .await
            {
                Err(e) => {
                    if retry && no_such_table(&e) {
                        Self::create_table(db).await?;
                        retry = false;
                        continue;
                    }
                    warn!(
                        "Failed to delete collection '{}' with id '{}'. {}",
                        collection_name.as_ref(),
                        collection_id,
                        e
                    );
                    Err(e.into())
                }
                Ok(_) => Ok(()),
            };
        }
    }
}