use std::collections::HashMap;
use crate::config::Connection;
use crate::types::{DbMode, Mode, Params, TableOptions};
use crate::pools;
use chrono::Local;
use json::{array, object, JsonValue};
use lazy_static::lazy_static;
use log::{error, info};
use std::sync::Mutex;
use std::sync::Arc;
use std::thread;
use br_pgsql::connect::Connect;
use br_pgsql::pools::Pools;

lazy_static! {
   static ref TR: Arc<Mutex<HashMap<String, Arc<Mutex<Connect>>>>> = Arc::new(Mutex::new(HashMap::new()));
   static ref TRANS: Arc<Mutex<HashMap<String, i32>>> = Arc::new(Mutex::new(HashMap::new()));
}
#[derive(Clone)]
pub struct Pgsql {
    /// 当前连接配置
    pub connection: Connection,
    /// 当前选中配置
    pub default: String,
    pub params: Params,
    pub client: Pools,
}

impl Pgsql {
    pub fn connect(connection: Connection, default: String) -> Result<Self, String> {
        let port = connection.hostport.parse::<i32>().map_err(|e| format!("parse hostport to i32 err: {e:?}"))?;

        let cp_connection = connection.clone();
        let config = object! {
            debug: cp_connection.debug,
            username: cp_connection.username,
            userpass: cp_connection.userpass,
            database: cp_connection.database,
            hostname: cp_connection.hostname,
            hostport: port,
            charset: cp_connection.charset.str(),
        };
        let mut pgsql = br_pgsql::Pgsql::new(&config)?;

        let pools = pgsql.pools()?;
        Ok(Self {
            connection,
            default: default.clone(),
            params: Params::default("pgsql"),
            client: pools,
        })
    }

    fn query(&mut self, sql: &str) -> (bool, JsonValue) {
        let thread_id = format!("{:?}", thread::current().id());
        // let _key = format!("{}{}", self.default, thread_id);

        // === 事务环境 ===
        if TRANS.lock().unwrap().get(&*thread_id).is_some() {
            // 事务环境下的代码保持不变
            let key = format!("{}{}", self.default, thread_id);
            let db = TR.lock().unwrap().get_mut(&*key).unwrap().clone();

            let mut t = db.lock().unwrap();
            match t.query(sql) {
                Ok(e) => {
                    if self.connection.debug {
                        info!("查询成功: {} {}", thread_id.clone(), sql);
                    }
                    (true, e.rows)
                }
                Err(e) => {
                    error!("事务查询失败: 线程ID: {thread_id} 错误: {e} SQL语句: [{sql}]");
                    (false, JsonValue::from(e.to_string()))
                }
            }
        } else {
            // 非事务环境下使用 ConnectionGuard
            let mut guard = match self.client.get_guard() {
                Ok(g) => g,
                Err(e) => {
                    error!("非事务查询失败: get_guard 线程ID: {thread_id} 错误: {e} SQL语句: [{sql}]");
                    return (false, JsonValue::from(e.to_string()));
                }
            };

            let res = guard.conn().query(sql);
            match res {
                Ok(e) => {
                    if self.connection.debug {
                        info!("查询成功: {} {}", thread_id.clone(), sql);
                    }
                    (true, e.rows)
                }
                Err(e) => {
                    error!("非事务查询失败: 线程ID: {thread_id} 错误: {e} SQL语句: [{sql}]");
                    (false, JsonValue::from(e.to_string()))
                }
            }
            // guard 离开作用域时自动归还连接
        }
    }
    fn execute(&mut self, sql: &str) -> (bool, JsonValue) {
        let thread_id = format!("{:?}", thread::current().id());

        if TRANS.lock().unwrap().get(&*thread_id).is_some() {
            let key = format!("{}{}", self.default, thread_id);
            let db = TR.lock().unwrap().get_mut(&*key).unwrap().clone();
            let mut t = db.lock().unwrap();
            match t.execute(sql) {
                Ok(e) => {
                    if self.connection.debug {
                        info!("提交成功: {} {}", thread_id.clone(), sql);
                    }
                    if sql.contains("INSERT") {
                        (true, e.rows)
                    } else {
                        (true, e.affect_count.into())
                    }
                }
                Err(e) => {
                    error!("事务提交失败: {thread_id} {e} {sql}");
                    (false, JsonValue::from(e.to_string()))
                }
            }
        } else {
            // 非事务环境下使用 ConnectionGuard
            let mut guard = match self.client.get_guard() {
                Ok(g) => g,
                Err(e) => {
                    error!("非事务执行失败: get_guard 线程ID: {thread_id} 错误: {e} SQL语句: [{sql}]");
                    return (false, JsonValue::from(e.to_string()));
                }
            };

            let res = guard.conn().execute(sql);
            match res {
                Ok(e) => {
                    if self.connection.debug {
                        info!("提交成功: {} {}", thread_id.clone(), sql);
                    }
                    if sql.contains("INSERT") {
                        (true, e.rows)
                    } else {
                        (true, e.affect_count.into())
                    }
                }
                Err(e) => {
                    error!("非事务提交失败: {thread_id} {e} {sql}");
                    (false, JsonValue::from(e.to_string()))
                }
            }
            // guard 离开作用域时自动归还连接
        }
    }
}

impl DbMode for Pgsql {
    fn database_tables(&mut self) -> JsonValue {
        let sql = "SHOW TABLES".to_string();
        match self.sql(sql.as_str()) {
            Ok(e) => {
                let mut list = vec![];
                for item in e.members() {
                    for (_, value) in item.entries() {
                        list.push(value.clone());
                    }
                }
                list.into()
            }
            Err(_) => {
                array![]
            }
        }
    }

    fn database_create(&mut self, name: &str) -> bool {
        let sql = format!("CREATE DATABASE {name}");

        let (state, data) = self.execute(sql.as_str());
        match state {
            true => data.as_bool().unwrap(),
            false => {
                error!("创建数据库失败: {data:?}");
                false
            }
        }
    }
}

impl Mode for Pgsql {
    fn table_create(&mut self, mut options: TableOptions) -> JsonValue {
        let mut sql = String::new();
        let mut comments = vec![];

        // 唯一约束
        if !options.table_unique.is_empty() {
            let unique = format!(
                "CREATE UNIQUE INDEX {01}_unique_{} ON {} ({});",
                options.table_unique.join("_"),
                options.table_name,
                options.table_unique.join(",")
            );
            comments.push(unique);
        }

        // 唯一索引
        for row in options.table_index.iter() {
            let index = format!(
                "CREATE INDEX {01}_index_{} ON {} ({})",
                row.join("_"),
                options.table_name,
                row.join(",")
            );
            comments.push(index);
        }

        for (name, field) in options.table_fields.entries_mut() {
            field["table_name"] = options.table_name.clone().into();
            let row = br_fields::field("pgsql", name, field.clone());
            let rows = row.split("comment").collect::<Vec<&str>>();
            comments.push(format!(
                "COMMENT ON COLUMN {}.{} IS {};",
                options.table_name, name, rows[1]
            ));
            sql = format!("{} {},\r\n", sql, rows[0]);
        }

        let primary_key = format!(
            "CONSTRAINT {}_{} PRIMARY KEY ({})",
            options.table_name, options.table_key, options.table_key
        );
        let sql = format!(
            "CREATE TABLE IF NOT EXISTS {} (\r\n{},\r\n{}\r\n);\r\n",
            options.table_name,
            sql.trim_end_matches(",\r\n"),
            primary_key
        );
        comments.insert(0, sql);

        for (_name, field) in options.table_fields.entries() {
            field["mode"].as_str().unwrap();
            {}
        }

        if self.params.sql {
            let info = comments.join("\r\n");
            return JsonValue::from(info);
        }
        for comment in comments {
            let (state, _) = self.execute(comment.as_str());
            match state {
                true => {}
                false => {
                    return JsonValue::from(state);
                }
            }
        }
        JsonValue::from(true)
    }

    fn table_update(&mut self, options: TableOptions) -> JsonValue {
        let fields_list = self.table_info(&options.table_name);
        let mut put = vec![];
        let mut add = vec![];
        let mut del = vec![];
        let mut comments = vec![];

        for (key, _) in fields_list.entries() {
            if options.table_fields[key].is_empty() {
                del.push(key);
            }
        }
        for (name, field) in options.table_fields.entries() {
            if !fields_list[name].is_empty() {
                let old_comment = fields_list[name]["comment"].to_string();
                let new_comment = br_fields::field("pgsql", name, field.clone());
                let new_comment: Vec<&str> = new_comment.split(" comment ").collect();
                let new_comment_text = new_comment[1].trim().trim_start_matches("'").trim_end_matches("'");
                if old_comment == new_comment_text {
                    continue;
                }
                put.push(name);
            } else {
                add.push(name);
            }
        }

        for name in add.iter() {
            let name = name.to_string();
            let row = br_fields::field("pgsql", &name, options.table_fields[name.as_str()].clone());
            let rows = row.split("comment").collect::<Vec<&str>>();
            comments.push(format!(
                r#"ALTER TABLE "{}" add {};"#,
                options.table_name, rows[0]
            ));
            comments.push(format!(
                "COMMENT ON COLUMN {}.{} IS {};",
                options.table_name, name, rows[1]
            ));
        }
        for name in del.iter() {
            comments.push(format!(
                "ALTER TABLE {} DROP {};\r\n",
                options.table_name, name
            ));
        }
        for name in put.iter() {
            let name = name.to_string();
            let row = br_fields::field("pgsql", &name, options.table_fields[name.as_str()].clone());
            let rows = row.split("comment").collect::<Vec<&str>>();

            let sql = rows[0].split(" ").collect::<Vec<&str>>();

            if sql[1].contains("BOOLEAN") {
                let text = format!(
                    "ALTER TABLE {} ALTER COLUMN \"{}\" DROP DEFAULT;\r\n",
                    options.table_name, name
                );
                comments.push(text.clone());
                let text = format!(
                    "ALTER TABLE {} ALTER COLUMN \"{}\" TYPE {} USING {1}::boolean;\r\n",
                    options.table_name, name, sql[1]
                );
                comments.push(text.clone());
            } else {
                let text = format!(
                    "ALTER TABLE {} ALTER COLUMN \"{}\" TYPE {};\r\n",
                    options.table_name, name, sql[1]
                );
                comments.push(text.clone());
            };

            if sql.len() > 3 && !sql[3].is_empty() {
                let text = format!(
                    "ALTER TABLE {} ALTER COLUMN {} SET DEFAULT '{}';\r\n",
                    options.table_name, name, sql[3]
                );
                comments.push(text.clone());
            }

            comments.push(format!(
                "COMMENT ON COLUMN {}.{} IS {};",
                options.table_name, name, rows[1]
            ));
        }

        let mut unique_new = vec![];
        let mut index_new = vec![];
        let mut primary_key = vec![];
        let (_, index_list) = self.query(
            format!(
                "SELECT * FROM pg_indexes WHERE tablename = '{}'",
                options.table_name
            ).as_str(),
        );
        for item in index_list.members() {
            let key_name = item["indexname"].as_str().unwrap();
            let indexdef = item["indexdef"].to_string();

            if indexdef.contains(
                format!(
                    "CREATE UNIQUE INDEX {}_{} ON",
                    options.table_name, options.table_key
                ).as_str(),
            ) {
                primary_key.push(key_name.to_string());
                continue;
            }
            if indexdef.contains("CREATE UNIQUE INDEX") {
                unique_new.push(key_name.to_string());
                continue;
            }
            if indexdef.contains("CREATE INDEX") {
                index_new.push(key_name.to_string());
                continue;
            }
        }

        if !options.table_unique.is_empty() {
            let name = format!(
                "{}_unique_{}",
                options.table_name,
                options.table_unique.join("_")
            );
            let unique = format!(
                "CREATE UNIQUE INDEX {} ON {} ({});",
                name,
                options.table_name,
                options.table_unique.join(",")
            );
            if !unique_new.contains(&name) {
                comments.push(unique);
            } else {
                unique_new.retain(|x| *x != name);
            }
        }

        // 唯一索引
        for row in options.table_index.iter() {
            let name = format!("{}_index_{}", options.table_name, row.join("_"));
            let index = format!(
                "CREATE INDEX {} ON {} ({})",
                name,
                options.table_name,
                row.join(",")
            );
            if !index_new.contains(&name) {
                comments.push(index);
            } else {
                index_new.retain(|x| *x != name);
            }
        }

        for item in unique_new {
            comments.push(format!("DROP INDEX {};\r\n", item.clone()));
        }
        for item in index_new {
            comments.push(format!("DROP INDEX {};\r\n", item.clone()));
        }

        if self.params.sql {
            return JsonValue::from(comments.join(""));
        }

        if comments.is_empty() {
            return JsonValue::from(-1);
        }

        for item in comments.iter() {
            let (state, res) = self.execute(item.as_str());
            match state {
                true => {}
                false => {
                    error!("{} 更新失败: {} \r\n {}", options.table_name, item, res);
                    return JsonValue::from(0);
                }
            }
        }
        JsonValue::from(1)
    }

    fn table_info(&mut self, table: &str) -> JsonValue {
        let sql = format!(
            "SELECT  COL.COLUMN_NAME,
    COL.DATA_TYPE,
    COALESCE(DESCRIPTION.DESCRIPTION, '') AS COMMENT FROM INFORMATION_SCHEMA.COLUMNS COL
    LEFT JOIN
    pg_catalog.pg_description DESCRIPTION
    ON DESCRIPTION.objsubid = COL.ORDINAL_POSITION
    AND DESCRIPTION.objoid = (SELECT oid FROM pg_catalog.pg_class WHERE relname = COL.TABLE_NAME LIMIT 1) WHERE  COL.TABLE_NAME = '{table}'");
        let (state, data) = self.query(sql.as_str());
        let mut list = object! {};
        if state {
            for item in data.members() {
                let mut row = object! {};
                row["field"] = item["column_name"].clone();
                row["comment"] = item["comment"].clone();
                row["type"] = item["data_type"].clone();
                list[row["field"].as_str().unwrap()] = row.clone();
            }
            list
        } else {
            list
        }
    }

    fn table_is_exist(&mut self, name: &str) -> bool {
        let sql = format!("SELECT EXISTS (SELECT 1  FROM information_schema.tables   WHERE table_schema = 'public'  AND table_name = '{name}')");
        let (state, data) = self.query(sql.as_str());
        match state {
            true => {
                for item in data.members() {
                    if item.has_key("exists") {
                        return item["exists"].as_bool().unwrap();
                    }
                }
                false
            }
            false => false,
        }
    }

    fn table(&mut self, name: &str) -> &mut Pgsql {
        self.params = Params::default(self.connection.mode.str().as_str());
        self.params.table = format!("{}{}", self.connection.prefix, name);
        self.params.join_table = self.params.table.clone();
        self
    }

    fn change_table(&mut self, name: &str) -> &mut Self {
        self.params.join_table = name.to_string();
        self
    }

    fn autoinc(&mut self) -> &mut Self {
        self.params.autoinc = true;
        self
    }

    fn fetch_sql(&mut self) -> &mut Self {
        self.params.sql = true;
        self
    }

    fn order(&mut self, field: &str, by: bool) -> &mut Self {
        self.params.order[field] = {
            if by {
                "DESC"
            } else {
                "ASC"
            }
        }.into();
        self
    }

    fn group(&mut self, field: &str) -> &mut Self {
        let fields: Vec<&str> = field.split(",").collect();
        for field in fields.iter() {
            let field = field.to_string();
            self.params.group[field.as_str()] = field.clone().into();
            self.params.fields[field.as_str()] = field.clone().into();
        }
        self
    }

    fn distinct(&mut self) -> &mut Self {
        self.params.distinct = true;
        self
    }

    fn json(&mut self, field: &str) -> &mut Self {
        let list: Vec<&str> = field.split(",").collect();
        for item in list.iter() {
            self.params.json[item.to_string().as_str()] = item.to_string().into();
        }
        self
    }

    fn location(&mut self, field: &str) -> &mut Self {
        let list: Vec<&str> = field.split(",").collect();
        for item in list.iter() {
            self.params.location[item.to_string().as_str()] = item.to_string().into();
        }
        self
    }

    fn field(&mut self, field: &str) -> &mut Self {
        let list: Vec<&str> = field.split(",").collect();
        let join_table = if self.params.join_table.is_empty() {
            self.params.table.clone()
        } else {
            self.params.join_table.clone()
        };
        for item in list.iter() {
            if item.contains(" as ") {
                let text = item.split(" as ").collect::<Vec<&str>>();
                if text[0].contains("count(") {
                    self.params.fields[item.to_string().as_str()] = format!("{} as {}", text[0], text[1]).into();
                } else {
                    self.params.fields[item.to_string().as_str()] = format!("{}.{} as {}", join_table, text[0], text[1]).into();
                }
            } else {
                self.params.fields[item.to_string().as_str()] = format!("{join_table}.{item}").into();
            }
        }
        self
    }

    fn hidden(&mut self, name: &str) -> &mut Self {
        let hidden: Vec<&str> = name.split(",").collect();

        let fields_list = self.table_info(self.params.clone().table.as_str());
        let mut data = array![];
        for item in fields_list.members() {
            data.push(object! {
                "name":item["field"].as_str().unwrap()
            }).unwrap();
        }

        for item in data.members() {
            let name = item["name"].as_str().unwrap();
            if !hidden.contains(&name) {
                self.params.fields[name] = name.into();
            }
        }
        self
    }

    fn where_and(&mut self, field: &str, compare: &str, mut value: JsonValue) -> &mut Self {
        let join_table = if self.params.join_table.is_empty() {
            self.params.table.clone()
        } else {
            self.params.join_table.clone()
        };
        if value.is_boolean() {
            if value.as_bool().unwrap() {
                value = 1.into();
            } else {
                value = 0.into();
            }
        }
        match compare {
            "between" => {
                self.params.where_and.push(format!(
                    "{}.{} between '{}' AND '{}'",
                    join_table, field, value[0], value[1]
                ));
            }
            "set" => {
                let list: Vec<&str> = value.as_str().unwrap().split(",").collect();
                let mut wheredata = vec![];
                for item in list.iter() {
                    wheredata.push(format!("'{item}' = ANY (string_to_array({join_table}.{field},','))"));
                }
                self.params.where_and.push(format!("({})", wheredata.join(" or ")));
            }
            "notin" => {
                let mut text = String::new();
                for item in value.members() {
                    text = format!("{text},'{item}'");
                }
                text = text.trim_start_matches(",").into();
                self.params.where_and.push(format!("{join_table}.{field} not in ({text})"));
            }
            "is" => {
                self.params.where_and.push(format!("{join_table}.{field} is {value}"));
            }
            "isnot" => {
                self.params.where_and.push(format!("{join_table}.{field} is not {value}"));
            }
            "notlike" => {
                self.params.where_and.push(format!("{join_table}.{field} not like '{value}'"));
            }
            "in" => {
                let mut text = String::new();
                if value.is_array() {
                    for item in value.members() {
                        text = format!("{text},'{item}'");
                    }
                } else if value.is_null() {
                    text = format!("{text},null");
                } else {
                    let value = value.as_str().unwrap();

                    let value: Vec<&str> = value.split(",").collect();
                    for item in value.iter() {
                        text = format!("{text},'{item}'");
                    }
                }
                text = text.trim_start_matches(",").into();

                self.params.where_and.push(format!("{join_table}.{field} {compare} ({text})"));
            }
            _ => {
                self.params.where_and.push(format!("{join_table}.{field} {compare} '{value}'"));
            }
        }
        self
    }

    fn where_or(&mut self, field: &str, compare: &str, mut value: JsonValue) -> &mut Self {
        let join_table = if self.params.join_table.is_empty() {
            self.params.table.clone()
        } else {
            self.params.join_table.clone()
        };

        if value.is_boolean() {
            if value.as_bool().unwrap() {
                value = 1.into();
            } else {
                value = 0.into();
            }
        }

        match compare {
            "between" => {
                self.params.where_or.push(format!(
                    "{}.{} between '{}' AND '{}'",
                    join_table, field, value[0], value[1]
                ));
            }
            "set" => {
                let list: Vec<&str> = value.as_str().unwrap().split(",").collect();
                let mut wheredata = vec![];
                for item in list.iter() {
                    wheredata.push(format!("'{item}' = ANY (string_to_array({join_table}.{field},','))"));
                }
                self.params.where_or.push(format!("({})", wheredata.join(" or ")));
            }
            "notin" => {
                let mut text = String::new();
                for item in value.members() {
                    text = format!("{text},'{item}'");
                }
                text = text.trim_start_matches(",").into();
                self.params.where_or.push(format!("{join_table}.{field} not in ({text})"));
            }
            "is" => {
                self.params.where_or.push(format!("{join_table}.{field} is {value}"));
            }
            "isnot" => {
                self.params.where_or.push(format!("{join_table}.{field} is not {value}"));
            }
            "in" => {
                let mut text = String::new();
                if value.is_array() {
                    for item in value.members() {
                        text = format!("{text},'{item}'");
                    }
                } else {
                    let value = value.as_str().unwrap();
                    let value: Vec<&str> = value.split(",").collect();
                    for item in value.iter() {
                        text = format!("{text},'{item}'");
                    }
                }
                text = text.trim_start_matches(",").into();
                self.params.where_or.push(format!("{join_table}.{field} {compare} ({text})"));
            }
            _ => {
                self.params.where_or.push(format!("{join_table}.{field} {compare} '{value}'"));
            }
        }
        self
    }

    fn where_column(&mut self, field_a: &str, compare: &str, field_b: &str) -> &mut Self {
        self.params.where_column = format!(
            "{}.{} {} {}.{}",
            self.params.table, field_a, compare, self.params.table, field_b
        );
        self
    }

    fn page(&mut self, page: i32, limit: i32) -> &mut Self {
        self.params.page = page;
        self.params.limit = limit;
        self
    }

    fn column(&mut self, field: &str) -> JsonValue {
        self.field(field);
        let sql = self.params.select_sql();

        if self.params.sql {
            return JsonValue::from(sql);
        }
        let (state, data) = self.query(sql.as_str());
        match state {
            true => {
                let mut list = array![];
                for item in data.members() {
                    if self.params.json[field].is_empty() {
                        list.push(item[field].clone()).unwrap();
                    } else {
                        let data = json::parse(item[field].as_str().unwrap()).unwrap_or(array![]);
                        list.push(data).unwrap();
                    }
                }
                list
            }
            false => {
                array![]
            }
        }
    }

    fn count(&mut self) -> JsonValue {
        self.params.fields["count"] = "count(*) as count".to_string().into();
        let sql = self.params.select_sql();
        if self.params.sql {
            return JsonValue::from(sql.clone());
        }
        let (state, data) = self.query(sql.as_str());
        if state {
            data[0]["count"].clone()
        } else {
            JsonValue::from(0)
        }
    }

    fn max(&mut self, field: &str) -> JsonValue {
        self.params.fields[field] = format!("max({field}) as {field}").into();
        let sql = self.params.select_sql();
        if self.params.sql {
            return JsonValue::from(sql.clone());
        }
        let (state, data) = self.query(sql.as_str());
        if state {
            if data.len() > 1 {
                return data.clone();
            }
            data[0][field].clone()
        } else {
            JsonValue::from(0)
        }
    }

    fn min(&mut self, field: &str) -> JsonValue {
        self.params.fields[field] = format!("min({field}) as {field}").into();
        let sql = self.params.select_sql();
        if self.params.sql {
            return JsonValue::from(sql.clone());
        }
        let (state, data) = self.query(sql.as_str());
        if state {
            if data.len() > 1 {
                return data;
            }
            data[0][field].clone()
        } else {
            JsonValue::from(0)
        }
    }

    fn sum(&mut self, field: &str) -> JsonValue {
        self.params.fields[field] = format!("sum({field}) as {field}").into();
        let sql = self.params.select_sql();
        if self.params.sql {
            return JsonValue::from(sql.clone());
        }
        let (state, data) = self.query(sql.as_str());
        match state {
            true => {
                if data.len() > 1 {
                    return data;
                }
                data[0][field].clone()
            }
            false => JsonValue::from(0),
        }
    }

    fn avg(&mut self, field: &str) -> JsonValue {
        self.params.fields[field] = format!("avg({field}) as {field}").into();
        let sql = self.params.select_sql();
        if self.params.sql {
            return JsonValue::from(sql.clone());
        }
        let (state, data) = self.query(sql.as_str());
        if state {
            if data.len() > 1 {
                return data;
            }
            data[0][field].clone()
        } else {
            JsonValue::from(0)
        }
    }

    fn select(&mut self) -> JsonValue {
        let sql = self.params.select_sql();
        if self.params.sql {
            return JsonValue::from(sql.clone());
        }
        let (state, mut data) = self.query(sql.as_str());
        match state {
            true => {
                for (field, _) in self.params.json.entries() {
                    for item in data.members_mut() {
                        if !item[field].is_empty() {
                            let json = item[field].to_string();
                            item[field] = match json::parse(&json) {
                                Ok(e) => e,
                                Err(_) => JsonValue::from(json),
                            };
                        }
                    }
                }
                data.clone()
            }
            false => array![],
        }
    }

    fn find(&mut self) -> JsonValue {
        self.params.page = 1;
        self.params.limit = 1;
        let sql = self.params.select_sql();
        if self.params.sql {
            return JsonValue::from(sql.clone());
        }
        let (state, mut data) = self.query(sql.as_str());
        match state {
            true => {
                if data.is_empty() {
                    return object! {};
                }
                for (field, _) in self.params.json.entries() {
                    if !data[0][field].is_empty() {
                        let json = data[0][field].to_string();
                        let json = json::parse(&json).unwrap_or(array![]);
                        data[0][field] = json;
                    } else {
                        data[0][field] = array![];
                    }
                }
                data[0].clone()
            }
            false => {
                object! {}
            }
        }
    }

    fn value(&mut self, field: &str) -> JsonValue {
        self.params.fields = object! {};
        self.params.fields[field] = format!("{}.{}", self.params.table, field).into();
        self.params.page = 1;
        self.params.limit = 1;
        let sql = self.params.select_sql();
        if self.params.sql {
            return JsonValue::from(sql.clone());
        }
        let (state, mut data) = self.query(sql.as_str());
        match state {
            true => {
                for (field, _) in self.params.json.entries() {
                    if !data[0][field].is_empty() {
                        let json = data[0][field].to_string();
                        let json = json::parse(&json).unwrap_or(array![]);
                        data[0][field] = json;
                    } else {
                        data[0][field] = array![];
                    }
                }
                data[0][field].clone()
            }
            false => {
                if self.connection.debug {
                    info!("{data:?}");
                }
                JsonValue::Null
            }
        }
    }

    fn insert(&mut self, mut data: JsonValue) -> JsonValue {
        let mut fields = vec![];
        let mut values = vec![];
        if !self.params.autoinc && data["id"].is_empty() {
            data["id"] = format!("{:X}", Local::now().timestamp_nanos_opt().unwrap()).into();
        }
        for (field, value) in data.entries() {
            fields.push(field);

            if value.is_string() {
                values.push(format!("'{}'", value.to_string().replace("'", "''")));
                continue;
            } else if value.is_array() {
                if self.params.json[field].is_empty() {
                    let array = value.members().map(|x| x.as_str().unwrap()).collect::<Vec<&str>>().join(",");
                    values.push(format!("'{array}'"));
                } else {
                    let json = value.to_string();
                    let json = json.replace("'", "''");
                    values.push(format!("'{json}'"));
                }
                continue;
            } else if value.is_object() {
                if self.params.json[field].is_empty() {
                    values.push(format!("'{value}'"));
                } else {
                    let json = value.to_string();
                    let json = json.replace("'", "''");
                    values.push(format!("'{json}'"));
                }
                continue;
            } else if value.is_number() || value.is_boolean() || value.is_null() {
                values.push(format!("{value}"));
                continue;
            } else {
                values.push(format!("'{value}'"));
                continue;
            }
        }
        let fields = fields.join(",");
        let values = values.join(",");

        let sql = format!(
            "INSERT INTO {} ({}) VALUES ({});",
            self.params.table, fields, values
        );
        if self.params.sql {
            return JsonValue::from(sql.clone());
        }
        let (state, ids) = self.execute(sql.as_str());

        match state {
            true => match self.params.autoinc {
                true => ids.clone(),
                false => data["id"].clone(),
            },
            false => {
                let thread_id = format!("{:?}", thread::current().id());
                error!("添加失败: {thread_id} {ids:?} {sql}");
                JsonValue::from("")
            }
        }
    }
    fn insert_all(&mut self, mut data: JsonValue) -> JsonValue {
        let mut fields = String::new();
        if !self.params.autoinc && data[0]["id"].is_empty() {
            data[0]["id"] = "".into();
        }
        for (field, _) in data[0].entries() {
            fields = format!("{fields},{field}");
        }
        fields = fields.trim_start_matches(",").parse().unwrap();

        let core_count = num_cpus::get();
        let mut p = pools::Pool::new(core_count * 4);

        let autoinc = self.params.autoinc;
        for list in data.members() {
            let mut item = list.clone();
            let i = br_fields::str::Code::verification_code(3);
            p.execute(move |pcindex| {
                if !autoinc && item["id"].is_empty() {
                    let id = format!(
                        "{:X}{:X}{}",
                        Local::now().timestamp_nanos_opt().unwrap(),
                        pcindex,
                        i
                    );
                    item["id"] = id.into();
                }
                let mut values = "".to_string();
                for (_, value) in item.entries() {
                    if value.is_string() {
                        values = format!("{},'{}'", values, value.to_string().replace("'", "''"));
                    } else if value.is_number() {
                        values = format!("{values},{value}");
                    } else if value.is_boolean() {
                        values = format!("{values},{value}");
                        continue;
                    } else {
                        values = format!("{},'{}'", values, value.to_string().replace("'", "''"));
                    }
                }
                values = format!("({})", values.trim_start_matches(","));
                array![item["id"].clone(), values]
            });
        }
        let (ids_list, mut values) = p.insert_all();
        values = values.trim_start_matches(",").parse().unwrap();
        let sql = format!(
            "INSERT INTO {} ({}) VALUES {};",
            self.params.table, fields, values
        );

        if self.params.sql {
            return JsonValue::from(sql.clone());
        }
        let (state, data) = self.execute(sql.as_str());
        match state {
            true => match autoinc {
                true => data,
                false => JsonValue::from(ids_list),
            },
            false => {
                error!("insert_all: {data:?}");
                array![]
            }
        }
    }
    fn update(&mut self, data: JsonValue) -> JsonValue {
        let mut values = vec![];
        for (field, value) in data.entries() {
            if value.is_string() {
                values.push(format!(
                    "{}='{}'",
                    field,
                    value.to_string().replace("'", "''")
                ));
            } else if value.is_number() {
                values.push(format!("{field}= {value}"));
            } else if value.is_array() {
                if self.params.json[field].is_empty() {
                    let array = value.members().map(|x| x.as_str().unwrap()).collect::<Vec<&str>>().join(",");
                    values.push(format!("{field}='{array}'"));
                } else {
                    let json = value.to_string();
                    let json = json.replace("'", "''");
                    values.push(format!("{field}='{json}'"));
                }
                continue;
            } else if value.is_object() {
                if self.params.json[field].is_empty() {
                    values.push(format!("{field}='{value}'"));
                } else {
                    if value.is_empty() {
                        values.push(format!("{field}=''"));
                        continue;
                    }
                    let json = value.to_string();
                    let json = json.replace("'", "''");
                    values.push(format!("{field}='{json}'"));
                }
                continue;
            } else if value.is_boolean() {
                values.push(format!("{field}= {value}"));
            } else {
                values.push(format!("{field}=\"{value}\""));
            }
        }

        for (field, value) in self.params.inc_dec.entries() {
            values.push(format!("{} = {}", field, value.to_string().clone()));
        }

        let values = values.join(",");

        let sql = format!(
            "UPDATE {} SET {} {};",
            self.params.table.clone(),
            values,
            self.params.where_sql()
        );
        if self.params.sql {
            return JsonValue::from(sql.clone());
        }
        let (state, data) = self.execute(sql.as_str());
        if state {
            data
        } else {
            let thread_id = format!("{:?}", thread::current().id());
            error!("update: {thread_id} {data:?} {sql}");
            0.into()
        }
    }
    fn update_all(&mut self, data: JsonValue) -> JsonValue {
        let mut values = vec![];

        let mut ids = vec![];
        for (field, _) in data[0].entries() {
            if field == "id" {
                continue;
            }
            let mut fields = vec![];
            for row in data.members() {
                let value = row[field].clone();
                let id = row["id"].clone();
                ids.push(id.clone());
                if value.is_string() {
                    fields.push(format!(
                        "WHEN '{}' THEN '{}'",
                        id,
                        value.to_string().replace("'", "''")
                    ));
                } else if value.is_array() || value.is_object() {
                    if self.params.json[field].is_empty() {
                        fields.push(format!("WHEN '{id}' THEN '{value}'"));
                    } else {
                        let json = value.to_string();
                        let json = json.replace("'", "''");
                        fields.push(format!("WHEN '{id}' THEN '{json}'"));
                    }
                    continue;
                } else if value.is_number() || value.is_boolean() || value.is_null() {
                    fields.push(format!("WHEN '{id}' THEN {value}"));
                } else {
                    fields.push(format!("WHEN '{id}' THEN '{value}'"));
                }
            }
            values.push(format!("{} = CASE id {} END", field, fields.join(" ")))
        }
        self.where_and("id", "in", ids.into());
        for (field, value) in self.params.inc_dec.entries() {
            values.push(format!("{} = {}", field, value.to_string().clone()));
        }

        let values = values.join(",");
        let sql = format!(
            "UPDATE {} SET {} {} {};",
            self.params.table.clone(),
            values,
            self.params.where_sql(),
            self.params.page_limit_sql()
        );
        if self.params.sql {
            return JsonValue::from(sql.clone());
        }
        let (state, data) = self.execute(sql.as_str());
        if state {
            data
        } else {
            error!("update_all: {data:?}");
            JsonValue::from(0)
        }
    }

    fn delete(&mut self) -> JsonValue {
        let sql = format!(
            "delete FROM {} {} {};",
            self.params.table.clone(),
            self.params.where_sql(),
            self.params.page_limit_sql()
        );
        if self.params.sql {
            return JsonValue::from(sql.clone());
        }
        let (state, data) = self.execute(sql.as_str());
        match state {
            true => data,
            false => {
                error!("delete 失败>>> {data:?}");
                JsonValue::from(0)
            }
        }
    }

    fn transaction(&mut self) -> bool {
        let thread_id = format!("{:?}", thread::current().id());

        if TRANS.lock().unwrap().get(&*thread_id).is_some() {
            let mut t = *TRANS.lock().unwrap().get_mut(&*thread_id).unwrap();
            t += 1;
            TRANS.lock().unwrap().insert(thread_id.clone(), t);
            return true;
        }

        TRANS.lock().unwrap().insert(thread_id.clone(), 1);
        let key = format!("{}{}", self.default, thread_id);

        // 获取连接并克隆（事务需要长期持有）
        let mut guard = match self.client.get_guard() {
            Ok(g) => g,
            Err(e) => {
                error!("获取事务连接失败: {e}");
                return false;
            }
        };

        let conn = guard.conn().clone();

        // 手动归还这个守卫，因为我们只需要获取连接
        drop(guard);

        TR.lock().unwrap().insert(key.clone(), Arc::new(Mutex::new(conn)));

        let sql = "START TRANSACTION;".to_string();
        let (state, _) = self.execute(sql.as_str());
        match state {
            true => {
                let sql = "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;".to_string();
                let (state, _) = self.execute(sql.as_str());
                match state {
                    true => state,
                    false => {
                        TRANS.lock().unwrap().remove(&*thread_id.clone());
                        TR.lock().unwrap().remove(&key.clone());
                        state
                    }
                }
            }
            false => {
                TRANS.lock().unwrap().remove(&*thread_id.clone());
                TR.lock().unwrap().remove(&key.clone());
                state
            }
        }
    }
    fn commit(&mut self) -> bool {
        let thread_id = format!("{:?}", thread::current().id());
        let sql = "COMMIT".to_string();

        let mut t = *TRANS.lock().unwrap().get(&*thread_id).unwrap_or(&0);
        if t > 1 {
            t -= 1;
            TRANS.lock().unwrap().insert(thread_id.clone(), t);
            return true;
        }
        let (state, data) = self.execute(sql.as_str());
        TRANS.lock().unwrap().remove(&thread_id);
        let key = format!("{}{}", self.default, thread_id);
        TR.lock().unwrap().remove(&*key);
        match state {
            true => {}
            false => {
                error!("提交事务失败: {data}");
            }
        }
        state
    }

    fn rollback(&mut self) -> bool {
        let thread_id = format!("{:?}", thread::current().id());
        let sql = "ROLLBACK".to_string();

        let mut t = *TRANS.lock().unwrap().get(&thread_id).unwrap();
        if t > 1 {
            t -= 1;
            TRANS.lock().unwrap().insert(thread_id.clone(), t);
            return true;
        }
        let (state, data) = self.execute(sql.as_str());
        TRANS.lock().unwrap().remove(&thread_id);
        let key = format!("{}{}", self.default, thread_id);
        TR.lock().unwrap().remove(&*key);
        match state {
            true => {}
            false => {
                error!("回滚失败: {data}");
            }
        }
        state
    }

    fn sql(&mut self, sql: &str) -> Result<JsonValue, String> {
        let (state, data) = self.query(sql);
        match state {
            true => Ok(data),
            false => Err(data.to_string()),
        }
    }

    fn sql_execute(&mut self, sql: &str) -> Result<JsonValue, String> {
        let (state, data) = self.execute(sql);
        match state {
            true => Ok(data),
            false => Err(data.to_string()),
        }
    }

    fn inc(&mut self, field: &str, num: f64) -> &mut Self {
        self.params.inc_dec[field] = format!("{field} + {num}").into();
        self
    }

    fn dec(&mut self, field: &str, num: f64) -> &mut Self {
        self.params.inc_dec[field] = format!("{field} - {num}").into();
        self
    }
    fn buildsql(&mut self) -> String {
        self.fetch_sql();
        let sql = self.select().to_string();
        format!("( {} ) {}", sql, self.params.table)
    }

    fn join_fields(&mut self, fields: Vec<&str>) -> &mut Self {
        for field in fields {
            self.params.fields[field] = format!("{field} as {}", field.replace(".", "_")).into();
        }
        self
    }

    fn join(&mut self, main_table: &str, main_fields: &str, right_table: &str, right_fields: &str) -> &mut Self {
        let main_table = if main_table.is_empty() {
            self.params.table.clone()
        } else {
            main_table.to_string()
        };
        self.params.join_table = right_table.to_string();
        self.params.join.push(format!(" LEFT JOIN {right_table} ON {main_table}.{main_fields} = {right_table}.{right_fields} "));
        self
    }

    fn join_inner(&mut self, table: &str, main_fields: &str, second_fields: &str) -> &mut Self {
        let main_fields = if main_fields.is_empty() {
            "id"
        } else {
            main_fields
        };
        let second_fields = if second_fields.is_empty() {
            self.params.table.clone()
        } else {
            second_fields.to_string().clone()
        };
        let sec_table_name = format!("{}{}", table, "_2");
        let second_table = format!("{} {}", table, sec_table_name.clone());
        self.params.join_table = sec_table_name.clone();
        self.params.join.push(format!(
            " INNER JOIN {} ON {}.{} = {}.{}",
            second_table, self.params.table, main_fields, sec_table_name, second_fields
        ));
        self
    }
}
