/*
 * @Author: venom
 * @Date: 2021-08-13 16:42:05
 * @LastEditors: BuddyCoder
 * @LastEditTime: 2024-05-23 07:55:41
 * @Description:
 * @FilePath: /pgbatis/src/lib.rs
 * MIT
 */

use lazy_static::lazy_static;
use std::collections::HashMap;
use std::sync::Mutex;
use tokio_postgres::Error;

use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod};
use once_cell::sync::Lazy;
pub use pgmacro;
pub use plugin::logic_delete::LogicDelete;
pub use plugin::logic_delete::LogicDeletePlugin;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::fmt::Debug;
use tokio::sync::OnceCell;
pub use tokio_postgres;
pub use tokio_postgres::{types::ToSql, Client, NoTls, Row};
pub mod plugin;
pub mod wrapper;
pub use wrapper::Wrapper;
pub mod unit;
use tracing::{debug, error, info, trace};

//迫不得已，当连接断开之后 无法重新连接。只能启用连接一次 重连一次的做法。每次都生成独立的连接
//性能必定受损，罢了罢了。虽然 deadpool 是一个不错的连接池
static EMPTY_STRING: &'static str = r#""#;
pub const LOG_SPACE: &'static str =
    "                                                                ";
lazy_static! {
    static  ref EXCLUSION_TABLE: Mutex<Vec<String>>  = {
        Mutex::new(Vec::new())
      //  Arc::new(m)
    };
}

static DATABASE_POOL: Lazy<Pgbatis> = Lazy::new(|| {
    let mut m = Pgbatis::new();
    let exclusion_table_m = EXCLUSION_TABLE.lock().unwrap();
    let exclusion_table = exclusion_table_m.clone();
    // let mut exclusion_table = Vec::new();
    // for exclusion in exclusion_table_str {
    //     let table = String::from(exclusion);
    //     exclusion_table.push(table);
    // }

    m.logic_plugin = Some(Box::new(LogicDeletePlugin::new_opt(
        "is_deleted",
        1,
        0,
        exclusion_table,
    )));
    // let mut exclusion_table = Vec::new();
    m
});

//定义参数的特性
pub trait Parameters {
    // fn get_table_name(prefix: &str,suffix:&str) -> String;
    fn get_table_name(prefix: Option<String>, suffix: Option<String>) -> String;
    fn get_field_list() -> String;
    //生成保存的字符串
    fn gen_save(
        &self,
        prefix: Option<String>,
        suffix: Option<String>,
    ) -> Result<(String, Vec<&(dyn ToSql + Sync)>), String>;
    //生成修改更新的字符串
    fn gen_update(
        &self,
        prefix: Option<String>,
        suffix: Option<String>,
    ) -> Result<(String, Vec<&(dyn ToSql + Sync)>, u32), String>;
    fn return_one(rows: Row) -> Self;
    fn return_list(rows: Vec<Row>) -> Vec<Self>
    where
        Self: Sized,
    {
        let mut result_list = Vec::new();
        for row in rows {
            result_list.push(Self::return_one(row))
        }

        return result_list;
    }
}

pub trait ColumnExt {
    fn get(&self) -> &'static str;
}

//声明一个tokio_postgres的连接管理类

pub struct Pgbatis {
    pub pools: OnceCell<Pool>,
    //pub client:HashMap<i32,Client>,
    pub logic_plugin: Option<Box<dyn LogicDelete>>,
}

impl Pgbatis {
    pub fn new() -> Self {
        return Self::new_with_opt();
    }

    ///new Rbatis from Option
    pub fn new_with_opt() -> Self {
        return Self {
            pools: OnceCell::new(),
            //逻辑删除插件
            logic_plugin: None,
        };
    }

    pub fn set_logic_delete(&mut self, arg: Option<impl LogicDelete + 'static>) {
        match arg {
            Some(v) => {
                self.logic_plugin = Some(Box::new(v));
            }
            None => {
                self.logic_plugin = None;
            }
        }
        //   self.logic_plugin = Some(arg)
    }
}

fn trace_exec_log(sql: &str, args: &[&(dyn ToSql + Sync)]) {
    let format = format!(
        "Exec   ==> {}\n{}[pgbatis] [{}] Args   ==> {:?}",
        &sql, LOG_SPACE, "", args
    );
    trace!("{}", format);
}

//添加
///设置无逻辑删除的表
pub fn set_unlogic_delete_table(table_name: &str) {
    let mut exclusion_table_m = EXCLUSION_TABLE.lock().unwrap();
    exclusion_table_m.push(table_name.to_string());
}

pub async fn link(
    host: &str,
    port: u16,
    user: &str,
    password: &str,
    dbname: &str,
    max_size: usize,
) -> Result<(), String> {
    let mut pg_config = tokio_postgres::Config::new();
    //   pg_config.options(&driver_url);
    pg_config.host(host);
    pg_config.port(port);
    pg_config.user(user);
    pg_config.password(password);
    pg_config.dbname(dbname);
    let mgr_config = ManagerConfig {
        recycling_method: RecyclingMethod::Verified,
    };

    let mgr = Manager::from_config(pg_config, NoTls, mgr_config);
    //创建连接池数量
    let pool = Pool::new(mgr, max_size);
    let _ = DATABASE_POOL.pools.set(pool);

    let driver_url = format!("postgresql://{}:{}@{}/{}", user, password, host, dbname);
    info!("数据库连接:{}", driver_url);
    //trace_exec_log(&driver_url, args);

    //DATABASE_POOL.logic_plugin.as_ref().unwrap().insert_exclusion_table("hellow");

    return Ok(());
}

#[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq)]
pub struct PageT<T>
where
    T: Parameters,
{
    /// data
    pub records: Vec<T>,
    /// total num
    pub total: u64,
    /// pages
    pub pages: u64,
    /// current page index
    pub page_no: u64,
    /// default 10
    pub page_size: u64,
}
#[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq)]
pub struct PageHash {
    /// data
    pub records: Vec<HashMap<String, Value>>,
    /// total num
    pub total: u64,
    /// pages
    pub pages: u64,
    /// current page index
    pub page_no: u64,
    /// default 10
    pub page_size: u64,
}

pub async fn execute(sql: &String, params: &[&(dyn ToSql + Sync)]) -> Result<u64, Error> {
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    let statement = client.prepare(sql.as_str()).await.unwrap();
    let result = client.execute(&statement, &params).await;
    match result {
        Ok(t) => {
            trace!("execute 执行成功");
            return Ok(t);
        }
        Err(e) => {
            // let err_str = format!("e:{}", e);
            error!("{:?}", e);
            return Err(e);
        }
    }
}

//如果表名需要前缀
//pub async fn pix_save<T>(p: T,prefix:&str) -> Result<u64, String>
/// 执行保存操作，执行insert操作
pub async fn save<T>(p: T, prefix: Option<String>, suffix: Option<String>) -> Result<u64, Error>
where
    T: Parameters,
{
    let (sql, args) = p.gen_save(prefix, suffix).unwrap();
    //跟踪日志
    trace_exec_log(&sql, &args);
    //取出  client
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    let statement = client.prepare(sql.as_str()).await.unwrap();
    let result = client.execute(&statement, &args).await;
    match result {
        Ok(t) => {
            trace!("SAVE 执行成功");
            return Ok(t);
        }
        Err(e) => {
            //let err_str = format!("e:{}", e);
            error!("{:?}", e);
            return Err(e);
        }
    }
}
//如果表名需要前缀
//pub async fn update<T>(p: T, wrapper: Wrapper<'_>,prefix:&str) -> Result<u64, String>
/// 执行更新操作，执行update操作
pub async fn update<T>(
    p: T,
    wrapper: Wrapper<'_>,
    prefix: Option<String>,
    suffix: Option<String>,
) -> Result<u64, Error>
where
    T: Parameters,
{
    //p.set_prefix(prefix);
    let (mut sql, mut args, args_number) = p.gen_update(prefix.clone(), suffix.clone()).unwrap();

    //获取逻辑删除部分的SQL
    let table_name = T::get_table_name(prefix, suffix);
    let logic_plugin_sql = get_logic_undelete(&table_name);

    let (where_sql, mut where_args) = wrapper.build(args_number).unwrap();
    if !where_sql.is_empty() && !logic_plugin_sql.is_empty() {
        sql = format!("{} WHERE {} AND {}", sql, where_sql, logic_plugin_sql);
    } else if !where_sql.is_empty() && logic_plugin_sql.is_empty() {
        sql = format!("{} WHERE {}  ", sql, where_sql);
    } else if where_sql.is_empty() && !logic_plugin_sql.is_empty() {
        sql = format!("{} WHERE {}  ", sql, logic_plugin_sql);
    }
    args.append(&mut where_args);
    //跟踪日志
    trace_exec_log(&sql, &args);
    //取出  client
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    let statement = client.prepare(sql.as_str()).await.unwrap();
    let result = client.execute(&statement, &args).await;

    match result {
        Ok(t) => {
          //  info!("更新执行成功");
            return Ok(t);
        }
        Err(e) => {
            //  let err_str = format!("e:{}", e);
            error!("{:?}", e);
            return Err(e);
        }
    }
}

/// 通过唯一ID执行删除操作
pub async fn remove<P>(
    wrapper: Wrapper<'_>,
    prefix: Option<String>,
    suffix: Option<String>,
) -> Result<u64, Error>
where
    P: Parameters,
{
    //正常的写法  DELETE FROM tableName    WHERE
    let table_name = P::get_table_name(prefix, suffix);
    let mut sql = format!("DELETE FROM \"{}\"", &table_name);

    //判断是否有逻辑删除插件
    //如果有，进入逻辑删除插件
    if DATABASE_POOL.logic_plugin.is_some() {
        // 判断表名是否存在排除表中， 如果不存在，sql变成逻辑删除的方式
        if !DATABASE_POOL
            .logic_plugin
            .as_ref()
            .unwrap()
            .is_exclusion_table(table_name.as_str())
        {
            sql = format!(
                "UPDATE \"{}\" set \"{}\" = {} ",
                table_name,
                DATABASE_POOL.logic_plugin.as_ref().unwrap().column(),
                DATABASE_POOL.logic_plugin.as_ref().unwrap().deleted(),
            );
        }
    }
    let (where_sql, args) = wrapper.build(1).unwrap();
    if !where_sql.is_empty() {
        sql = format!("{} where {}", sql, where_sql);
    }
    //跟踪日志
    trace_exec_log(&sql, &args);

    //取出  client
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    let statement = client.prepare(sql.as_str()).await.unwrap();
    let result = client.execute(&statement, &args).await;
    match result {
        Ok(t) => {
           // info!("更新执行成功");
            return Ok(t);
        }
        Err(e) => {
            //  let err_str = format!("e:{}", e);
            error!("{:?}", e);
            return Err(e);
        }
    }
}

/// 通过一个字段执行删除操作
// pub async fn remove_by_column<P,C,T >(column: C, value: T,prefix: Option<String>,suffix:Option<String>) -> Result<u64, Error>
// where
//     P: Parameters,
//     C: ColumnExt,
//     T: ToSql + Sync,
// {
//     //获取逻辑删除部分的SQL
//     let table_name = P::get_table_name(prefix,suffix);
//     let wrapper = Wrapper::new().eq(&column, &value);
//     remove(&table_name, wrapper).await
// }

/// 1、需要定义一个 trait fetch_result{}  用于赋值返回值与 新增修改的宏 进行分离
///   主要目的需要查询数据库数据回来进行计算的业务。定义返回值，单表使用比较多
/// fetch<T>(wrapper) -> Result<Vec<T>, Error>   单表不翻页   完成
/// fetch_one<T>(wrapper) ->Result<T, Error>         单表单条数据
/// fetch_page<T>(wrapper) ->Result<PageT<>, Error>     单表翻页

/// 2、返回值定义Vec<HashMap<String,Value>> ,好处就是不需要定义返回值的结构体，可以根据需求调整sql的返回值，直接rep.
/// query(&sql,&[args])->Result<Vec<HashMap<String,Value>>, Error>  主要用于联合查询,统计查询
/// query_one(&sql,&[args])->Result<HashMap<String,Value>,Error>    联合查询单条数据
/// query_page(&[返回字段],SQL,&[args])->Result<PageHash<HashMap<String,Value>>,Error> 联合查询进行翻页 统计查询

///单表不翻页
pub async fn fetch<T>(
    wrapper: Wrapper<'_>,
    prefix: Option<String>,
    suffix: Option<String>,
) -> Result<Vec<T>, Error>
where
    T: Parameters,
{
    let field_name = wrapper.clone().get_recoder_field::<T>();
    let table_name = T::get_table_name(prefix, suffix);
    // let order_by = wrapper.order_by.clone();
    //  let desc = wrapper.desc.clone();
    let (where_sql, args) = wrapper.clone().build(1).unwrap();
    //获取逻辑未删除部分的SQL
    let logic_plugin_sql = get_logic_undelete(&table_name);

    let mut sql = String::new();
    sql.push_str("SELECT ");
    sql.push_str(field_name.as_str());
    sql.push_str(" FROM ");
    sql.push_str(table_name.as_str());

    let mut had_where = false;
    // if args.len() != 0 {
    //     sql.push_str(" WHERE ");
    //     sql.push_str(where_sql.as_str());
    //     had_where = true;
    // }
    if !where_sql.is_empty() {
        sql.push_str(" WHERE ");
        sql.push_str(where_sql.as_str());
        had_where = true;
    }

    if !logic_plugin_sql.is_empty() {
        if had_where {
            sql.push_str(" AND ");
        } else {
            sql.push_str(" WHERE ");
        }
        sql.push_str(logic_plugin_sql.as_str());
    }

    // if order_by !== None{
    //     let desc_str = if desc {"desc"} else {"asc"};
    //     let order_by_format = format!("order by {} {} " ,order_by.clone(),desc_str);
    let order_by_format = match wrapper.get_order_by() {
        Some(t) => t,
        None => " ".to_string(),
    };
    sql.push_str(&order_by_format);

    //跟踪日志
    trace_exec_log(&sql, &args);
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    let statement = client.prepare(sql.as_str()).await.unwrap();
    let result = client.query(&statement, &args).await?;
    //  println!("result:{:?}", result);

    let result_t = T::return_list(result);

    return Ok(result_t);
}

pub async fn fetch_with_recoder_field<T>(
    wrapper: Wrapper<'_>,
    prefix: Option<String>,
    suffix: Option<String>,
) -> Result<Vec<HashMap<String, Value>>, Error>
where
    T: Parameters,
{
    let field_name = wrapper.clone().get_recoder_field::<T>();
    let table_name = T::get_table_name(prefix, suffix);
    // let order_by = wrapper.order_by.clone();
    //  let desc = wrapper.desc.clone();
    let (where_sql, args) = wrapper.clone().build(1).unwrap();
    //获取逻辑未删除部分的SQL
    let logic_plugin_sql = get_logic_undelete(&table_name);

    let mut sql = String::new();
    sql.push_str("SELECT ");
    sql.push_str(field_name.as_str());
    sql.push_str(" FROM ");
    sql.push_str(table_name.as_str());

    let mut had_where = false;
    // if args.len() != 0 {
    //     sql.push_str(" WHERE ");
    //     sql.push_str(where_sql.as_str());
    //     had_where = true;
    // }
    if !where_sql.is_empty() {
        sql.push_str(" WHERE ");
        sql.push_str(where_sql.as_str());
        had_where = true;
    }

    if !logic_plugin_sql.is_empty() {
        if had_where {
            sql.push_str(" AND ");
        } else {
            sql.push_str(" WHERE ");
        }
        sql.push_str(logic_plugin_sql.as_str());
    }

    // if order_by !== None{
    //     let desc_str = if desc {"desc"} else {"asc"};
    //     let order_by_format = format!("order by {} {} " ,order_by.clone(),desc_str);
    let order_by_format = match wrapper.get_order_by() {
        Some(t) => t,
        None => " ".to_string(),
    };
    sql.push_str(&order_by_format);

    //跟踪日志
    trace_exec_log(&sql, &args);
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    let statement = client.prepare(sql.as_str()).await.unwrap();
    let result = client.query(&statement, &args).await?;
    //  println!("result:{:?}", result);

    // let result_t = T::return_list(result);
    let mut new_row_hashmap = Vec::new();

    for row in result {
        new_row_hashmap.push(unit::pg_value_to_json_value(&row));
    }

    return Ok(new_row_hashmap);
}

///fetch_one<T>(wrapper) ->Result<T, Error>         单表单条数据
pub async fn fetch_one<T>(
    wrapper: Wrapper<'_>,
    prefix: Option<String>,
    suffix: Option<String>,
) -> Result<T, Error>
where
    T: Parameters,
{
    //let field_name = T::get_field_list();
    let field_name = wrapper.clone().get_recoder_field::<T>();
    let table_name = T::get_table_name(prefix, suffix);

    let (where_sql, args) = wrapper.build(1).unwrap();

    //获取逻辑未删除部分的SQL
    let logic_plugin_sql = get_logic_undelete(&table_name);

    let mut sql = String::new();
    sql.push_str("SELECT ");
    sql.push_str(field_name.as_str());
    sql.push_str(" FROM ");
    sql.push_str(table_name.as_str());

    let mut had_where = false;
    // if args.len() != 0 {
    //     sql.push_str(" WHERE ");
    //     sql.push_str(where_sql.as_str());
    //     had_where = true;
    // }
    if !where_sql.is_empty() {
        sql.push_str(" WHERE ");
        sql.push_str(where_sql.as_str());
        had_where = true;
    }

    if !logic_plugin_sql.is_empty() {
        if had_where {
            sql.push_str(" AND ");
        } else {
            sql.push_str(" WHERE ");
        }
        sql.push_str(logic_plugin_sql.as_str());
    }
    //跟踪日志
    trace_exec_log(&sql, &args);
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    if client.is_closed() {}
    let statement = client.prepare(sql.as_str()).await.unwrap();
    let result = client.query_one(&statement, &args).await;
    // println!("result:{:?}", result);
    match result {
        Ok(row) => {
            let result_t = T::return_one(row);
            return Ok(result_t);
        }
        Err(e) => {
            // let error = format!("{}", e);
            return Err(e);
        }
    }
}

//fetch_page<T>(wrapper) ->Result<PageT<>, Error>     单表翻页
pub async fn fetch_page<T>(
    wrapper: Wrapper<'_>,
    prefix: Option<String>,
    suffix: Option<String>,
) -> Result<PageT<T>, Error>
where
    T: Parameters,
{
    // let wrapper = &wrapper;
    //验证入参数
    let (_limit_str, _limit, page_no, page_size) = wrapper.clone().get_page_info();

    let recoder_field = wrapper.clone().get_recoder_field::<T>();
    // let field_name = T::get_field_list();
    let table_name = T::get_table_name(prefix, suffix);

    let (where_sql, args) = wrapper.clone().build(1).unwrap();

    //获取逻辑未删除部分的SQL
    let logic_plugin_sql = get_logic_undelete(&table_name);

    let mut sql = String::new();
    sql.push_str("SELECT ");
    sql.push_str(recoder_field.as_str());
    sql.push_str(" FROM ");
    sql.push_str(table_name.as_str());

    let mut had_where = false;
    // if args.len() != 0 {
    //     sql.push_str(" WHERE ");
    //     sql.push_str(where_sql.as_str());
    //     had_where = true;
    // }
    if !where_sql.is_empty() {
        sql.push_str(" WHERE ");
        sql.push_str(where_sql.as_str());
        had_where = true;
    }

    //增加逻辑删除
    if !logic_plugin_sql.is_empty() {
        if had_where {
            sql.push_str(" AND ");
        } else {
            sql.push_str(" WHERE ");
        }
        sql.push_str(logic_plugin_sql.as_str());
    }

    let re_query_page_total = query_page_total(&sql, &args).await;
    debug!("re_query_page_total: {:?}", re_query_page_total);
    let total = match re_query_page_total {
        Ok(total) => total,
        Err(err_str) => return Err(err_str),
    };
    if total == 0 {
        let result = PageT {
            // data
            records: Vec::new(),
            // total num
            total: 0,
            // pages
            pages: 0,
            // current page index
            page_no,
            // default 10
            page_size,
        };

        return Ok(result);
    }

    //增加是否排序
    match wrapper.clone().get_order_by() {
        Some(order_by) => {
            sql.push_str(order_by.as_str());
        }
        None => (),
    };

    //增加翻页
    let (limit_str, _limit, page_no, page_size) = wrapper.clone().get_page_info();
    let mut pages = total / page_size;
    let duoyu = total % page_size;
    if duoyu > 0 {
        pages = pages + 1
    }

    sql.push_str(limit_str.as_str());

    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    let statement = client.prepare(sql.as_str()).await.unwrap();
    let result = client.query(&statement, &args).await;

    let result = match result {
        Ok(t) => t,
        Err(e) => {
            // let error_str = format!("{}", e);
            return Err(e);
        }
    };

    let records = T::return_list(result);

    let result = PageT {
        // data
        records,
        // total num
        total,
        // pages
        pages,
        // current page index
        page_no,
        // default 10
        page_size,
    };

    return Ok(result);
}

//============================================================================================
//不需要翻页的查询
pub async fn query(
    sql: &str,
    params: &[&(dyn ToSql + Sync)],
) -> Result<Vec<HashMap<String, Value>>, Error> {
    //跟踪日志
    trace_exec_log(sql, &params);
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    let statement = client.prepare(&sql).await.unwrap();
    let opt_result = client.query(&statement, params).await;
    let mut new_row_hashmap = Vec::new();
    match opt_result {
        Ok(result) => {
            for row in result {
                new_row_hashmap.push(unit::pg_value_to_json_value(&row));
            }
        }
        Err(e) => {
            return Err(e);
        }
    }
    return Ok(new_row_hashmap);
}

//返回结构体的方式 排序由此决定
pub async fn query_t<T>(sql: &str, params: &[&(dyn ToSql + Sync)]) -> Result<Vec<T>, Error>
where
    T: Parameters,
{
    //跟踪日志
    trace_exec_log(&sql, &params);
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    let statement = client.prepare(&sql).await.unwrap();
    let result = client.query(&statement, params).await;

    let result = match result {
        Ok(t) => t,
        Err(e) => {
            //let error_str = format!("{}", e);
            return Err(e);
        }
    };
    let records = T::return_list(result);

    return Ok(records);
}

pub async fn query_one(
    sql: &str,
    params: &[&(dyn ToSql + Sync)],
) -> Result<HashMap<String, Value>, String> {
    trace_exec_log(&sql, &params);
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    let statement = client.prepare(&sql).await.unwrap();
    let opt_result = client.query_one(&statement, params).await;
    // let mut new_row_hashmap = Vec::new();
    match opt_result {
        Ok(result) => {
            return Ok(unit::pg_value_to_json_value(&result));
        }
        Err(e) => {
            let err_str = format!("{}", e);
            return Err(err_str);
        }
    }
}

//查询行数
//原理就是找出第一个 from 然后截取后面的条件。前面拼接上select count(*) from ......
async fn query_page_total(sql: &str, params: &[&(dyn ToSql + Sync)]) -> Result<u64, Error> {
    let mut opt_index = sql.find(" from ");
    debug!("opt_index1: {:?}",opt_index);
    opt_index =  if opt_index.is_none() {
        sql.find(" FROM ")
    }else{
        opt_index
    };
    let index = opt_index.take().unwrap();
    

    //然后进行拼接
    let sql_from = &sql[index..sql.len()];
    let total_sql = format!("select count(*) as total {}", sql_from);
    //跟踪日志
    trace_exec_log(&sql, params);
    //执行
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    let statement = client.prepare(&total_sql).await.unwrap();
    let opt_result = client.query_one(&statement, params).await;
    match opt_result {
        Ok(result) => {
            let total: i64 = result.get("total");
            return Ok(total as u64);
        }
        Err(e) => {
            // let err_str = format!("{}", e);
            return Err(e);
        }
    }
}

pub async fn query_page(
    sql: &str,
    params: &[&(dyn ToSql + Sync)],
    order_by: &str,
    desc: bool,
    page_no: u64,
    page_size: u64,
) -> Result<PageHash, Error> {
    //验证入参
    // if page_size <= 0 {return Err("page_size 必须大于0".to_string())};

    let re_query_page_total = query_page_total(&sql, params).await;

    let total = match re_query_page_total {
        Ok(total) => total,
        Err(err_str) => return Err(err_str),
    };

    //一共多少页
    let pages = total / page_size;

    let mut new_row_hashmap = Vec::new();
    //如果等于0 就是没有数据 直接返回
    if total == 0 {
        let page_data = PageHash {
            records: new_row_hashmap,
            total: total,
            pages,
            page_no,
            page_size,
        };
        return Ok(page_data);
    }

    //生成查询语句 拼接排序，页数，每页的数据条数
    let b_order_desc = if desc { "DESC" } else { "ASC" };
    //let order_str = order.join(",");
    //LIMIT [no of rows] OFFSET [row num]
    let offset = (page_no - 1) * page_size;

    let row_sql = format!(
        "{} ORDER BY  {} {} LIMIT {} OFFSET {};",
        sql, order_by, b_order_desc, page_size, offset
    );
    trace_exec_log(&row_sql, params);
    trace_exec_log(&sql, &params);
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    let statement = client.prepare(&row_sql).await.unwrap();
    let opt_result = client.query(&statement, params).await;

    match opt_result {
        Ok(result) => {
            for row in result {
                let hashmap = unit::pg_value_to_json_value(&row);
                new_row_hashmap.push(hashmap);
            }
        }
        Err(e) => {
            // let err_str = format!("{}", e);
            return Err(e);
        }
    }

    let page_data = PageHash {
        records: new_row_hashmap,
        total: total,
        pages,
        page_no,
        page_size,
    };
    // return Ok(new_row_hashmap);
    Ok(page_data)
}

pub async fn check_row_by_column<C, T>(
    table_name: &str,
    column: &C,
    value: T,
) -> Result<bool, Error>
where
    C: ColumnExt,
    T: ToSql + Sync,
{
    let column_name = column.get();
    let mut sql = format!(
        "SELECT \"{}\" FROM \"{}\" WHERE \"{}\" = $1",
        column_name, table_name, column_name
    );

    //获取逻辑未删除部分的SQL
    let logic_plugin_sql = get_logic_undelete(&table_name);

    if !logic_plugin_sql.is_empty() {
        sql.push_str(" AND ");
        sql.push_str(logic_plugin_sql.as_str());
    }
    debug!("check_row_by_column sql :{}", sql);
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    // let result = client.query_opt(sql.as_str(), &[&value]).await;
    // match result {
    //     Ok(opt_row) => match opt_row {
    //         Some(_row) => return Ok(true),
    //         None => return Ok(false),
    //     },
    //     Err(e) => {
    //        // let err_str = format!("{}", e);
    //         error!("check_row_by_column:{}",e);
    //         return Err(e);
    //     }
    // }
    let result = client.query(sql.as_str(), &[&value]).await;
    match result {
        Ok(opt_row) => {
            if opt_row.len() > 0 {
                return Ok(true);
            } else {
                return Ok(false);
            }
        }
        Err(e) => {
            // let err_str = format!("{}", e);
            error!("check_row_by_column:{}", e);
            return Err(e);
        }
    }
}

pub async fn check_row_wrap<T>(
    wrapper: Wrapper<'_>,
    prefix: Option<String>,
    suffix: Option<String>,
) -> Result<bool, Error>
where
    T: Parameters,
{
    let table_name = T::get_table_name(prefix, suffix);
    let (where_sql, args) = wrapper.clone().build(1).unwrap();

    //获取逻辑未删除部分的SQL
    let logic_plugin_sql = get_logic_undelete(&table_name);

    let mut sql = String::new();
    sql.push_str("SELECT ");
    sql.push_str(" 1 ");
    sql.push_str(" FROM ");
    sql.push_str(table_name.as_str());

    let mut had_where = false;
    if args.len() != 0 {
        sql.push_str(" WHERE ");
        sql.push_str(where_sql.as_str());
        had_where = true;
    }

    //增加逻辑删除
    if !logic_plugin_sql.is_empty() {
        if had_where {
            sql.push_str(" AND ");
        } else {
            sql.push_str(" WHERE ");
        }
        sql.push_str(logic_plugin_sql.as_str());
    }
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let client = pools.get().await.unwrap();
    let statement = client.prepare(sql.as_str()).await.unwrap();
    let result = client.query(&statement, &args).await;
    match result {
        Ok(opt_row) => {
            if opt_row.len() == 0 {
                return Ok(false);
            } else {
                return Ok(true);
            }
            // Some(_row) => return Ok(true),
            // None => return Ok(false),
        }
        Err(e) => {
            // let err_str = format!("{}", e);
            return Err(e);
        }
    }
}

fn get_logic_undelete(table_name: &str) -> String {
    let logic_plugin_sql = if DATABASE_POOL.logic_plugin.is_some() {
        // 判断表名是否存在排除表中， 如果不存在，sql变成逻辑删除的方式
        if !DATABASE_POOL
            .logic_plugin
            .as_ref()
            .unwrap()
            .is_exclusion_table(table_name)
        {
            format!(
                " \"{}\" = {} ",
                DATABASE_POOL.logic_plugin.as_ref().unwrap().column(),
                DATABASE_POOL.logic_plugin.as_ref().unwrap().un_deleted(),
            )
        } else {
            EMPTY_STRING.to_string()
        }
    } else {
        EMPTY_STRING.to_string()
    };
    logic_plugin_sql
}
/// 字符串左右新增百分号
pub fn format_like(obj: &Option<String>) -> Option<String> {
    match obj {
        Some(t) => {
            let two = format!("%{}%", t);
            Some(two)
        }
        None => None,
    }
}
/// 字符串左新增百分号
pub fn format_like_left(obj: &Option<String>) -> Option<String> {
    match obj {
        Some(t) => {
            let two = format!("%{}", t);
            Some(two)
        }
        None => None,
    }
}
/// 字符串左新增百分号
pub fn format_like_right(obj: &Option<String>) -> Option<String> {
    match obj {
        Some(t) => {
            let two = format!("{}%", t);
            Some(two)
        }
        None => None,
    }
}

///事务新增 用于按指定ID批量删除，然后批量新增
pub async fn transaction_delete_and_save<T, V>(
    delete_column: &'static str,
    delete_value: &V,
    p: Vec<T>,
    prefix: Option<String>,
    suffix: Option<String>,
) -> Result<u64, Error>
where
    T: Parameters,
    V: ToSql + Sync,
{
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let mut client = pools.get().await.unwrap();
    let transaction = client.transaction().await.unwrap();
    //delete_columnlet column_name = delete_column.get();
    let table_name = T::get_table_name(prefix.clone(), suffix.clone());
    let sql = format!(
        "DELETE FROM \"{}\" where \"{}\" = $1",
        &table_name, delete_column
    );
    //执行删除
    let statement = transaction.prepare(sql.as_str()).await.unwrap();
    match transaction.execute(&statement, &[delete_value]).await {
        Ok(_) => {
            debug!("删除执行成功");
        }
        Err(e) => {
            error!("删除执行失败: {}", e);
            transaction.rollback().await.unwrap();
            return Err(e);
        }
    }

    for item in p.iter() {
        let (sql, params) = item.gen_save(prefix.clone(), suffix.clone()).unwrap();
        let statement = transaction.prepare(sql.as_str()).await.unwrap();
        let result = transaction.execute(&statement, &params).await;
        match result {
            Ok(_t) => {
                trace!("SAVE 执行成功");
                //return Ok(t);
            }
            Err(e) => {
                transaction.rollback().await.unwrap();
                error!("{:?}", e);
                return Err(e);
            }
        }
    }
    let result = transaction.commit().await;

    if result.is_err() {
        error!(" transaction.commit() 执行 提交失败！");
        //transaction.rollback().await.unwrap();
    }
    return Ok(0);
}

#[derive(Clone, Debug)]
pub struct TransactionBatchParam<'a> {
    pub statement_sql: &'a str,
    pub params: Vec<&'a (dyn ToSql + Sync)>,
}

///事务 批量更新/新增 ，不同的表
pub async fn transaction_batch(param: Vec<TransactionBatchParam<'_>>) -> Result<u64, String> {
    let pools: &Pool = DATABASE_POOL.pools.get().unwrap();
    let mut client = pools.get().await.unwrap();
    let transaction = client.transaction().await.unwrap();
    //  transaction.execute(statement, params);
    for item in param.iter() {
        let statement = transaction.prepare(item.statement_sql).await.unwrap();
        match transaction.execute(&statement, &item.params).await {
            Ok(t) => {
                //如果执行没有相应任何一行，表示执行的SQL 没有意义 直接回滚??
                if t == 0 {
                    if !item.statement_sql.find("DELETE").is_some(){
                        transaction.rollback().await.unwrap();
                        error!("执行的SQL没有任何意义,响应行数为0");
                        return Err("执行的SQL没有任何意义,响应行数为0".to_string());
                    }
                }
            }
            Err(e) => {
                //执行回滚
                transaction.rollback().await.unwrap();
                error!("{:?}", e);
                return Err(e.to_string());
            }
        }
    }
    let result = transaction.commit().await;

    if result.is_err() {
        error!(" transaction.commit() 执行 提交失败！");
        //transaction.rollback().await.unwrap();
    }

    return Ok(0);
}