<?php

namespace Jiaoyin;

use Swoole\Database\MysqliConfig;
use Swoole\Database\MysqliPool;
use Swoole\Coroutine;
use Swoole\Coroutine\Channel;
use Jiaoyin\StreamLogger;

class MysqlCli
{

    private $pool   = null;
    private $prefix = ''; // 前缀

    private int $reconnectCount    = 0; //period时间内重连次数
    private int $period            = 300;
    private int $lastReconnectTime = 0;
    private     $connectConfig     = null;
    private     $database          = null;
    private     $logger            = null;

    public function __construct($host, $port, $database, $username, $password, $charset = 'utf8', $prefix = '', $connectCount = 20)
    {
        $this->database      = $database;
        $this->prefix        = $prefix;
        $this->connectConfig = [
            'host'         => $host,
            'port'         => $port,
            'database'     => $database,
            'username'     => $username,
            'password'     => $password,
            'charset'      => $charset,
            'prefix'       => $prefix,
            'connectCount' => $connectCount,
        ];
        $this->poolConnect();
        $this->logger = new StreamLogger(); //默认路径
    }

    private function poolConnect()
    {
        if ($this->pool) {
            $this->pool->close();
        }
        $this->pool   = new MysqliPool((new MysqliConfig)
                ->withHost($this->connectConfig['host'])
                ->withPort($this->connectConfig['port'])
                ->withDbName($this->connectConfig['database'])
                ->withCharset($this->connectConfig['charset'])
                ->withUsername($this->connectConfig['username'])
                ->withPassword($this->connectConfig['password']),
            $this->connectConfig['connectCount']
        );
        $this->prefix = $this->connectConfig['prefix'];
    }

    public function execute($action, $sql, float $timeout = 5.0, int $retry = 1, bool $assoc = false)
    {
        retry_label:
        $db = null;
        $start_time = microtime(true);
        try {
            $db = $this->pool->get();
            if (!$db) {
                $this->logger->error('mysql', "数据库连接池耗尽，请检查数据库连接配置...");
                throw new \RuntimeException("数据库连接池耗尽");
            }
            $chan = new Channel(1);
            Coroutine::create(function () use ($chan, $db, $sql) {
                try {
                    $result = $db->query($sql);
                    $chan->push($result);
                } catch (\Throwable $e) {
                    $chan->push($e);
                }
            });

            /** 等待执行结果（支持超时） */
            $result = $chan->pop($timeout);

            /** ---------- 处理超时 ---------- */
            if ($result === false) {
                $this->logger->error('mysql', "SQL超时: {$timeout}s, SQL: {$sql}");
                if ($retry > 0) {
                    $this->logger->error('mysql', "执行超时，SQL:{$sql} 正在销毁连接并重试...");
                    // 1. 永远不要把超时的连接放回池
                    if ($db) {
                        @$db->close();   // 强制关闭
                        // 重要：告诉连接池不要再使用这个连接
                        $db = null;
                    }
                    // . 重建一个新连接补入池中
                    $retry--;
                    goto retry_label;
                }

                return false;
            }
            /** ---------- 处理异常（断开 / gone away / server has gone away） ---------- */
            if ($result instanceof \Throwable) {
                $msg = $result->getMessage();
                $this->logger->error('mysql', "执行异常: {$msg}; SQL: {$sql}");
                $needReconnect = false;
                if (
                    stripos($msg, 'gone away') !== false ||
                    stripos($msg, 'Lost connection') !== false ||
                    stripos($msg, 'server has gone away') !== false ||
                    stripos($msg, 'server closed') !== false ||
                    stripos($msg, 'Connection reset') !== false
                ) {
                    $needReconnect = true;
                }

                if ($needReconnect && $retry > 0) {
                    $this->logger->error('mysql', "数据库连接断开，SQL:{$sql}自动重连并重试 SQL...");
                    if ($db) {
                        try {
                            $db->close();   // 强制关闭
                            // 重要：告诉连接池不要再使用这个连接
                            $db = null;
                        } catch (\Throwable $e) {
                            $this->logger->error('mysql', "db close 失败，SQL:{$sql}忽略");
                        }
                    }
                    $this->poolConnect();

                    $retry--;
                    goto retry_label;
                }

                return false;
            }

            /** ---------- 成功执行 ---------- */
            if ($action === 'SELECT') {
                if ($assoc) {
                    $rows = $result->fetch_all(MYSQLI_ASSOC);
                } else {
                    $rows = $result->fetch_all();
                }
                $result->free();
                return $rows;
            }

            return $result;
        } catch (\Throwable $e) {
            $this->logger->error('mysql', "execute() SQL:{$sql}致命异常: " . $e->getMessage());
            if ($retry > 0) {
                $this->logger->error('mysql', "致命异常：SQL:{$sql}尝试重连并重试");
                if ($db) {
                    @$db->close();   // 强制关闭
                    $db = null;
                }
                $retry--;
                goto retry_label;
            }

            return false;
        } finally {
            $end_time = microtime(true);
            $count_time = $end_time - $start_time;
            if ($count_time > 0.5) {
                $this->logger->log('mysql', "SQL执行时间: " . $count_time . "s, SQL: {$sql}");
            }
            if ($db) {
                try {
                    $this->pool->put($db);
                } catch (\Throwable $e) {
                    $this->logger->error('mysql', "连接 put 回池失败（可能已断开），SQL:{$sql}忽略");
                }
            }
        }
    }

    //插入数据
    public function insert($table, $data)
    {
        $sql = $this->parseInsert($table, $data);
        if (!$sql) {
            return false;
        }
        return $this->execute('INSERT', $sql);
    }

    //select数据
    public function select($table, $where = [], $col = [], $orderBy = '', $limit = '', $groupBy = '')
    {
        $sql = $this->parseSelect($table, $where, $col, $orderBy, $limit, $groupBy);
        if (!$sql) {
            return false;
        }
        // 判断是否包含聚合函数
        $isAggregate = false;
        foreach ($col as $c) {
            if (stripos($c, 'sum(') !== false || stripos($c, 'count(') !== false || stripos($c, 'avg(') !== false || stripos($c, 'max(') !== false || stripos($c, 'min(') !== false) {
                $isAggregate = true;
                break;
            }
        }

        if ($isAggregate && empty($groupBy)) {
            // 没有 group by 的单行聚合查询
            $db = $this->pool->get();
            try {
                $stmt = $db->query($sql);
                $result = $stmt->fetch_assoc();
                $stmt->free(); // 释放结果集，防止Commands out of sync错误
                return [$result];
            } finally {
                $this->pool->put($db);
            }
        } elseif ($isAggregate && !empty($groupBy)) {
            // 有 group by 的聚合查询，多行返回
            $db = $this->pool->get();
            try {
                $stmt = $db->query($sql);
                $results = [];
                while ($row = $stmt->fetch_assoc()) {
                    $results[] = $row;
                }
                $stmt->free(); // 释放结果集，防止Commands out of sync错误
                return $results;
            } finally {
                $this->pool->put($db);
            }
        }

        // 非聚合查询，使用execute方法获取数据
        $data = $this->execute('SELECT', $sql);

        // 如果没有数据
        if (!$data) return [];

        // 普通查询才查字段结构
        if (empty($col)) {
            $newsql  = 'SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = "' . $this->database . '" AND TABLE_NAME="' . $this->parseTable($table) . '" ORDER BY `ORDINAL_POSITION` ASC';
            $coldata = $this->execute('SELECT', $newsql);
            if (!$coldata) {
                return false;
            } else {
                foreach ($coldata as $key => $value) {
                    $col[] = $value[0];
                }
            }
        }

        foreach ($data as $key => $value) {
            $tmp = [];
            foreach ($col as $k => $v) {
                $tmp[$v] = $value[$k];
            }
            $data[$key] = $tmp;
        }
        return $data;
    }

    public function update($table, $data, $where = [])
    {
        $sql = $this->parseUpdate($table, $where, $data);
        if (!$sql) {
            return false;
        }
        return $this->execute('UPDATE', $sql);
    }

    //查找符合条件的一条记录
    public function find($table, $where, $col = [], $orderBy = '')
    {
        $data = $this->select($table, $where, $col, $orderBy, 1);
        if ($data && isset($data[0])) {
            return $data['0'];
        }
        return $data;
    }

    //解析insert
    private function parseInsert($table, $data)
    {
        $table = $this->parseTable($table);
        if (!is_array($data)) {
            $this->logger->error('mysql', 'insert data 不合法');
            return false;
        }
        $keys   = array_keys($data);
        $values = array_values($data);
        foreach ($values as $key => $value) {
            $values[$key] = $this->transferValue($value);
        }
        $keysFormat   = implode(',', $keys);
        $valuesFormat = implode(',', $values);
        $sql          = "insert into {$table}({$keysFormat}) values({$valuesFormat})";
        return $sql;
    }

    //解析select
    private function parseSelect($table, $where, $col, $orderBy, $limit, $groupBy = '')
    {
        $table = $this->parseTable($table);

        if (!empty($col) && !is_array($col)) {
            $this->logger->error('mysql', 'select where col 不合法');
            return false;
        }

        $colTxt     = empty($col) ? '*' : implode(',', $col);
        $whereTxt   = $this->parseWhere($where);
        $groupByTxt = !empty($groupBy) ? 'group by ' . $groupBy : '';
        $orderByTxt = (!empty($orderBy) && isset($orderBy[0]) && isset($orderBy[1]))
            ? 'order by ' . $orderBy[0] . ' ' . $orderBy[1]
            : '';
        $limitTxt   = !empty($limit) ? 'limit ' . $limit : '';

        $sql = "select {$colTxt} from {$table} {$whereTxt} {$groupByTxt} {$orderByTxt} {$limitTxt}";
        return $sql;
    }


    //解析where
    private function parseWhere($where)
    {
        if (empty($where)) {
            return '';
        }
        $whereTxt = 'where ';
        if (is_string($where)) {
            $whereTxt .= $where;
        } else if (is_array($where)) {
            $tmp = '';
            foreach ($where as $key => $value) {
                if (!is_array($value) && !isset($value[1])) {
                    $this->logger->error('mysql', 'where 不合法');
                    var_dump($where);
                    return false;
                }
                $symbol = $value[2] ?? '=';
                // in + 数组专门处理
                if (strtolower($symbol) === 'in' && is_array($value[1])) {
                    $val = "('" . implode("','", array_map('addslashes', $value[1])) . "')";
                    $tmp .= $value[0] . ' in ' . $val;
                } elseif ($value[1] === 'NULL') {
                    $tmp .= $value[0] . ' is NULL';
                } else {
                    $val = $this->transferValue($value[1]);
                    $tmp .= $value[0] . ' ' . $symbol . ' ' . $val;
                }
                $tmp .= $key < count($where) - 1 ? ' and ' : '';
            }
            $whereTxt .= $tmp;
        } else {
            $this->logger->error('mysql', 'where 不合法');

            var_dump($where);
            return false;
        }
        return $whereTxt;
    }

    //解析udpate
    private function parseUpdate($table, $where, $data)
    {
        $table = $this->parseTable($table);
        if (!is_array($data) || empty($data)) {
            $this->logger->error('mysql', '更新数据不能为空');

            return false;
        }
        $setTxt = 'set ';
        $count  = 0;
        foreach ($data as $key => $value) {
            $setTxt .= $key . '=';
            $setTxt .= $this->transferValue($value);
            if ($count < count($data) - 1) {
                $setTxt .= ',';
            }
            $count += 1;
        }
        $whereTxt = $this->parseWhere($where);
        $sql      = 'update ' . $table . ' ' . $setTxt . ' ' . $whereTxt;
        return $sql;
    }

    private function parseTable($table)
    {
        return $this->prefix . $table;
    }

    //值转义
    private function transferValue($value)
    {
        $valueNew = null;
        if (is_string($value)) {
            $valueNew = '"' . addslashes($value) . '"';
        } else if ($value === null) {
            $valueNew = 'NULL';
        } else {
            if (is_numeric($value) && strpos($value . '', 'E') !== false) {
                $value = number_format($value, 8);
                while (substr($value, -1) === '0') {
                    $value = substr($value, 0, -1);
                }
            }
            $valueNew = $value;
        }
        return $valueNew;
    }
    // 删除数据
    public function delete($table, $where = [])
    {
        $sql = $this->parseDelete($table, $where);
        if (!$sql) {
            return false;
        }
        return $this->execute('DELETE', $sql);
    }

    // 解析 delete
    private function parseDelete($table, $where)
    {
        $table = $this->parseTable($table);
        $whereTxt = $this->parseWhere($where);

        if (empty($whereTxt)) {
            $this->logger->error('mysql', 'delete 必须指定 where 条件，防止全表删除');
            return false;
        }

        $sql = "delete from {$table} {$whereTxt}";
        return $sql;
    }
}