ylib/src/db/mysql.cpp

/*Software License

Copyright(C) 2024 [liuyingjie]

License Terms
Usage Rights

Any individual or entity is free to use, copy, and distribute the binary form of this software without modification to the source code, without the need to disclose the source code.
If the source code is modified, the modifications must be open-sourced under the same license. This means that the modifications must be disclosed and accompanied by a copy of this license.

Future Versions Updates

From this version onwards, all future releases will be governed by the terms of the latest version of the license. This license will automatically be nullified and replaced by the new version.
Users must comply with the terms of the new license issued in future releases.

Liability and Disclaimer

This software is provided “as is”, without any express or implied warranties, including but not limited to the warranties of merchantability, fitness for a particular purpose, and non-infringement. In no event shall the author or copyright holder be liable for any claims, damages, or other liabilities, whether in an action of contract, tort, or otherwise, arising from, out of, or in connection with the software or the use or other dealings in the software.

Contact Information

If you have any questions, please contact us: 1585346868@qq.com Or visit our website fwlua.com.
*/

#include "db/mysql.h"

#if defined(_WIN64) || defined(__linux__)

#include <cassert>
#include <iostream>
#include <memory>
#include <sstream>
#include <stdexcept>

#include "cppconn/connection.h"
#include "cppconn/datatype.h"
#include "cppconn/exception.h"
#include "cppconn/metadata.h"
#include "cppconn/prepared_statement.h"
#include "cppconn/resultset.h"
#include "cppconn/resultset_metadata.h"
#include "cppconn/statement.h"
#include "mysql_driver.h"

namespace ylib {
    namespace mysql {

        static std::string make_throw_error(const char* msg) {
            std::string err = std::string("[ylib->mysql->throw]: ") + (msg ? msg : "Unknown error");
#ifdef _DEBUG
            assert(false && "SQL exception occurred");
#endif
            return err;
        }

        static ylib::buffer read_blob_stream(std::istream* stream) {
            ylib::buffer buf;
            if (!stream) return buf;
            char tmp[8192];
            while (stream->read(tmp, sizeof(tmp)) || stream->gcount() > 0) {
                buf.append(tmp, static_cast<size_t>(stream->gcount()));
                if (stream->eof()) break;
            }
            return buf;
        }

        // 事务失败时丢弃连接，避免脏会话回到池中
        static void discard_connection(conn* self) {
            try {
                self->close();
            }
            catch (...) {
            }
        }

        conn::conn()
            : m_handle(nullptr), m_ppst(nullptr), m_sw(0) {
        }

        conn::~conn() {
            close();
        }

        EXAMPLE_START_RESULT conn::start(const mysql_conn_info& info) {
            if (info.ipaddress.empty()) {
                m_lastErrorDesc = "address is empty";
                return SR_FAILED;
            }
            if (info.username.empty()) {
                m_lastErrorDesc = "username is empty";
                return SR_FAILED;
            }
            if (info.charset.empty()) {
                m_lastErrorDesc = "charset is empty";
                return SR_FAILED;
            }
            if (info.port < 1 || info.port > 65535) {
                m_lastErrorDesc = "port out of range";
                return SR_FAILED;
            }

            m_info = info;
            sql::Driver* driver = get_driver_instance();
            sql::ConnectOptionsMap opts;
            opts["hostName"] = info.ipaddress;
            opts["userName"] = info.username;
            opts["password"] = info.password;
            if (!info.database.empty()) opts["schema"] = info.database;
            opts["port"] = static_cast<int>(info.port);
            opts["OPT_CONNECT_TIMEOUT"] = 3;
            opts["OPT_CHARSET_NAME"] = info.charset;
            opts["OPT_SSL_MODE"] = sql::SSL_MODE_DISABLED;
            try {
                m_handle = driver->connect(opts);
                m_sw = 0;
                return SR_SUCCESS;
            }
            catch (const sql::SQLException& e) {
                m_lastErrorDesc = e.what();
                return (m_lastErrorDesc.find("timeout") != std::string::npos)
                    ? SR_TIMEOUT : SR_FAILED;
            }
        }

        void conn::close() {
            if (m_ppst) {
                delete m_ppst;
                m_ppst = nullptr;
            }
            if (!m_handle) return;
            try {
                auto cn = static_cast<sql::Connection*>(m_handle);
                cn->close();
            }
            catch (const std::exception& e) {
                std::cerr << make_throw_error(e.what()) << std::endl;
            }
            delete static_cast<sql::Connection*>(m_handle);
            m_handle = nullptr;
            m_sw = 0;
        }

        void conn::recover() {
            // 归还连接池前必须保证不抛异常，否则池计数会泄漏
            if (m_sw == 1) {
                try {
                    if (m_handle) {
                        auto cn = static_cast<sql::Connection*>(m_handle);
                        cn->rollback();
                        cn->setAutoCommit(true);
                    }
                    m_sw = 0;
                }
                catch (...) {
                    try { close(); }
                    catch (...) {}
                    m_sw = 0;
                }
            }
            else if (m_handle) {
                try {
                    auto cn = static_cast<sql::Connection*>(m_handle);
                    if (!cn->getAutoCommit()) {
                        cn->setAutoCommit(true);
                    }
                    m_sw = 0;
                }
                catch (...) {
                    try { close(); }
                    catch (...) {}
                    m_sw = 0;
                }
            }
            clear();
        }

        void conn::clear() {
            if (m_ppst) {
                delete m_ppst;
                m_ppst = nullptr;
            }
        }

        void conn::task_out() {
            try {
                auto cn = static_cast<sql::Connection*>(m_handle);
                // 仅在连接无效/已丢弃时重建；成功事务不再强制重连
                if (!cn || !cn->isValid()) {
                    close();
                    if (start(m_info) != SR_SUCCESS) {
                        throw exception(make_throw_error(m_lastErrorDesc.c_str()));
                    }
                }
            }
            catch (const sql::SQLException& e) {
                throw exception(make_throw_error(e.what()));
            }
        }

        prepare_statement* conn::setsql(const std::string& sql) {
            if (!m_handle) throw exception("Invalid SQL connection");
            if (sql.empty()) throw exception("SQL string is empty");

            if (m_ppst) {
                delete m_ppst;
                m_ppst = nullptr;
            }
            m_ppst = new prepare_statement();
            try {
                auto cn = static_cast<sql::Connection*>(m_handle);
                m_ppst->m_handle = cn->prepareStatement(sql);
            }
            catch (const sql::SQLException& e) {
                delete m_ppst;
                m_ppst = nullptr;
                throw exception(make_throw_error(e.what()));
            }
            return m_ppst;
        }

        uint64 conn::insert_id() {
            if (!m_handle) throw exception("Invalid SQL connection");
            try {
                auto cn = static_cast<sql::Connection*>(m_handle);
                std::unique_ptr<sql::Statement> stmt(cn->createStatement());
                std::unique_ptr<sql::ResultSet> rs(stmt->executeQuery("SELECT LAST_INSERT_ID()"));
                if (rs && rs->next()) {
                    return rs->getUInt64(1);
                }
                return 0;
            }
            catch (const sql::SQLException& e) {
                throw exception(make_throw_error(e.what()));
            }
        }

        void conn::begin(bool autocommit) {
            if (!m_handle) throw exception("Invalid SQL connection");
            try {
                auto cn = static_cast<sql::Connection*>(m_handle);
                cn->setAutoCommit(autocommit);
                // autocommit=false 才算进入事务
                m_sw = autocommit ? 0 : 1;
            }
            catch (const sql::SQLException& e) {
                throw exception(make_throw_error(e.what()));
            }
        }

        void conn::commit() {
            if (!m_handle) throw exception("Invalid SQL connection");
            try {
                auto cn = static_cast<sql::Connection*>(m_handle);
                cn->commit();
                cn->setAutoCommit(true);
                m_sw = 0;
            }
            catch (const sql::SQLException& e) {
                discard_connection(this);
                throw exception(make_throw_error(e.what()));
            }
        }

        void conn::rollback() {
            if (!m_handle) throw exception("Invalid SQL connection");
            try {
                auto cn = static_cast<sql::Connection*>(m_handle);
                cn->rollback();
                cn->setAutoCommit(true);
                m_sw = 0;
            }
            catch (const sql::SQLException& e) {
                discard_connection(this);
                throw exception(make_throw_error(e.what()));
            }
        }

        void conn::setDatabase(const std::string& name) {
            if (!m_handle) throw exception("Invalid SQL connection");
            try {
                auto cn = static_cast<sql::Connection*>(m_handle);
                cn->setSchema(name);
            }
            catch (const sql::SQLException& e) {
                throw exception(make_throw_error(e.what()));
            }
        }


        // ---------- prepare_statement 实现 ----------

        prepare_statement::prepare_statement()
            : m_handle(nullptr), m_result(nullptr) {
        }

        prepare_statement::~prepare_statement() {
            if (m_result) {
                delete m_result;
                m_result = nullptr;
            }
            if (m_handle) {
                try {
                    auto ps = static_cast<sql::PreparedStatement*>(m_handle);
                    while (ps->getMoreResults()) {
                        auto rs = ps->getResultSet();
                        if (rs) rs->close();
                    }
                    ps->close();
                }
                catch (const sql::SQLException& e) {
                    std::cerr << make_throw_error(e.what()) << std::endl;
                }
                delete static_cast<sql::PreparedStatement*>(m_handle);
                m_handle = nullptr;
            }
        }

#define ENSURE_PS() if (!m_handle) throw exception("Invalid PreparedStatement")

        void prepare_statement::set_bigint(uint32 idx, const std::string& val) {
            ENSURE_PS();
            static_cast<sql::PreparedStatement*>(m_handle)->setBigInt(idx, val);
        }

        void prepare_statement::set_boolean(uint32 idx, bool val) {
            ENSURE_PS();
            static_cast<sql::PreparedStatement*>(m_handle)->setBoolean(idx, val);
        }

        void prepare_statement::set_datetime(uint32 idx, const std::string& val) {
            ENSURE_PS();
            auto ps = static_cast<sql::PreparedStatement*>(m_handle);
            if (val.empty())
                ps->setNull(idx, sql::DataType::DATE);
            else
                ps->setDateTime(idx, val);
        }

        void prepare_statement::set_double(uint32 idx, double val) {
            ENSURE_PS();
            static_cast<sql::PreparedStatement*>(m_handle)->setDouble(idx, val);
        }

        void prepare_statement::set_int32(uint32 idx, int32 val) {
            ENSURE_PS();
            static_cast<sql::PreparedStatement*>(m_handle)->setInt(idx, val);
        }

        void prepare_statement::set_uint32(uint32 idx, uint32 val) {
            ENSURE_PS();
            static_cast<sql::PreparedStatement*>(m_handle)->setUInt(idx, val);
        }

        void prepare_statement::set_int64(uint32 idx, int64 val) {
            ENSURE_PS();
            static_cast<sql::PreparedStatement*>(m_handle)->setInt64(idx, val);
        }

        void prepare_statement::set_uint64(uint32 idx, uint64 val) {
            ENSURE_PS();
            static_cast<sql::PreparedStatement*>(m_handle)->setUInt64(idx, val);
        }

        void prepare_statement::set_null(uint32 idx) {
            ENSURE_PS();
            static_cast<sql::PreparedStatement*>(m_handle)
                ->setNull(idx, sql::DataType::UNKNOWN);
        }

        void prepare_statement::set_string(uint32 idx, const std::string& val) {
            ENSURE_PS();
            static_cast<sql::PreparedStatement*>(m_handle)
                ->setString(idx, sql::SQLString(val));
        }

        void prepare_statement::set_string(uint32 idx, const char* data, int size) {
            ENSURE_PS();
            static_cast<sql::PreparedStatement*>(m_handle)
                ->setString(idx, sql::SQLString(data, size));
        }

        void prepare_statement::set_blob(uint32 idx, const char* data, int size) {
            ENSURE_PS();
            auto ss = std::make_shared<std::stringstream>();
            if (data && size > 0) {
                ss->write(data, size);
                ss->seekg(0);
            }
            m_blobs.push(ss);
            static_cast<sql::PreparedStatement*>(m_handle)->setBlob(idx, ss.get());
        }

        void prepare_statement::clear() {
            if (m_result) {
                delete m_result;
                m_result = nullptr;
            }
            m_blobs = std::queue<std::shared_ptr<std::stringstream>>();
        }


        uint64 prepare_statement::update() {
            ENSURE_PS();
            clear();
            try {
                return static_cast<sql::PreparedStatement*>(m_handle)
                    ->executeUpdate();
            }
            catch (const sql::SQLException& e) {
                throw exception(make_throw_error(e.what()));
            }
        }

        ylib::mysql::result* prepare_statement::query() {
            ENSURE_PS();
            clear();
            try {
                auto rs = static_cast<sql::PreparedStatement*>(m_handle)
                    ->executeQuery();
                m_result = new result(rs);
                return m_result;
            }
            catch (const sql::SQLException& e) {
                throw exception(make_throw_error(e.what()));
            }
        }

#undef ENSURE_PS


        // ---------- data 实现 ----------

        int32 data::to_int32() const {
            return static_cast<int32>(to_int64());
        }

        uint32 data::to_uint32() const {
            return static_cast<uint32>(to_uint64());
        }

        int64 data::to_int64() const {
            switch (m_hold) {
            case hold::null: return 0;
            case hold::i64: return m_i64;
            case hold::u64: return static_cast<int64>(m_u64);
            case hold::f64: return static_cast<int64>(m_f64);
            case hold::boolean: return m_bool ? 1 : 0;
            case hold::str: return m_str.empty() ? 0 : std::stoll(m_str);
            case hold::blob: return 0;
            }
            return 0;
        }

        uint64 data::to_uint64() const {
            switch (m_hold) {
            case hold::null: return 0;
            case hold::i64: return static_cast<uint64>(m_i64);
            case hold::u64: return m_u64;
            case hold::f64: return static_cast<uint64>(m_f64);
            case hold::boolean: return m_bool ? 1 : 0;
            case hold::str: return m_str.empty() ? 0 : std::stoull(m_str);
            case hold::blob: return 0;
            }
            return 0;
        }

        bool data::to_boolean() const {
            switch (m_hold) {
            case hold::null: return false;
            case hold::boolean: return m_bool;
            case hold::i64: return m_i64 != 0;
            case hold::u64: return m_u64 != 0;
            case hold::f64: return m_f64 != 0;
            case hold::str: return !m_str.empty() && m_str != "0";
            case hold::blob: return !m_blob.empty();
            }
            return false;
        }

        double data::to_double() const {
            switch (m_hold) {
            case hold::null: return 0;
            case hold::f64: return m_f64;
            case hold::i64: return static_cast<double>(m_i64);
            case hold::u64: return static_cast<double>(m_u64);
            case hold::boolean: return m_bool ? 1.0 : 0.0;
            case hold::str: return m_str.empty() ? 0 : std::stod(m_str);
            case hold::blob: return 0;
            }
            return 0;
        }

        std::string data::to_string() const {
            switch (m_hold) {
            case hold::null: return "";
            case hold::str: return m_str;
            case hold::i64: return std::to_string(m_i64);
            case hold::u64: return std::to_string(m_u64);
            case hold::f64: return std::to_string(m_f64);
            case hold::boolean: return m_bool ? "1" : "0";
            case hold::blob: return m_blob.to_string();
            }
            return "";
        }

        ylib::buffer data::to_blob() const {
            switch (m_hold) {
            case hold::null: return ylib::buffer();
            case hold::blob: return m_blob;
            case hold::str: return ylib::buffer(m_str);
            default: {
                auto s = to_string();
                return ylib::buffer(s);
            }
            }
        }

        static void field_data_to_json(ylib::json& out, const field& f, const data& d) {
            if (d.is_null()) {
                out = ylib::json::type::null;
                return;
            }
            switch (f.type) {
            case field_type::TINYINT:
            case field_type::SMALLINT:
            case field_type::MEDIUMINT:
            case field_type::INTEGER:
            case field_type::YEAR:
                if (f.is_unsigned)
                    out = d.to_uint32();
                else
                    out = d.to_int32();
                break;
            case field_type::BIGINT:
                if (f.is_unsigned)
                    out = d.to_uint64();
                else
                    out = d.to_int64();
                break;
            case field_type::BIT:
                out = d.to_boolean();
                break;
            case field_type::REAL:
            case field_type::DOUBLE:
            case field_type::DECIMAL:
            case field_type::NUMERIC:
                out = d.to_double();
                break;
            case field_type::BINARY:
            case field_type::VARBINARY:
            case field_type::LONGVARBINARY:
            case field_type::GEOMETRY:
                out = d.to_blob().to_string();
                break;
            default:
                out = d.to_string();
                break;
            }
        }

        // ---------- row 实现 ----------

        bool row::has(const std::string& name) const {
            return m_name_index.find(name) != m_name_index.end();
        }

        const data& row::get(uint32 index) const {
            if (index < 1 || index > m_values.size()) {
                throw exception("Field index out of range");
            }
            return m_values[index - 1];
        }

        const data& row::get(const std::string& name) const {
            auto it = m_name_index.find(name);
            if (it == m_name_index.end()) {
                throw exception("Field not found: " + name);
            }
            return m_values[it->second];
        }

        const field& row::field_info(uint32 index) const {
            if (index < 1 || index > m_fields.size()) {
                throw exception("Field index out of range");
            }
            return m_fields[index - 1];
        }

        ylib::json row::to_json() const {
            ylib::json obj;
            obj = ylib::json::type::obj;
            for (size_t i = 0; i < m_values.size(); ++i) {
                field_data_to_json(obj[m_fields[i].name], m_fields[i], m_values[i]);
            }
            return obj;
        }

        // ---------- result 实现 ----------

        result::result(void* handle)
            : m_handle(handle) {
            auto rs = static_cast<sql::ResultSet*>(m_handle);
            sql::ResultSetMetaData* md = rs->getMetaData();
            if (!md) return;
            uint32 cnt = md->getColumnCount();
            m_fields.reserve(cnt);
            m_field_index.reserve(cnt);
            for (uint32 i = 1; i <= cnt; ++i) {
                field f;
                f.name = md->getColumnLabel(i);
                f.type = static_cast<field_type>(md->getColumnType(i));
                f.is_unsigned = !md->isSigned(i);
                f.index = i;
                m_field_index.emplace(f.name, i);
                m_fields.push_back(std::move(f));
            }
        }

        result::~result() {
            if (m_handle) {
                auto rs = static_cast<sql::ResultSet*>(m_handle);
                try {
                    rs->close();
                }
                catch (const std::exception& e) {
                    std::cerr << make_throw_error(e.what()) << std::endl;
                }
                delete rs;
                m_handle = nullptr;
            }
        }

        uint32 result::field_index(const std::string& name) const {
            auto it = m_field_index.find(name);
            if (it == m_field_index.end()) {
                throw exception("Field not found: " + name);
            }
            return it->second;
        }

#define ENSURE_RS() if (!m_handle) throw exception("Invalid ResultSet")

        std::string result::field_name(uint32 idx) {
            if (idx < 1 || idx > m_fields.size()) return "";
            return m_fields[idx - 1].name;
        }

        uint32 result::field_count() {
            return static_cast<uint32>(m_fields.size());
        }

        size_t result::row_count() {
            ENSURE_RS();
            try {
                return static_cast<sql::ResultSet*>(m_handle)->rowsCount();
            }
            catch (const sql::SQLException& e) {
                throw exception(make_throw_error(e.what()));
            }
        }

        bool result::next() {
            ENSURE_RS();
            try {
                return static_cast<sql::ResultSet*>(m_handle)->next();
            }
            catch (const sql::SQLException& e) {
                throw exception(make_throw_error(e.what()));
            }
        }

        ylib::mysql::data result::read_data(uint32 index) const {
            data d;
            if (index < 1 || index > m_fields.size()) {
                throw exception("Field index out of range");
            }
            auto rs = static_cast<sql::ResultSet*>(m_handle);
            const field& f = m_fields[index - 1];
            d.m_type = f.type;

            if (rs->isNull(index)) {
                d.m_hold = data::hold::null;
                return d;
            }

            switch (f.type) {
            case field_type::BIT:
                d.m_hold = data::hold::boolean;
                d.m_bool = rs->getBoolean(index);
                break;
            case field_type::TINYINT:
            case field_type::SMALLINT:
            case field_type::MEDIUMINT:
            case field_type::INTEGER:
            case field_type::YEAR:
                if (f.is_unsigned) {
                    d.m_hold = data::hold::u64;
                    d.m_u64 = rs->getUInt64(index);
                }
                else {
                    d.m_hold = data::hold::i64;
                    d.m_i64 = rs->getInt64(index);
                }
                break;
            case field_type::BIGINT:
                if (f.is_unsigned) {
                    d.m_hold = data::hold::u64;
                    d.m_u64 = rs->getUInt64(index);
                }
                else {
                    d.m_hold = data::hold::i64;
                    d.m_i64 = rs->getInt64(index);
                }
                break;
            case field_type::REAL:
            case field_type::DOUBLE:
            case field_type::DECIMAL:
            case field_type::NUMERIC:
                d.m_hold = data::hold::f64;
                d.m_f64 = rs->getDouble(index);
                break;
            case field_type::BINARY:
            case field_type::VARBINARY:
            case field_type::LONGVARBINARY:
            case field_type::GEOMETRY:
                d.m_hold = data::hold::blob;
                d.m_blob = read_blob_stream(rs->getBlob(index));
                break;
            case field_type::CHAR:
            case field_type::VARCHAR:
            case field_type::LONGVARCHAR:
            case field_type::TIMESTAMP:
            case field_type::DATE:
            case field_type::TIME:
            case field_type::ENUM:
            case field_type::SET:
            case field_type::JSON:
            case field_type::SQLNULL:
            case field_type::UNKNOWN:
            default:
                d.m_hold = data::hold::str;
                d.m_str = rs->getString(index).c_str();
                break;
            }
            return d;
        }

        ylib::mysql::data result::get(uint32 index) {
            ENSURE_RS();
            try {
                return read_data(index);
            }
            catch (const sql::SQLException& e) {
                throw exception(make_throw_error(e.what()));
            }
        }

        ylib::mysql::data result::get(const std::string& name) {
            ENSURE_RS();
            try {
                return read_data(field_index(name));
            }
            catch (const sql::SQLException& e) {
                throw exception(make_throw_error(e.what()));
            }
        }

        ylib::mysql::row result::get_row() {
            ENSURE_RS();
            try {
                row r;
                r.m_fields = m_fields;
                r.m_values.reserve(m_fields.size());
                r.m_name_index.reserve(m_fields.size());
                for (size_t i = 0; i < m_fields.size(); ++i) {
                    r.m_values.push_back(read_data(static_cast<uint32>(i + 1)));
                    r.m_name_index.emplace(m_fields[i].name, static_cast<uint32>(i));
                }
                return r;
            }
            catch (const sql::SQLException& e) {
                throw exception(make_throw_error(e.what()));
            }
        }

        ylib::json result::row_to_json() {
            return get_row().to_json();
        }

#undef ENSURE_RS

    } // namespace mysql
} // namespace ylib

#endif // _WIN64 || __linux__