/*
 * Copyright (c) 2016 The ZLToolKit project authors. All Rights Reserved.
 *
 * This file is part of ZLToolKit(https://github.com/ZLMediaKit/ZLToolKit).
 *
 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
 */

#ifndef SQL_SQLPOOL_H_
#define SQL_SQLPOOL_H_

#include <deque>
#include <mutex>
#include <memory>
#include <sstream>
#include <functional>
#include "logger.h"
#include "Poller/Timer.h"
#include "SqlConnection.h"
#include "Thread/WorkThreadPool.h"
#include "ResourcePool.h"

namespace toolkit {

class SqlPool : public std::enable_shared_from_this<SqlPool> {
public:
    using Ptr = std::shared_ptr<SqlPool>;
    using PoolType = ResourcePool<SqlConnection>;
    using SqlRetType = std::vector<std::vector<std::string> >;

    static SqlPool &Instance();

    ~SqlPool() {
        _timer.reset();
        flushError();
        _threadPool.reset();
        _pool.reset();
        InfoL;
    }

    /**
     * 设置循环池对象个数
     * @param size
     * Sets the number of loop pool objects
     * @param size
     
     * [AUTO-TRANSLATED:a3f3a8ac]
     */
    void setSize(int size) {
        checkInited();
        _pool->setSize(size);
    }

    /**
     * 初始化循环池，设置数据库连接参数
     * @tparam Args
     * @param arg
     * Initializes the loop pool and sets the database connection parameters
     * @tparam Args
     * @param arg
     
     * [AUTO-TRANSLATED:fff84748]
     */
    template<typename ...Args>
    void Init(Args &&...arg) {
        _pool.reset(new PoolType(std::forward<Args>(arg)...));
        _pool->obtain();
    }


    /**
     * 异步执行sql
     * @param str sql语句
     * @param tryCnt 重试次数
     * Asynchronously executes SQL
     * @param str SQL statement
     * @param tryCnt Number of retries
     
     * [AUTO-TRANSLATED:9d2414e1]
     */
    template<typename ...Args>
    void asyncQuery(Args &&...args) {
        asyncQuery_l(SqlConnection::queryString(std::forward<Args>(args)...));
    }


    /**
     * 同步执行sql
     * @tparam Args 可变参数类型列表
     * @param arg 可变参数列表
     * @return 影响行数
     * Synchronously executes SQL
     * @tparam Args Variable parameter type list
     * @param arg Variable parameter list
     * @return Number of affected rows
     
     * [AUTO-TRANSLATED:ba47cb3c]
     */

    template<typename ...Args>
    int64_t syncQuery(Args &&...arg) {
        checkInited();
        typename PoolType::ValuePtr mysql;
        try {
            //捕获执行异常  [AUTO-TRANSLATED:ae8a1093]
            //Capture execution exceptions
            mysql = _pool->obtain();
            return mysql->query(std::forward<Args>(arg)...);
        } catch (std::exception &e) {
            mysql.quit();
            throw;
        }
    }


    /**
     * sql转义
     * @param str
     * @return
     * Escapes SQL
     * @param str
     * @return
     
     * [AUTO-TRANSLATED:e7a99a20]
     */
    std::string escape(const std::string &str) {
        checkInited();
        return _pool->obtain()->escape(const_cast<std::string &>(str));
    }

private:
    SqlPool() {
        _threadPool = WorkThreadPool::Instance().getExecutor();
        _timer = std::make_shared<Timer>(30, [this]() {
            flushError();
            return true;
        }, nullptr);
    }

    /**
     * 异步执行sql
     * @param sql sql语句
     * @param tryCnt 重试次数
     * Asynchronously executes SQL
     * @param sql SQL statement
     * @param tryCnt Number of retries
     
     * [AUTO-TRANSLATED:6f585bf1]
     */
    void asyncQuery_l(const std::string &sql, int tryCnt = 3) {
        auto lam = [this, sql, tryCnt]() {
            int64_t rowID;
            auto cnt = tryCnt - 1;
            try {
                syncQuery(rowID, sql);
            } catch (std::exception &ex) {
                if (cnt > 0) {
                    //失败重试  [AUTO-TRANSLATED:ef091479]
                    //Retry on failure
                    std::lock_guard<std::mutex> lk(_error_query_mutex);
                    sqlQuery query(sql, cnt);
                    _error_query.push_back(query);
                } else {
                    WarnL << "SqlPool::syncQuery failed: " << ex.what();
                }
            }
        };
        _threadPool->async(lam);
    }

    /**
     * 定时重试失败的sql
     * Periodically retries failed SQL
     
     * [AUTO-TRANSLATED:33048898]
     */
    void flushError() {
        decltype(_error_query) query_copy;
        {
            std::lock_guard<std::mutex> lck(_error_query_mutex);
            query_copy.swap(_error_query);
        }
        for (auto &query : query_copy) {
            asyncQuery(query.sql_str, query.tryCnt);
        }
    }

    /**
     * 检查数据库连接池是否初始化
     * Checks if the database connection pool is initialized
     
     * [AUTO-TRANSLATED:176fceed]
     */
    void checkInited() {
        if (!_pool) {
            throw SqlException("SqlPool::checkInited", "Mysql connection pool not initialized");
        }
    }

private:
    struct sqlQuery {
        sqlQuery(const std::string &sql, int cnt) : sql_str(sql), tryCnt(cnt) {}

        std::string sql_str;
        int tryCnt = 0;
    };

private:
    std::deque<sqlQuery> _error_query;
    TaskExecutor::Ptr _threadPool;
    std::mutex _error_query_mutex;
    std::shared_ptr<PoolType> _pool;
    Timer::Ptr _timer;
};

/**
 * Sql语句生成器，通过占位符'？'的方式生成sql语句
 * SQL statement generator, generates SQL statements through the '?' placeholder
 
 * [AUTO-TRANSLATED:12f34981]
 */
class SqlStream {
public:
    SqlStream(const char *sql) : _sql(sql) {}

    ~SqlStream() {}

    template<typename T>
    SqlStream &operator<<(T &&data) {
        auto pos = _sql.find('?', _startPos);
        if (pos == std::string::npos) {
            return *this;
        }
        _str_tmp.str("");
        _str_tmp << std::forward<T>(data);
        std::string str = SqlPool::Instance().escape(_str_tmp.str());
        _startPos = pos + str.size();
        _sql.replace(pos, 1, str);
        return *this;
    }

    const std::string &operator<<(std::ostream &(*f)(std::ostream &)) const {
        return _sql;
    }

    operator std::string() {
        return _sql;
    }

private:
    std::stringstream _str_tmp;
    std::string _sql;
    std::string::size_type _startPos = 0;
};


/**
 * sql查询器
 * SQL query executor
 
 * [AUTO-TRANSLATED:50396624]
 */
class SqlWriter {
public:
    /**
     * 构造函数
     * @param sql 带'？'占位符的sql模板
     * @param throwAble 是否抛异常
     * Constructor
     * @param sql SQL template with '?' placeholder
     * @param throwAble Whether to throw exceptions
     
     * [AUTO-TRANSLATED:97c6d354]
     */
    SqlWriter(const char *sql, bool throwAble = true) : _sqlstream(sql), _throwAble(throwAble) {}

    ~SqlWriter() {}

    /**
     * 输入参数替换占位符'？'以便生成sql语句；可能抛异常
     * @tparam T 参数类型
     * @param data 参数
     * @return 本身引用
     * Replaces '?' placeholders with input parameters to generate SQL statements; may throw exceptions
     * @tparam T Parameter type
     * @param data Parameter
     * @return Self-reference
     
     * [AUTO-TRANSLATED:9bdc6917]
     */
    template<typename T>
    SqlWriter &operator<<(T &&data) {
        try {
            _sqlstream << std::forward<T>(data);
        } catch (std::exception &ex) {
            //在转义sql时可能抛异常  [AUTO-TRANSLATED:ce6314cc]
            //May throw exceptions when escaping SQL
            if (!_throwAble) {
                WarnL << "Commit sql failed: " << ex.what();
            } else {
                throw;
            }
        }
        return *this;
    }

    /**
     * 异步执行sql，不会抛异常
     * @param f std::endl
     * Asynchronously executes SQL, does not throw exceptions
     * @param f std::endl
     
     * [AUTO-TRANSLATED:e203d266]
     */
    void operator<<(std::ostream &(*f)(std::ostream &)) {
        //异步执行sql不会抛异常  [AUTO-TRANSLATED:4370797e]
        //Asynchronously executes SQL, does not throw exceptions
        SqlPool::Instance().asyncQuery((std::string) _sqlstream);
    }

    /**
     * 同步执行sql，可能抛异常
     * @tparam Row 数据行类型，可以是vector<string>/list<string>等支持 obj.emplace_back("value")操作的数据类型
     * 			   也可以是map<string,string>/Json::Value 等支持 obj["key"] = "value"操作的数据类型
     * @param ret 数据存放对象
     * @return 影响行数
     * Synchronously executes SQL, may throw exceptions
     * @tparam Row Data row type, can be vector<string>/list<string> or other types that support obj.emplace_back("value") operations
     *              Can also be map<string,string>/Json::Value or other types that support obj["key"] = "value" operations
     * @param ret Data storage object
     * @return Number of affected rows
     
     * [AUTO-TRANSLATED:d8e40f96]
     */
    template<typename Row>
    int64_t operator<<(std::vector<Row> &ret) {
        try {
            _affectedRows = SqlPool::Instance().syncQuery(_rowId, ret, (std::string) _sqlstream);
        } catch (std::exception &ex) {
            if (!_throwAble) {
                WarnL << "SqlPool::syncQuery failed: " << ex.what();
            } else {
                throw;
            }
        }
        return _affectedRows;
    }

    /**
     * 在insert数据库时返回插入的rowid
     * @return
     * Returns the rowid inserted into the database when inserting data
     * @return
     
     * [AUTO-TRANSLATED:699edcc4]
     */
    int64_t getRowID() const {
        return _rowId;
    }

    /**
     * 返回影响数据库数据行数
     * @return
     * Returns the number of rows affected in the database
     * @return
     
     * [AUTO-TRANSLATED:81af02d9]
     */
    int64_t getAffectedRows() const {
        return _affectedRows;
    }

private:
    SqlStream _sqlstream;
    int64_t _rowId = -1;
    int64_t _affectedRows = -1;
    bool _throwAble = true;
};

} /* namespace toolkit */
#endif /* SQL_SQLPOOL_H_ */
