use std::cell::Cell;
use std::ops::DerefMut;
use std::sync::{Arc};
use std::time::Duration;
use tokio::sync::Mutex;
use serde::{Deserialize, Serialize};

use rbatis::{Error, RBatis};
use flyway::{MigrationExecutor, MigrationState, MigrationStateManager, MigrationsError, MigrationStatus, ChangelogFile};
use rbs::{value, Value};
use async_trait::async_trait;
use rbatis::executor::RBatisTxExecutor;
use rbatis::rbatis_codegen::ops::AsProxy;
use rbatis::rbdc::datetime::DateTime;
use rbatis::rbdc::timestamp::Timestamp;

/// Default table name for the migration state management table
pub const DEFAULT_MIGRATIONS_TABLE: &str = "flyway_migrations";



#[derive(Clone, Debug, Serialize, Deserialize)]
struct MigrationInfo {
    ts:DateTime,
    version: u32,
    name: Option<String>,
    checksum: Option<String>,
    status:Option<String>,
}
/// Available driver types supported by Rbatis
pub enum RbatisDbDriverType {
    MySql,
    Pg,
    Sqlite,
    MsSql,
    TDengine,
    Other(String),
}
/// 不同数据库的建表语句
fn create_table_sql(db_type:RbatisDbDriverType, migrations_table_name: String) -> String {
        match db_type {
            RbatisDbDriverType::MySql => {
              format!(r#"CREATE TABLE IF NOT EXISTS {} (
                version INTEGER PRIMARY KEY,
                ts       varchar(255) null,
                name     varchar(255) null,
                checksum   varchar(255) null,
                status VARCHAR(16)
            );"#,migrations_table_name)
            }
            RbatisDbDriverType::Pg => {
                format!(r#"CREATE TABLE IF NOT EXISTS {} (
                version INTEGER PRIMARY KEY,
                ts       varchar(255) null,
                name     varchar(255) null,
                checksum   varchar(255) null,
                status VARCHAR(16)
            );"#,migrations_table_name)
            }
            RbatisDbDriverType::Sqlite => {
                format!(r#"CREATE TABLE IF NOT EXISTS {} (
                version INTEGER PRIMARY KEY,
                ts       varchar(255) null,
                name     varchar(255) null,
                checksum   varchar(255) null,
                status VARCHAR(16)
            );"#,migrations_table_name)
            }
            RbatisDbDriverType::MsSql => {
                format!(r#"IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='{}' AND xtype='U')
                CREATE TABLE {} (
                version INTEGER PRIMARY KEY,
                ts       varchar(255) null,
                name     varchar(255) null,
                checksum   varchar(255) null,
                status VARCHAR(16)
            );"#,migrations_table_name, migrations_table_name)
            }
            RbatisDbDriverType::TDengine => {
               format!(r#"CREATE TABLE IF NOT EXISTS {} (`ts` TIMESTAMP, `version` int,`name` nchar(255) , `checksum` nchar(255), `status` nchar(255))
                  "#,migrations_table_name)
            }
            RbatisDbDriverType::Other(_) => {
                format!(r#"CREATE TABLE IF NOT EXISTS {} (
                version INTEGER PRIMARY KEY,
                ts       varchar(255) null,
                name     varchar(255) null,
                checksum   varchar(255) null,
                status VARCHAR(16)
            );"#,migrations_table_name)
            }
        }
}
/// 不同数据库的update
fn update_sql(db_type:RbatisDbDriverType,migrations_table_name: String,status:String,version:u64)->String{
    match db_type {
        RbatisDbDriverType::MySql => {
            format!(r#"UPDATE {} SET status='{}' where version={};"#,
                    migrations_table_name.as_str(),status.as_str(), version)
        }
        RbatisDbDriverType::Pg => {
            format!(r#"UPDATE {} SET status='{}' where version={};"#,
                    migrations_table_name.as_str(),status.as_str(), version)
        }
        RbatisDbDriverType::Sqlite => {
            format!(r#"UPDATE {} SET status='{}' where version={};"#,
                    migrations_table_name.as_str(),status.as_str(), version)
        }
        RbatisDbDriverType::MsSql => {
            format!(r#"UPDATE {} SET status='{}' where version={};"#,
                    migrations_table_name.as_str(),status.as_str(), version)
        }
        RbatisDbDriverType::TDengine => {
            // TDengine 作为时序数据库，不支持标准 UPDATE 语法
            // 需要通过插入新记录来更新状态
            format!(r#"INSERT INTO {}(ts,version,name,checksum,status) VALUES (NOW,{},'','','{}');"#,
                    migrations_table_name.as_str(), version, status.as_str())
        }
        RbatisDbDriverType::Other(_) => {
            format!(r#"UPDATE {} SET status='{}' where version={};"#,
                    migrations_table_name.as_str(),status.as_str(), version)
        }
    }
}
/// 不同数据库的insert
fn insert_sql(db_type:RbatisDbDriverType,migrations_table_name: String,status:String)->String{
    match db_type {
        RbatisDbDriverType::MySql => {
            format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, '{}');"#,
                    migrations_table_name.as_str(),status.as_str())
        }
        RbatisDbDriverType::Pg => {
            format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, '{}');"#,
                    migrations_table_name.as_str(),status.as_str())
        }
        RbatisDbDriverType::Sqlite => {
            format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, '{}');"#,
                    migrations_table_name.as_str(),status.as_str())
        }
        RbatisDbDriverType::MsSql => {
            format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, '{}');"#,
                    migrations_table_name.as_str(),status.as_str())
        }
        RbatisDbDriverType::TDengine => {
            format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, '{}');"#,
                    migrations_table_name.as_str(),status.as_str())
        }
        RbatisDbDriverType::Other(_) => {
            format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, '{}');"#,
                    migrations_table_name.as_str(),status.as_str())
        }
    }
}

/// Rbatis implementation of `MigrationStateManager` and `MigrationExecutor`
pub struct RbatisMigrationDriver {
    db: Arc<RBatis>,
    migrations_table_name: String,
    tx: Mutex<Cell<Option<RBatisTxExecutor>>>,
}

impl RbatisMigrationDriver {
    /// Create a new driver
    ///
    ///  * `db`: The `Rbatis` instance for accessing the database
    ///  * `migrations_table_name`: The optional name of the table the migration state information
    ///    should be stored in. If `None`, the `DEFAULT_MIGRATIONS_TABLE` will be used.
    pub fn new(db: Arc<RBatis>, migrations_table_name: Option<&str>) -> RbatisMigrationDriver {
        return RbatisMigrationDriver {
            db: db.clone(),
            migrations_table_name: migrations_table_name.map(|v| v.to_string())
                .or(Some(DEFAULT_MIGRATIONS_TABLE.to_string()))
                .unwrap(),
            tx: Mutex::new(Cell::new(None)),
        }
    }

    /// The the driver type of the `Rbatis` instance
    ///
    /// This method will get the driver type from `Rbatis` (which is a string) and convert it into
    /// an `RbatisDbDriverType`. `Other(String)` will be used for any database drivers not directly
    /// known to `flyway-rbatis`.
    pub fn driver_type(&self) -> rbatis::Result<RbatisDbDriverType> {
        let db = self.db.clone();
        let driver_type_name = db.driver_type()?;
        let result = match driver_type_name {
            "mssql" => RbatisDbDriverType::MsSql,
            "mysql" => RbatisDbDriverType::MySql,
            "postgres" => RbatisDbDriverType::Pg,
            "sqlite" => RbatisDbDriverType::Sqlite,
            "Taos"=>RbatisDbDriverType::TDengine,
            _ => RbatisDbDriverType::Other(driver_type_name.to_string())
        };
        return Ok(result);
    }
}

/// Implementation of the `MigrationStateManager`
#[async_trait]
impl MigrationStateManager for RbatisMigrationDriver {
    async fn prepare(&self) -> flyway::Result<()> {
        log::debug!("Preparing Migrations Table ...");
        let db = self.db.clone();
      let statement=create_table_sql(self.driver_type().unwrap(),self.migrations_table_name.clone());
        let mut db = db.acquire()
            .await
            .or_else(|err| Err(MigrationsError::migration_database_failed(None, Some(err.into()))))?;

        log::debug!("Preparation Statement: {}", statement.as_str());
        let _result = db.exec(statement.as_str(), vec![])
            .await
            .or_else(|err| Err(MigrationsError::migration_setup_failed(Some(err.into()))))?;
        log::debug!("Preparing Migrations Table ... done");
        return Ok(());
    }

    async fn lowest_version(&self) -> flyway::Result<Option<MigrationState>> {
        log::debug!("Retrieving lowest version ... ");
        let db = self.db.clone();
        let mut db = db.acquire()
            .await
            .or_else(|err| Err(MigrationsError::migration_database_failed(None, Some(err.into()))))?;
        let version: Option<u64> = db.rb.query_decode(format!("SELECT MIN(version) FROM {} WHERE status='deployed';",
                                                           self.migrations_table_name.as_str()).as_str(), vec![])
            .await
            .or_else(|err| Err(MigrationsError::migration_versioning_failed(Some(err.into()))))?;

        log::debug!("Retrieving lowest version ... {:?}", &version);
        return Ok(version.and_then(|version|
            Some(MigrationState {
                version,
                status: MigrationStatus::Deployed
            })));
    }

    async fn highest_version(&self) -> flyway::Result<Option<MigrationState>> {
        log::debug!("Retrieving highest version ... ");
        let db = self.db.clone();
        let mut db = db.acquire()
            .await
            .or_else(|err| Err(MigrationsError::migration_database_failed(None, Some(err.into()))))?;
        let version: Option<u64> = db.rb.query_decode(format!("SELECT MAX(version) FROM {} WHERE status='deployed';",
                                                           self.migrations_table_name.as_str()).as_str(), vec![])
            .await
            .or_else(|err| Err(MigrationsError::migration_versioning_failed(Some(err.into()))))?;

        log::debug!("Retrieving highest version ... {:?}", &version);
        return Ok(version.and_then(|version|
            Some(MigrationState {
                version,
                status: MigrationStatus::Deployed
            })));
    }

    async fn list_versions(&self) -> flyway::Result<Vec<MigrationState>> {
        log::debug!("Listing versions ... ");
        let db = self.db.clone();
        let mut db = db.acquire()
            .await
            .or_else(|err| Err(MigrationsError::migration_database_failed(None, Some(err.into()))))?;
        let versions: Vec<u64> = db.rb.query_decode(format!("SELECT version FROM {} WHERE status='deployed' ORDER BY version asc;",
                                                         self.migrations_table_name.as_str()).as_str(), vec![])
            .await
            .or_else(|err| Err(MigrationsError::migration_versioning_failed(Some(err.into()))))?;

        let versions: Vec<MigrationState> = versions.iter()
            .map(|version|
                MigrationState {
                    version: *version,
                    status: MigrationStatus::Deployed
                })
            .collect();

        log::debug!("Listing versions ... {:?}", &versions);
        return Ok(versions);
    }

    async fn begin_version(&self, changelog_file: &ChangelogFile) -> flyway::Result<()> {
        log::debug!("Beginning version ... {}", changelog_file.version);
        let db = self.db.clone();
        let mut db = db.acquire()
            .await
            .or_else(|err| Err(MigrationsError::migration_database_failed(None, Some(err.into()))))?;

       match   self.driver_type(){
           Ok(db_type) => {
               match db_type {
                   RbatisDbDriverType::TDengine => {
                       let mut ts:i64=DateTime::utc().unix_timestamp_millis()+changelog_file.version() as i64;
                       let ts_select=format!(r#"select ts,version from {} where status='in_progress' and version=? limit 1;"#, self.migrations_table_name.as_str());
                       match   db.rb.query_decode::<Vec<MigrationInfo>>(ts_select.as_str(),vec![value!(changelog_file.version.clone())]).await{
                           Ok(result) => {
                               // println!("{:?}",result);
                              if result.first().is_some(){
                                  let  time=result.first().unwrap().ts.clone().set_offset(-16*60*60);
                                   ts=time.unix_timestamp_millis();
                              }
                           }
                           Err(e) => {
                               log::error!("数据异常:{}",e.to_string())
                           }
                       };
                       let insert_statement = format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, 'in_progress');"#,
                                                      self.migrations_table_name.as_str());
                       log::debug!("Insert statement: {}", insert_statement.as_str());
                       let _insert_result = db.exec(insert_statement.as_str(), vec![value!(ts),value!(changelog_file.version.clone()),value!(changelog_file.name.clone()),value!(changelog_file.checksum.clone())])
                           .await
                           .or_else(|err| Err(MigrationsError::migration_versioning_failed(Some(err.into()))))?;
                       return Ok(());
                   }
                 _ => {}
               }
           }
           Err(_) => {}
       }

        // let update_statement = format!(r#"UPDATE {} SET status='in_progress' where version={};"#,
        //                                self.migrations_table_name.as_str(), changelog_file.version);
        let update_statement =update_sql(self.driver_type().unwrap(),self.migrations_table_name.clone(),"in_progress".to_string(),changelog_file.version.clone());

        log::debug!("Update statement: {}", update_statement.as_str());
        let update_result = db.exec(update_statement.as_str(), vec![])
            .await
            .or_else(|err| Err(MigrationsError::migration_versioning_failed(Some(err.into()))))?;

        if update_result.rows_affected < 1 {
            let  ts:i64=DateTime::utc().unix_timestamp_millis()+changelog_file.version as i64;

            // let insert_statement = format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, 'in_progress');"#,
            //                                self.migrations_table_name.as_str());
           let insert_statement=insert_sql(self.driver_type().unwrap(),self.migrations_table_name.clone(),"in_progress".to_string());
            log::debug!("Insert statement: {}", insert_statement.as_str());
            let _insert_result = db.exec(insert_statement.as_str(), vec![value!(ts),value!(changelog_file.version.clone()),value!(changelog_file.name.clone()),value!(changelog_file.checksum.clone())])
                .await
                .or_else(|err| Err(MigrationsError::migration_versioning_failed(Some(err.into()))))?;
        }

        return Ok(());
    }

    async fn finish_version(&self, changelog_file: &ChangelogFile) -> flyway::Result<()> {
        log::debug!("Finishing version ... {}", changelog_file.version);
        let db = self.db.clone();
        let mut db = db.acquire()
            .await
            .or_else(|err| Err(MigrationsError::migration_database_failed(None, Some(err.into()))))?;


        match   self.driver_type(){
            Ok(db_type) => {
                match db_type {
                    RbatisDbDriverType::TDengine => {
                        let mut ts:i64=DateTime::utc().unix_timestamp_millis()+changelog_file.version as i64;
                        let ts_select=format!(r#"select ts,version from {} where status='in_progress' and version=? limit 1;"#, self.migrations_table_name.as_str());
                        match   db.rb.query_decode::<Vec<MigrationInfo>>(ts_select.as_str(),vec![value!(changelog_file.version.clone())]).await{
                            Ok(result) => {
                                if result.first().is_some(){
                                    let  time=result.first().unwrap().ts.clone().set_offset(-16*60*60);
                                    ts=time.unix_timestamp_millis();                               }
                            }
                            Err(e) => {
                                log::error!("数据异常:{}",e.to_string())
                            }
                        };

                        let insert_statement = format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, 'deployed');"#,
                                                       self.migrations_table_name.as_str());
                        log::debug!("Insert statement: {}", insert_statement.as_str());
                        let _insert_result = db.exec(insert_statement.as_str(), vec![value!(ts),value!(changelog_file.version.clone()),value!(changelog_file.name.clone()),value!(changelog_file.checksum.clone())])
                            .await
                            .or_else(|err| Err(MigrationsError::migration_versioning_failed(Some(err.into()))))?;
                        return Ok(());
                    }
                    _ => {}
                }
            }
            Err(_) => {}
        }


        // let update_statement = format!(r#"UPDATE {} SET status='deployed' where version={};"#,
        //                                self.migrations_table_name.as_str(), changelog_file.version);
        let update_statement =update_sql(self.driver_type().unwrap(),self.migrations_table_name.clone(),"deployed".to_string(),changelog_file.version.clone());

        log::debug!("Update statement: {}", update_statement.as_str());
        let update_result = db.exec(update_statement.as_str(), vec![])
            .await
            .or_else(|err| Err(MigrationsError::migration_versioning_failed(Some(err.into()))))?;

        if update_result.rows_affected < 1 {
            let  ts:i64=DateTime::utc().unix_timestamp_millis()+changelog_file.version as i64;

            // let insert_statement = format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, 'in_progress');"#,
            //                                self.migrations_table_name.as_str());
            let insert_statement=insert_sql(self.driver_type().unwrap(),self.migrations_table_name.clone(),"in_progress".to_string());

            log::debug!("Insert statement: {}", insert_statement.as_str());
            let _insert_result = db.exec(insert_statement.as_str(), vec![value!(ts),value!(changelog_file.version.clone()),value!(changelog_file.name.clone()),value!(changelog_file.checksum.clone())])
                .await
                .or_else(|err| Err(MigrationsError::migration_versioning_failed(Some(err.into()))))?;
        }

        return Ok(());
    }

    async fn skip_version(&self, changelog_file: &ChangelogFile) -> flyway::Result<()> {
        log::debug!("Skip version ... {}", changelog_file.version);
        let db = self.db.clone();
        let mut db = db.acquire()
            .await
            .or_else(|err| Err(MigrationsError::migration_database_failed(None, Some(err.into()))))?;


        match   self.driver_type(){
            Ok(db_type) => {
                match db_type {
                    RbatisDbDriverType::TDengine => {
                        let mut ts:i64=DateTime::utc().unix_timestamp_millis()+changelog_file.version as i64;
                        let ts_select=format!(r#"select ts,version from {} where status='in_progress' and version=? limit 1;"#, self.migrations_table_name.as_str());
                        match   db.rb.query_decode::<Vec<MigrationInfo>>(ts_select.as_str(),vec![value!(changelog_file.version.clone())]).await{
                            Ok(result) => {
                                if result.first().is_some(){
                                    let  time=result.first().unwrap().ts.clone().set_offset(-16*60*60);
                                    ts=time.unix_timestamp_millis();                               }
                            }
                            Err(e) => {
                                log::error!("数据异常:{}",e.to_string())
                            }
                        };

                        let insert_statement = format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, 'fail');"#,
                                                       self.migrations_table_name.as_str());
                        log::debug!("Insert statement: {}", insert_statement.as_str());
                        let _insert_result = db.exec(insert_statement.as_str(), vec![value!(ts),value!(changelog_file.version.clone()),value!(changelog_file.name.clone()),value!(changelog_file.checksum.clone())])
                            .await
                            .or_else(|err| Err(MigrationsError::migration_versioning_failed(Some(err.into()))))?;
                        return Ok(());
                    }
                    _ => {}
                }
            }
            Err(_) => {}
        }


        // let update_statement = format!(r#"UPDATE {} SET status='deployed' where version={};"#,
        //                                self.migrations_table_name.as_str(), changelog_file.version);
        let update_statement =update_sql(self.driver_type().unwrap(),self.migrations_table_name.clone(),"fail".to_string(),changelog_file.version.clone());

        log::debug!("Update statement: {}", update_statement.as_str());
        let update_result = db.exec(update_statement.as_str(), vec![])
            .await
            .or_else(|err| Err(MigrationsError::migration_versioning_failed(Some(err.into()))))?;

        if update_result.rows_affected < 1 {
            let  ts:i64=DateTime::utc().unix_timestamp_millis()+changelog_file.version as i64;

            // let insert_statement = format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, 'in_progress');"#,
            //                                self.migrations_table_name.as_str());
            let insert_statement=insert_sql(self.driver_type().unwrap(),self.migrations_table_name.clone(),"in_progress".to_string());

            log::debug!("Insert statement: {}", insert_statement.as_str());
            let _insert_result = db.exec(insert_statement.as_str(), vec![value!(ts),value!(changelog_file.version.clone()),value!(changelog_file.name.clone()),value!(changelog_file.checksum.clone())])
                .await
                .or_else(|err| Err(MigrationsError::migration_versioning_failed(Some(err.into()))))?;
        }

        return Ok(());
    }
}

/// Implementation of the `MigrationExecutor`
#[async_trait]
impl MigrationExecutor for RbatisMigrationDriver {
    async fn begin_transaction(&self) -> flyway::Result<()> {
        log::debug!("Beginning transaction ...");
        {
            let mut tx_guard = self.tx.lock().await;
            if tx_guard.get_mut().is_some() {
                return Err(MigrationsError::migration_database_failed(None, None));
            }
        }

        let tx = {
            let db = self.db.clone();
            db.acquire_begin()
                .await
                .or_else(|err| Err(MigrationsError::migration_versioning_failed(Some(err.into()))))?
        };

        let tx_guard = self.tx.lock().await;
        tx_guard.set(Some(tx));
        return Ok(());
    }

    async fn execute_changelog_file(&self, changelog_file: &flyway::ChangelogFile) -> flyway::Result<()> {
        log::debug!("Executing changelog file ... {:?}", &changelog_file);
        let mut tx_guard = self.tx.lock().await;
        let tx = tx_guard.get_mut().as_mut();
        match tx {
            Some(tx) => {
                for statement in changelog_file.iter() {
                    log::debug!("Executing statement: {}", statement.statement.as_str());
                    let may_fail = statement.annotation.as_ref()
                        .map(|a| a.may_fail())
                        .unwrap_or(false);
                    let result = tx.exec(statement.statement.as_str(), vec![]).await;
                    match result {
                        Ok(_) => {},
                        Err(err) => {
                            if may_fail {
                                log::warn!("Statement failed but may_fail is set, continuing: {}", err);
                            } else {
                                return Err(MigrationsError::migration_database_step_failed(None, Some(err.into())));
                            }
                        }
                    }
                }
            },
            None => {
                return Err(MigrationsError::migration_database_failed(None, None));
            }
        };
        return Ok(());
    }

    async fn commit_transaction(&self) -> flyway::Result<()> {
        log::debug!("Committing transaction ...");
        let mut tx = {
            let tx_guard = self.tx.lock().await;
            tx_guard.replace(None)
        };

        match tx.as_mut() {
            Some(tx) => {
                return tx.commit().await
                    .map(|_| ())
                    .or_else(|err| Err(MigrationsError::migration_database_failed(None, Some(err.into()))));
            }
            None => {
                return Err(MigrationsError::migration_database_failed(None, None));
            }
        }
    }

    async fn rollback_transaction(&self) -> flyway::Result<()> {
        log::debug!("Rolling back transaction ...");
        let mut tx = {
            let tx_guard = self.tx.lock().await;
            tx_guard.replace(None)
        };

        match tx.as_mut() {
            Some(tx) => {
                return tx.rollback().await
                    .map(|_| ())
                    .or_else(|err| Err(MigrationsError::migration_database_failed(None, Some(err.into()))));
            }
            None => {
                return Err(MigrationsError::migration_database_failed(None, None));
            }
        }
    }
}