use crate::error;
use crate::error::QdrantError::*;
use crate::error::Result;
use crate::{metrics, Args};
use log::*;
use model::error::{db_exists, is_bad_db};
use model::meta::tenant::TenantCollectionMeta;
// use model::meta::CollectionMeta;
use model::run_migrate;
use moka::sync::Cache;
use sqlx::{
    mysql::{MySqlConnectOptions, MySqlPoolOptions},
    ConnectOptions, MySqlPool,
};
// use std::sync::atomic::AtomicU64;
// use std::sync::atomic::Ordering::SeqCst;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::RwLock;

const INDEX_VERSION_CACHE_DURATION_SEC: u64 = 60;
pub(crate) struct State {
    args: Args,
    tenants_db: Cache<String, Arc<MySqlPool>>,
    options: MySqlPoolOptions,
    meta_db: RwLock<Option<Arc<MySqlPool>>>,
    // meta_db_check_timestamp: AtomicU64,
    indexes_version: Cache<String, (u64, u64)>,
}

pub type ApiState = Arc<State>;

impl State {
    async fn connect_to_internal(
        ctx: &context::Context,
        user: &str,
        password: &str,
        host: &str,
        database: &str,
        options: MySqlPoolOptions,
    ) -> Result<Arc<MySqlPool>> {
        let mut database_in_url = Some(database);
        let mut create_table = false;
        let mut retry = true;
        loop {
            let opts: MySqlConnectOptions = format!(
                "mysql://{}:{}@{}:4000/{}",
                user,
                password,
                host,
                database_in_url.unwrap_or("")
            )
            .parse()?;

            let db = options
                .clone()
                .connect_with(
                    opts.log_slow_statements(LevelFilter::Warn, Duration::from_millis(500)),
                )
                .await;

            ctx.push_phase("create-db-connection", None);
            ctx.slow(Duration::from_secs(1));

            return match db {
                Ok(db) => {
                    if database_in_url.is_none() {
                        database_in_url = Some(database);

                        // create database and retry
                        if let Err(e) = sqlx::query(&format!("CREATE DATABASE {}", database))
                            .execute(&db)
                            .await
                        {
                            // database already exists, may be created by a different request
                            if db_exists(&e) {
                                warn!(
                                    "[{}] Database {} already exists for {}, may be created by a different request, try to reconnecting: {}",
                                    ctx.id(),
                                    database,
                                    user,
                                    e
                                );
                                continue;
                            }

                            error!(
                                "[{}] Failed to create database {}: {}",
                                ctx.id(),
                                database,
                                e
                            );
                            return Err(Unauthorized);
                        }
                        create_table = true;
                        continue;
                    }

                    if create_table {
                        // create tables
                        TenantCollectionMeta::create_table(&db).await?;
                    }

                    Ok(Arc::new(db))
                }
                Err(e) => {
                    if retry && is_bad_db(&e) {
                        // create database and connect again
                        database_in_url = None;
                        retry = false;
                        warn!("[{}] Database {} does not exist for {}, create database and try connecting again: {}",ctx.id(), database, user, e);
                        continue;
                    }
                    error!(
                        "[{}] Failed to connect to user[{}] host[{}]: {}",
                        ctx.id(),
                        user,
                        host,
                        e
                    );
                    Err(Unauthorized)
                }
            };
        }
    }

    async fn connect_to(
        &self,
        ctx: &context::Context,
        api_key: &str,
        user: &str,
        password: &str,
    ) -> Result<Arc<MySqlPool>> {
        metrics::REPORTS_TOTALS
            .with_label_values(&["cache", "tenant_db"])
            .set(self.tenants_db.weighted_size() as f64);
        if let Some(db) = self.tenants_db.get(api_key) {
            ctx.push_phase("get_tenant_db_from_cache", None);
            return Ok(db);
        }
        info!("[{}] connect to db for user[{}]", ctx.id(), user,);
        let db = match Self::connect_to_internal(
            ctx,
            user,
            password,
            &self.args.host,
            &self.args.database,
            copy_mysql_pool_options(&self.options),
        )
        .await
        {
            Ok(db) => db,
            Err(e) => {
                error!(
                    "[{}] Failed to connect to tenant-db [{}]: {}",
                    ctx.id(),
                    user,
                    e
                );
                return Err(e);
            }
        };
        self.tenants_db.insert(api_key.to_string(), db.clone());
        ctx.push_phase("get_tenant_db_from_new", None);
        Ok(db)
    }

    pub(crate) async fn get_index_version(&self, index_name: &str) -> Result<u64> {
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();
        if let Some((version, duration)) = self.indexes_version.get(index_name) {
            debug!(
                "Got version from cache: {}, now:{}, duration: {}",
                version, now, duration
            );
            if now - duration < INDEX_VERSION_CACHE_DURATION_SEC {
                return Ok(version);
            }
        }

        debug!(
            "Getting version from fts-index: {}, now:{}",
            index_name, now
        );
        let versions = match self.fts_client()?.get_versions(index_name).await {
            Ok(resp) => {
                info!("Got versions for {}: {:?}", index_name, resp);
                (resp.0, resp.1)
            }
            Err(e) => {
                warn!("Failed to get versions for {}: {}", index_name, e);
                return Ok(0);
            }
        };
        self.indexes_version
            .insert(index_name.into(), (versions.1, now));
        Ok(versions.1)
    }

    fn parse_api_key(api_key: &str) -> Result<(&str, &str, &str)> {
        let parts = api_key.split(':').collect::<Vec<&str>>();
        if parts.len() != 2 {
            return Err(Unauthorized);
        }
        let user = parts[0];
        let password = parts[1];
        let parts = user.split('.').collect::<Vec<&str>>();
        if parts.len() != 2 {
            return Err(Unauthorized);
        }
        let token = parts[0];
        Ok((token, user, password))
    }

    pub(crate) fn fts_client(&self) -> Result<fts_index::client::Client> {
        Ok(fts_index::client::Client::new(
            self.args.search_service.clone(),
        )?)
    }

    pub(crate) async fn get_meta_db(&self) -> Option<Arc<MySqlPool>> {
        self.meta_db.read().await.clone()
    }

    async fn get_meta_db_or_init(&self, ctx: &context::Context) -> Option<Arc<MySqlPool>> {
        let meta_db = self.meta_db.read().await.clone();
        match meta_db {
            Some(db) => Some(db.clone()),
            None => {
                match init_meta_db(
                    ctx,
                    &self.args.user,
                    &self.args.password,
                    &self.args.host,
                    &self.args.database,
                    self.options.clone(),
                )
                .await
                {
                    Some(db) => {
                        *self.meta_db.write().await = Some(db.clone());
                        Some(db)
                    }
                    None => None,
                }
            }
        }
    }

    pub(crate) async fn get_meta_db_and_token<'a>(
        &self,
        ctx: &context::Context,
        api_key: &'a str,
    ) -> Result<(Arc<MySqlPool>, &'a str)> {
        let (token, _, _) = Self::parse_api_key(api_key)?;
        ctx.push_phase("get_meta_db", None);

        let meta_db = self
            .get_meta_db_or_init(ctx)
            .await
            .ok_or(error::QdrantError::InternalServerError)?;

        Ok((meta_db, token))
    }

    pub(crate) async fn get_meta_and_tenant_db<'a>(
        &self,
        ctx: &context::Context,
        api_key: &'a str,
    ) -> Result<(Option<Arc<MySqlPool>>, Arc<MySqlPool>, &'a str)> {
        let (token, user, password) = Self::parse_api_key(api_key)?;
        let tenant_db = self.connect_to(ctx, api_key, user, password).await?;
        ctx.slow(Duration::from_secs(1));
        ctx.push_phase("get_tenant_db", None);

        Ok((self.get_meta_db_or_init(ctx).await, tenant_db, token))
    }

    pub(crate) async fn get_tenant_db<'a>(
        &self,
        ctx: &context::Context,
        api_key: &'a str,
    ) -> Result<(Arc<MySqlPool>, &'a str)> {
        let (token, user, password) = Self::parse_api_key(api_key)?;
        Ok((self.connect_to(ctx, api_key, user, password).await?, token))
    }

    pub(crate) async fn check_health(&self) -> Result<()> {
        // let now = SystemTime::now()
        //     .duration_since(UNIX_EPOCH)
        //     .unwrap_or_default()
        //     .as_secs();
        // let last_check = self.meta_db_check_timestamp.load(SeqCst);

        // match &self.meta_db.read().await.clone() {
        //     Some(db) => {
        //         // we check connection once every ten seconds
        //         if now - last_check > 10 {
        //             _ = CollectionMeta::first(&db).await?;
        //             self.meta_db_check_timestamp.store(now, SeqCst);
        //         }
        //     }
        //     None => return Err(InternalServerError),
        // };
        Ok(())
    }

    pub(crate) fn get_qudrant_api(&self) -> Result<&str> {
        self.args.qdrant_service.as_deref().ok_or(NotImplemented)
    }
}

fn copy_mysql_pool_options(options: &MySqlPoolOptions) -> MySqlPoolOptions {
    MySqlPoolOptions::new()
        .max_connections(options.get_max_connections())
        .min_connections(options.get_min_connections())
        .acquire_timeout(options.get_acquire_timeout())
        .max_lifetime(options.get_max_lifetime())
        .idle_timeout(options.get_idle_timeout())
}

pub(crate) async fn new(args: Args) -> anyhow::Result<ApiState> {
    info!("Fts server within: {}", args.search_service,);
    let ctx = context::with_interface_and_metrics(
        "state",
        metrics::INTERFACE_PERFORMANCE_HISTOGRAM.clone(),
    );

    info!(
        "Connecting to meta-db: db[{}] max-lifetime[{}] idle-timeout[{}]",
        args.database, args.max_lifetime, args.idle_timeout
    );

    let acquire_timeout = std::env::var("ACQUIRE_TIMEOUT_SEC")
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .unwrap_or(15);
    let options = MySqlPoolOptions::new()
        .max_connections(args.max_connection)
        .min_connections(args.min_connection)
        .acquire_timeout(Duration::from_secs(acquire_timeout))
        .max_lifetime(Duration::from_secs(args.max_lifetime))
        .idle_timeout(Duration::from_secs(args.idle_timeout));

    let meta_db = init_meta_db(
        &ctx,
        &args.user,
        &args.password,
        &args.host,
        &args.database,
        options.clone(),
    )
    .await;

    let tenants_db_cache_size = std::env::var("TENANTS_DB_CACHE_SIZE")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(1000);

    Ok(Arc::new(State {
        args,
        options,
        meta_db: RwLock::new(meta_db),
        tenants_db: Cache::new(tenants_db_cache_size),
        // meta_db_check_timestamp: AtomicU64::default(),
        indexes_version: Cache::new(10),
    }))
}

async fn init_meta_db(
    ctx: &context::Context,
    user: &str,
    password: &str,
    host: &str,
    database: &str,
    options: MySqlPoolOptions,
) -> Option<Arc<MySqlPool>> {
    match State::connect_to_internal(ctx, user, password, host, database, options).await {
        Ok(db) => {
            info!("Running migrations on meta-db");
            run_migrate(&db).await.unwrap();

            Some(db)
        }
        Err(e) => {
            error!(
                "[{}] Failed to connect to meta-db [{}]: {}",
                ctx.id(),
                user,
                e
            );
            None
        }
    }
}

#[cfg(test)]
mod test {
    use super::*;
    use crate::Args;

    #[tokio::test]
    async fn test_new() {
        let args = Args::default();
        let database = args.database.clone();
        let _ = new(args).await.unwrap();
        assert_eq!(database, "ai");
    }
}