/* =================================================================
 Copyright (C) 2016 BizStation Corp All rights reserved.

 This program is free software; you can redistribute it and/or
 modify it under the terms of the GNU General Public License
 as published by the Free Software Foundation; either version 2
 of the License, or (at your option) any later version.

 This program is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 GNU General Public License for more details.

 You should have received a copy of the GNU General Public License
 along with this program; if not, write to the Free Software
 Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
 02111-1307, USA.
 ================================================================= */
/* MySQL 5.5 and Mariadb are no-rtti */
#include <my_config.h>
# include <string>

#ifdef __GNUC__
# pragma implementation "mysqlInternal.h" 
#endif

#include "mysqlProtocol.h"

#include <bzs/env/crosscompile.h>
#include <bzs/db/protocol/tdap/tdapcapi.h>
#include <bzs/db/IBlobBuffer.h>
#include "mysqlThd.h"

#if defined(USE_BINLOG_VAR) && (!defined(MARIADB_BASE_VERSION) &&  MYSQL_VERSION_ID > 50600)
#  include "sql/binlog.h"
#  include "rpl_gtid.h"
#endif

//----------------------------------------------------------------------
//  Implement dummyProtocol
//----------------------------------------------------------------------
#if defined(MYSQL_5_7)
#  define Protocol_mysql Protocol
#  define CP_PROTOCOL PROTOCOL_PLUGIN
#else
#  define Protocol_mysql Protocol
#  define CP_PROTOCOL PROTOCOL_BINARY
#endif


#pragma GCC diagnostic ignored "-Woverloaded-virtual"

class dummyProtocol : public Protocol_mysql
{
    THD* m_thd;
    Protocol_mysql* m_backup;
    
public:
#if defined(MYSQL_5_7)
    inline dummyProtocol(THD *thd_arg) : Protocol_mysql()
    {
        m_thd = thd_arg;
        m_backup = m_thd->get_protocol();
        m_thd->set_protocol(this);
    }
    inline virtual ~dummyProtocol()
    {
        m_thd->set_protocol(m_backup);    
    }
#else
    inline dummyProtocol(THD *thd_arg) : Protocol_mysql(thd_arg)
    {
        m_thd = thd_arg;
        m_backup = m_thd->protocol;
        m_thd->protocol = this;
    }
    inline virtual ~dummyProtocol()
    {
        m_thd->protocol = m_backup;       
    }
#endif
    bool send_result_set_metadata(List<Item> *list, uint flags){return false;}
    virtual bool write(){return false;};
    virtual void prepare_for_resend(){}
    virtual bool store_null(){return false;}
    virtual bool store_tiny(longlong from){return store_longlong(from, false);}
    virtual bool store_short(longlong from){return store_longlong(from, false);}
    virtual bool store_long(longlong from){return store_longlong(from, false);}
    virtual bool store_decimal(const my_decimal *){return false;}
    virtual bool store(float from, uint32 decimals, String *buffer){return false;}
    virtual bool store(double from, uint32 decimals, String *buffer){return false;}
    virtual bool store(MYSQL_TIME *time, uint precision){return false;}
    virtual bool store_date(MYSQL_TIME *time){return false;}
    virtual bool store_time(MYSQL_TIME *time, uint precision){return false;}
    virtual bool store(Field *field){return false;}
    virtual bool store(const char *from, size_t length, const CHARSET_INFO *fromcs,
                     const CHARSET_INFO* /*tocs*/){return false;}

    virtual bool send_out_parameters(List<Item_param> *sp_params){return false;}
    virtual Protocol::enum_protocol_type type(void){ return CP_PROTOCOL; };
#ifdef MARIADB_BASE_VERSION      //Mariadb 5.5 10.0 10.1
    virtual bool store(MYSQL_TIME *time, int decimals){return false;}
    virtual bool store_time(MYSQL_TIME *time, int decimals){ return false;}
#elif defined(MYSQL_5_5)
    virtual bool store_time(MYSQL_TIME *time){return true;};
    virtual bool store(MYSQL_TIME *time){return true;}
    virtual bool store(const char *from, size_t length, 
    CHARSET_INFO *fromcs, CHARSET_INFO *tocs){return false;}
#elif defined(MYSQL_5_7) 
    bool store_decimal(const my_decimal *, uint, uint){ return true; }
    bool store(Proto_field *){ return true; }
    void start_row(){}
    int read_packet(void){ return 0; }
    int get_command(COM_DATA *, enum_server_command *){ return m_thd->lex->sql_command; }
    enum_vio_type connection_type(void){ return VIO_TYPE_PLUGIN; }
    ulong get_client_capabilities(void){ return 0; }
    bool has_client_capability(unsigned long){ return false; }
    bool connection_alive(void){ return false; }
    bool end_row(void){ return false; }
    void abort_row(void){}
    void end_partial_result_set(void){}
    int shutdown(bool){ return 0; }
    SSL_handle get_ssl(void){ return NULL; }
    uint get_rw_status(void){ return 0; }
    bool get_compression(void){ return false; }
    bool start_result_metadata(uint, uint, const CHARSET_INFO *){ return false; }
    bool send_field_metadata(Send_field *, const CHARSET_INFO *){ return false; }
    bool end_result_metadata(void){ return false; }
    bool send_ok(uint, uint, ulonglong, ulonglong, const char *){ return false; }
    bool send_eof(uint, uint){ return false; }
    bool send_error(uint, const char *, const char *){ return false; }
#endif
};

//----------------------------------------------------------------------
//  class masterStatus for windows mysql 5.6 only
//----------------------------------------------------------------------
#if defined(NOTUSE_BINLOG_VAR)
class masterStatus : public dummyProtocol
{
    binlogPos* m_bpos;
    bzs::db::IblobBuffer* m_bb;
    int m_index;
    
public:
    inline masterStatus(THD *thd_arg, binlogPos* bpos, bzs::db::IblobBuffer* bb) : 
        dummyProtocol(thd_arg), m_bpos(bpos), m_bb(bb), m_index(0) {}
    bool store_longlong(longlong from, bool unsigned_flag)
    {
        m_bpos->pos = (ulonglong)from;
        m_bpos->type = REPL_POSTYPE_POS;
        return false;
    }

#if (MYSQL_VERSION_ID < 50600 || defined(MARIADB_BASE_VERSION)) // mariadb 5.5 
    bool store(const char *from, size_t length, CHARSET_INFO *cs)
    {
        if (m_index == 0)
            strncpy(m_bpos->filename, from, BINLOGNAME_SIZE); 
        ++m_index;
        return false;
    }
#else
    bool store(const char *from, size_t length, const CHARSET_INFO *cs)
    {
        if (m_index == 0)
            strncpy(m_bpos->filename, from, BINLOGNAME_SIZE);
        else if (m_index == 3)
        {
            if (length)
            {
                m_bpos->type = REPL_POSTYPE_GTID;
                m_bb->addBlob((unsigned int)length + 1, 0, (const unsigned char *)from);
                m_bb->setFieldCount(1);
            }
        }
        ++m_index;
        return false;
    }
#endif
};
#endif // NOTUSE_BINLOG_VAR

//----------------------------------------------------------------------
//  class safe_commit_lock  commit lock for binlogPos
//----------------------------------------------------------------------
safe_commit_lock::safe_commit_lock(THD* thd): m_thd(thd), m_commits_lock(NULL)
{

}

bool safe_commit_lock::lock()
{
    if (m_thd)
    {
        MDL_request mdl_request;
        #if ((MYSQL_VERSION_NUM > 50700) && !defined(MARIADB_BASE_VERSION))
        mdl_request.init_with_source(MDL_key::COMMIT, "", "", MDL_SHARED, MDL_EXPLICIT, __FILE__, __LINE__);
        #else
        mdl_request.init(MDL_key::COMMIT, "", "", MDL_SHARED, MDL_EXPLICIT);
        #endif
        if (m_thd->mdl_context.acquire_lock(&mdl_request,
                                    m_thd->variables.lock_wait_timeout))
            return false;
        m_commits_lock = mdl_request.ticket;
    }
    return true;
}

safe_commit_lock::~safe_commit_lock()
{
    if (m_commits_lock)
    {
        m_thd->mdl_context.release_lock(m_commits_lock);
        m_commits_lock= NULL;
    }
}


#ifdef NOTUSE_BINLOG_VAR //Only MySQL 5.6 Windows
    inline short getBinlogPosInternal(THD* currentThd, binlogPos* bpos, THD* tmpThd, bzs::db::IblobBuffer* bb)
    {
        short result = 0;
        {
            attachThd(tmpThd);
            if (tmpThd != currentThd)
                copyGrant(tmpThd, currentThd, NULL);
            masterStatus p(tmpThd, bpos, bb); 
            cp_query_command(tmpThd, "show master status");
            if (tmpThd->is_error())
                result = tmpThd->cp_get_sql_error();
            cp_lex_clear(tmpThd);
        }
        attachThd(currentThd);
        return result;
    }
#endif

#ifdef USE_BINLOG_GTID //MariaDB 10-
    inline short getBinlogPosInternal(THD* currentThd, binlogPos* bpos, THD* /*tmpThd*/, bzs::db::IblobBuffer* /*bb*/)
    {
        if (mysql_bin_log.is_open())
        {
            rpl_gtid gtid;
            bpos->type = REPL_POSTYPE_MARIA_GTID;
            if (mysql_bin_log.lookup_domain_in_binlog_state(currentThd->variables.gtid_domain_id,  &gtid))
            {
                sprintf_s(bpos->gtid, GTID_SIZE, "%u-%u-%llu", gtid.domain_id, gtid.server_id, gtid.seq_no); 
                size_t dir_len = dirname_length(mysql_bin_log.get_log_fname());
                strncpy(bpos->filename, mysql_bin_log.get_log_fname() + dir_len, BINLOGNAME_SIZE);
                bpos->pos = my_b_tell(mysql_bin_log.get_log_file());
                bpos->filename[BINLOGNAME_SIZE-1] = 0x00;
            }
        }
        return 0;
    }
#endif

#ifdef USE_BINLOG_VAR
    inline void readGtid(binlogPos* bpos, bzs::db::IblobBuffer* bb)
    {
        #if (MYSQL_VERSION_ID > 50600)
        char p[1024] = { 0 };
        char* buf = p;
        global_sid_lock->wrlock();
        const Gtid_set* gtid_set = gtid_state->cp_get_executed_gtids();
        int size = gtid_set->get_string_length();
        if (size)
        {
            ++size;
            if (size > 1024) buf = new char[size];
            gtid_set->to_string(buf);
            bpos->type = REPL_POSTYPE_GTID;
            bb->addBlob((unsigned int)size, TD_EXTENDED_VAR_BINLOG_GTID, (const unsigned char *)buf);
            bb->setFieldCount(1);
            if (p != buf) delete[] buf;
        }
        global_sid_lock->unlock();
        #endif
    }

    // Linux MySQL can access to the mysql_bin_log variable
    inline short getBinlogPosInternal(THD*, binlogPos* bpos, THD* /*tmpThd*/, bzs::db::IblobBuffer* bb)
    {
        if (mysql_bin_log.is_open())
        {
            size_t dir_len = dirname_length(mysql_bin_log.get_log_fname());
            strncpy(bpos->filename, mysql_bin_log.get_log_fname() + dir_len, BINLOGNAME_SIZE);
            bpos->pos = my_b_tell(mysql_bin_log.get_log_file());
            bpos->filename[BINLOGNAME_SIZE-1] = 0x00;
            bpos->type = REPL_POSTYPE_POS;
            readGtid(bpos, bb);
        }
        return 0;
    }
#endif //USE_BINLOG_VAR

short getBinlogPos(THD* thd, binlogPos* bpos, THD* tmpThd, bzs::db::IblobBuffer* bb)
{
    #ifndef NOTUSE_BINLOG_VAR
    safe_mysql_mutex_lock lck(mysql_bin_log.get_log_lock());
    #endif
    return getBinlogPosInternal(thd, bpos, tmpThd, bb);
}

int execSql(THD* thd, const char* sql)
{
    thd->variables.lock_wait_timeout = OPEN_TABLE_TIMEOUT_SEC;
    thd->clear_error();
    int result = cp_query_command(thd, (char*)sql);
    if (thd->is_error())
        result = thd->cp_get_sql_error();
    cp_lex_clear(thd); // reset values for insert
    return result;
}

//----------------------------------------------------------------------
//  slaveStatus  
//----------------------------------------------------------------------
using namespace bzs::db::transactd;
class slaveStatus : public dummyProtocol
{
protected:
    connection::records& m_records;
    bzs::db::IblobBuffer* m_bb;
    int m_blobfields;
    connection::record& getRec()
    {
        m_records.push_back(connection::record());
        size_t id = m_records.size() - 1;
        connection::record& rec = m_records[id];
        rec.id = (unsigned int)id;
        return rec;
    }
public:
    inline slaveStatus(THD *thd_arg, connection::records& recs, bzs::db::IblobBuffer* bb) : 
        dummyProtocol(thd_arg), m_records(recs), m_bb(bb), m_blobfields(0)
    {
        m_records.clear();   
    }
    virtual ~slaveStatus()
    {
        m_bb->setFieldCount(m_blobfields);
    }
    bool store_longlong(longlong from, bool unsigned_flag)
    {
        connection::record& rec = getRec();
        rec.type = 0;
        rec.longValue = from; 
        return false;
    }
    bool store_null()
    {
        connection::record& rec = getRec();
        rec.type = 1;
        strncpy(rec.name, "NULL", CON_REC_VALUE_SIZE); 
        return false;
    }
    bool str_store(const char *from, size_t length)
    {
        connection::record& rec = getRec();
        rec.type = 1;
        if (length && from)
        {
            if (length >= CON_REC_VALUE_SIZE)
            {
                ++m_blobfields;
                rec.type = 2;
                m_bb->addBlob((unsigned int)length, (unsigned short)(m_records.size() -1),
                        (const unsigned char*)from);
            }else
                strncpy(rec.name, from, CON_REC_VALUE_SIZE);
        }
        return false;
    }

#if (MYSQL_VERSION_ID < 50600 || defined(MARIADB_BASE_VERSION)) // MySQL 5.5 
    bool store(const char *from, size_t length, CHARSET_INFO *cs)
    {
        return str_store(from, length);
    }
#else
    bool store(const char *from, size_t length, const CHARSET_INFO *cs)
    {
        return str_store(from, length);
    }
#endif
};

//----------------------------------------------------------------------
//  slaveHosts  
//----------------------------------------------------------------------
// rec.id = server_id
// rec.readCount = port
// rec.name = host
// opt_show_slave_auth_info
class slaveInfo
{
public:
    std::string host;
    std::string user;
    std::string passwd;
    std::string uuid;
    void reset()
    {
        host = "";
        user = "";
        passwd = "";
        uuid = "";

    }
    std::string tabSeparateValue()
    {
        return host + "\t" + user + "\t" + passwd + "\t" + uuid;
    }
};


class slaveHosts : public slaveStatus
{
    connection::record m_rec;
    slaveInfo m_sinfo;
    int m_strCount;
public:
    inline slaveHosts(THD *thd_arg, connection::records& recs, bzs::db::IblobBuffer* bb) :
        slaveStatus(thd_arg, recs, bb), m_strCount(0){}
  
    bool store_longlong(longlong from, bool unsigned_flag)
    {
        if (m_rec.id == 0)
            m_rec.id = (unsigned int)from;
        else if (m_rec.readCount == 0)
            m_rec.readCount = (unsigned int)from;
        return false;
    }
    bool store_null()
    {
        return false;
    }
    bool str_store(const char *from, size_t length)
    {
        if (m_strCount == 0)
            m_sinfo.host = from; 
#if (defined(MYSQL_5_7))
        else if (opt_show_slave_auth_info)
        {
            if (m_strCount == 1)
                m_sinfo.user = from;
            else if (m_strCount == 2)
                m_sinfo.passwd = from;
            else if (m_strCount == 3)
                m_sinfo.uuid = from;
        }
#endif
        else if (m_strCount == 1)
            m_sinfo.uuid = from;

        ++m_strCount;
        return false;
    }

    void addHostInfo()
    {
        std::string s = m_sinfo.tabSeparateValue();
        m_rec.type = 1;
        if (s.size() >= CON_REC_VALUE_SIZE)
        {
            ++m_blobfields;
            m_rec.type = 2;
            m_bb->addBlob((unsigned int)s.size() + 1, (unsigned short)m_records.size(),
                (const unsigned char*)s.c_str());
        }
        else
            strncpy(m_rec.name, s.c_str(), CON_REC_VALUE_SIZE);
    }

#if (MYSQL_VERSION_ID < 50600 || defined(MARIADB_BASE_VERSION)) // MySQL 5.5 
    bool store(const char *from, size_t length, CHARSET_INFO *cs)
    {
        return str_store(from, length);
    }
    bool write()
    {
        addHostInfo();
        m_records.push_back(m_rec);
        m_rec.reset();
        m_sinfo.reset();
        m_strCount = 0;
        return false;
    }
#else
    bool store(const char *from, size_t length, const CHARSET_INFO *cs)
    {
        return str_store(from, length);
    }
    bool end_row()
    {
        if (m_rec.id)
        {
            addHostInfo();
            m_records.push_back(m_rec);
        }
        m_rec.reset();
        m_sinfo.reset();
        m_strCount = 0;
        return false;
    }
#endif
};


//----------------------------------------------------------------------
//  channel  
//----------------------------------------------------------------------
#pragma warning(disable : 4800)
#pragma warning(disable : 4271)
#pragma warning(disable : 4267)
#pragma warning(disable : 4244)
#if (defined(MARIADB_10_0) || defined(MARIADB_10_1))
#include "sql/slave.h"
#include "sql/rpl_mi.h"


int getChannels(THD* /*thd*/, connection::records& recs)
{
    mysql_mutex_assert_owner(&LOCK_active_mi);
    if (master_info_index)
    {
        uint_td size = master_info_index->master_info_hash.records;
        for (uint_td i = 0;i < size; ++i)
        {
            Master_info* mi = (Master_info*)my_hash_element(&master_info_index->master_info_hash, i);
            if (mi->host[0])
            {
                connection::record rec;
                rec.type = 1;
                strcpy_s(rec.name, CON_REC_VALUE_SIZE, mi->connection_name.str);
                recs.push_back(rec);
            }
        }
    }
    return 0;
}
#elif defined(MYSQL_5_7)
#include "sql/rpl_msr.h" //channel_map
int getChannels(THD* thd, connection::records& recs)
{
    channel_map.rdlock();
    uint_td size = channel_map.get_num_instances();
    if (size)
    {
        for (mi_map::iterator it = channel_map.begin(); it != channel_map.end(); it++)
        {
            Master_info* mi = it->second;
            if (mi && mi->host[0])
            {
                connection::record rec;
                rec.type = 1;
                strcpy_s(rec.name, CON_REC_VALUE_SIZE, mi->get_channel());
                recs.push_back(rec);
            }
        }
    }
    channel_map.unlock();
    return 0;
}

#else
int getChannels(THD* thd, connection::records& recs)
{
    return 0;
}
#endif
#pragma warning(default : 4800)
#pragma warning(default : 4271)
#pragma warning(default : 4267)
#pragma warning(default : 4244)
//------------------------------------------------------------------------------

int getSlaveStatus(THD* thd, const char* channel, connection::records& recs, bzs::db::IblobBuffer* bb)
{
    slaveStatus ss(thd, recs, bb);
    char tmp[128];
#if (defined(MYSQL_5_7))
    sprintf_s(tmp, 128, "show slave status for channel '%s'", channel);
#elif (defined(MARIADB_10_1) || defined(MARIADB_10_0))
    sprintf_s(tmp, 128, "show slave '%s' status", channel);
#else
    sprintf_s(tmp, 128, "show slave status");
#endif
    return execSql(thd, tmp);
}

int getSlaveHosts(THD* thd, connection::records& recs, bzs::db::IblobBuffer* bb)
{
    slaveHosts ss(thd, recs, bb);
    return execSql(thd, "show slave hosts");
}

#pragma GCC diagnostic warning "-Woverloaded-virtual"


//----------------------------------------------------------------------
//  database list
//----------------------------------------------------------------------
inline void appenDbList(connection::records& recs, LEX_STRING* db_name)
{
    recs.push_back(connection::record());
    connection::record& rec = recs[recs.size() - 1];
    strncpy(rec.name, db_name->str, 64);
    rec.name[64] = 0x00;
}

void readDbList(THD* thd, connection::records& recs)
{
    SQL_Strings files;
    db_list(thd, &files);
#if (defined(MARIADB_10_0) || defined(MARIADB_10_1))
    for (int i = 0; i < (int)files.elements(); ++i)
        appenDbList(recs, files.at(i));
#else
    List_iterator_fast<LEX_STRING> it(files);
    LEX_STRING* db_name;
    while ((db_name = it++))
        appenDbList(recs, db_name);
#endif
}

//----------------------------------------------------------------------
//  security
//----------------------------------------------------------------------

bool setGrant(THD* thd, const char* host, const char* user,  const char* db)
{
    // sctx->master_access and sctx->db_access
    return (acl_getroot(cp_security_ctx(thd), cp_strdup(user, MYF(0)),
    cp_strdup(host, MYF(0)), cp_strdup(host, MYF(0)), (char*)db)) == false;
}

bool copyGrant(THD* thd, THD* thdSrc, const char* db)
{
    Security_context* sctx = cp_security_ctx(thdSrc);
    if (sctx->cp_master_accsess() == (ulong)~NO_ACCESS)
    {
        cp_security_ctx(thd)->skip_grants();
        return true;
    }
    return setGrant(thd, sctx->cp_priv_host(), sctx->cp_priv_user(), db);
}

void setDbName(THD* thd, const char* name)
{
    cp_set_db(thd, name);
}

/*
class safe_global_read_lock
{
    THD* m_thd;
public:
    safe_global_read_lock(THD* thd): m_thd(thd){}
    bool lock()
    {
        if (m_thd->global_read_lock.lock_global_read_lock(m_thd))
        {
            m_thd = NULL;
            return false;
        }
        #ifdef NOTUSE_BINLOG_VAR
        close_cached_tables(NULL, NULL, FALSE , 50000000L);
        if (m_thd->global_read_lock.make_global_read_lock_block_commit(m_thd))
        {
            m_thd->global_read_lock.unlock_global_read_lock(m_thd);
            m_thd = NULL;
            return false;
        }
        #endif
        return true;
    }
    ~safe_global_read_lock()
    {
        if (m_thd)
            m_thd->global_read_lock.unlock_global_read_lock(m_thd);
    }
};*/