/**
 * Tencent is pleased to support the open source community by making Tars available.
 *
 * Copyright (C) 2016THL A29 Limited, a Tencent company. All rights reserved.
 *
 * Licensed under the BSD 3-Clause License (the "License"); you may not use this file except 
 * in compliance with the License. You may obtain a copy of the License at
 *
 * https://opensource.org/licenses/BSD-3-Clause
 *
 * Unless required by applicable law or agreed to in writing, software distributed 
 * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR 
 * CONDITIONS OF ANY KIND, either express or implied. See the License for the 
 * specific language governing permissions and limitations under the License.
 */

#include "NotifyImp.h"
#include "jmem/jmem_hashmap.h"
#include "NotifyServer.h"

extern TC_Config * g_pconf;
extern TarsHashMap<NotifyKey, NotifyInfo, ThreadLockPolicy, FileStorePolicy> * g_notifyHash;

void NotifyImp::loadconf()
{
    TC_DBConf tcDBConf;
    tcDBConf.loadFromMap(g_pconf->getDomainMap("/tars/db"));

    _mysqlConfig.init(tcDBConf);

    _sql = (*g_pconf)["/tars/<sql>"];
    _maxPageSize = TC_Common::strto<size_t>((*g_pconf)["/tars/hash<max_page_size>"]);
    _maxPageNum  = TC_Common::strto<size_t>((*g_pconf)["/tars/hash<max_page_num>"]);

    map<string, string> m = g_pconf->getDomainMap("/tars/filter"); 
    for(map<string,string>::iterator it= m.begin();it != m.end();it++)
    {
        vector<string> vFilters = TC_Common::sepstr<string>(it->second,";|");
        _setFilter[it->first].insert(vFilters.begin(),vFilters.end()); 
    }
    
}

void NotifyImp::initialize()
{
    loadconf();
}

bool NotifyImp::IsdbTableExist(const string& sTbName)
{
    try
    {
        TC_Mysql::MysqlData tTotalRecord = _mysqlConfig.queryRecord("show tables like '%"+sTbName+"%'");

        TLOGINFO("NotifyImp::IsdbTableExist show tables like '%" + sTbName + "%|affected:" << tTotalRecord.size() << endl);

        if (tTotalRecord.size() > 0)
        {
            return true;
        }
        else
        {
            TLOGERROR("NotifyImp::IsdbTableExist sTbName:" << sTbName << "|doesn't exist" << endl);
            return false;
        }

    }
    catch(TC_Mysql_Exception& ex)
    {
        TLOGERROR("NotifyImp::IsdbTableExist exception:" << ex.what() << endl);
        return false;
    }
    catch(...)
    {
        TLOGERROR("NotifyImp::IsdbTableExist unknown exception." << endl);
        return false;
    }
}

void NotifyImp::creatTb(const string &sTbName)
{
    try
    {
        if (!IsdbTableExist(sTbName))
        {
            string sSql = TC_Common::replace(_sql, "${TABLE}",sTbName);

            _mysqlConfig.execute(sSql);

            TLOGDEBUG("NotifyImp::creatTb sSql:" << sSql << endl);
        }
    }
    catch (TC_Mysql_Exception& ex)
    {
        TLOGERROR("NotifyImp::creatTb exception:" << ex.what() << endl);
    }
    catch(...)
    {
        TLOGERROR("NotifyImp::IsdbTableExist unknown exception." << endl);
    }
}

bool NotifyImp::IsNeedFilte(const string& sServerName,const string& sResult)
{
    if(_setFilter.find(sServerName) != _setFilter.end())
    {
        set<string>& setFilter = _setFilter.find(sServerName)->second;
        for(set<string>::iterator it = setFilter.begin(); it != setFilter.end();it++)
        {
            if(sResult.find(*it) != string::npos)
            {
                return true;
            }
        }    
    }

    const string sDefault = "default";
    if(_setFilter.find(sDefault) != _setFilter.end())
    {
        set<string>& setFilter = _setFilter.find(sDefault)->second;
        for(set<string>::iterator it = setFilter.begin(); it != setFilter.end();it++)
        {
            if(sResult.find(*it) != string::npos)
            {
                return true;
            }
        }        
    }

    return false;
}

void NotifyImp::reportServer(const string& sServerName, const string& sThreadId, const string& sResult, tars::TarsCurrentPtr current)
{
    TLOGDEBUG("NotifyImp::reportServer sServerName:" << sServerName << "|ip:" << current->getIp() << "|sThreadId:" << sThreadId << "|sResult:" << sResult << endl);
    //DLOG << "NotifyImp::reportServer sServerName:" << sServerName << "|ip:" << current->getIp() << "|sThreadId:" << sThreadId << "|sResult:" << sResult << endl;

    if(IsNeedFilte(sServerName,sResult))
    {
        TLOGWARN("NotifyImp::reportServer sServerName:" << sServerName << "|ip:" << current->getIp() << "|sThreadId:" << sThreadId << "|sResult:" << sResult <<"|filted"<< endl);
        return;
    }
    
    ReportInfo info;
    info.sApp = sServerName;
    info.sServer = sServerName;

    vector<string> vModule = TC_Common::sepstr<string>(sServerName,".");
    if(vModule.size() >= 2)
    {
        info.sApp = vModule[0];
        info.sServer = vModule[1];
    }

    info.eType     = REPORT;
    info.sSet      = g_app.getLoadDbThread()->getSetName(sServerName + current->getIp());
    info.sMessage  = sResult;
    info.sThreadId = sThreadId;

    reportNotifyInfo(info, current);
    
    return;
}

void NotifyImp::notifyServer(const string& sServerName, NOTIFYLEVEL level, const string& sMessage, tars::TarsCurrentPtr current)
{
    ReportInfo info;
    info.sApp    = sServerName;
    info.sServer = sServerName;


    vector<string> vModule = TC_Common::sepstr<string>(sServerName, ".");
    if (vModule.size() >= 2)
    {
        info.sApp = vModule[0];
        info.sServer = vModule[1];
    }

    info.eType    = NOTIFY;
    info.sSet     = g_app.getLoadDbThread()->getSetName(sServerName + current->getIp());
    info.sMessage = sMessage;
    info.eLevel   = level;

    reportNotifyInfo(info, current);
    
    return;
}

int NotifyImp::getNotifyInfo(const tars::NotifyKey & stKey,tars::NotifyInfo &stInfo,tars::TarsCurrentPtr current)
{
    int iRet = g_notifyHash->get(stKey, stInfo);

    ostringstream os;
    stKey.displaySimple(os);
    os << "|";
    stInfo.displaySimple(os);

    TLOGDEBUG("NotifyImp::getNotifyInfo iRet:" << iRet << "|os:" << os.str() << endl);

    return iRet;
}

void NotifyImp::reportNotifyInfo(const tars::ReportInfo & info, tars::TarsCurrentPtr current)
{
    switch (info.eType)
    {
        case (REPORT):
        {
            TLOGDEBUG("NotifyImp::reportNotifyInfo reportServer:" << info.sApp + "." + info.sServer << "|sSet:" << info.sSet << "|sContainer:" << info.sContainer << "|ip:" << current->getIp() 
                << "|sThreadId:" << info.sThreadId << "|sMessage:" << info.sMessage << endl);

            //DLOG << "NotifyImp::reportNotifyInfo reportServer:" << info.sApp + "." + info.sServer << "|sSet:" << info.sSet << "|sContainer:" << info.sContainer << "|ip:" << current->getIp()
            //    << "|sThreadId:" << info.sThreadId << "|sMessage:" << info.sMessage << endl;

            if (IsNeedFilte(info.sApp + info.sServer, info.sMessage))
            {
                TLOGWARN("NotifyImp::reportNotifyInfo reportServer:" << info.sApp + "." + info.sServer << "|sSet:" << info.sSet << "|sContainer:" << info.sContainer << "|ip:" << current->getIp() 
                    << "|sThreadId:" << info.sThreadId << "|sMessage:" << info.sMessage << "|filted" << endl);

                return;
            }

            string sql;
            TC_Mysql::RECORD_DATA rd;

            rd["application"]    = make_pair(TC_Mysql::DB_STR, info.sApp);
            rd["server_name"]    = make_pair(TC_Mysql::DB_STR, info.sServer);
            rd["container_name"] = make_pair(TC_Mysql::DB_STR, info.sContainer);
            rd["server_id"]      = make_pair(TC_Mysql::DB_STR, info.sApp +"."+ info.sServer + "_" + current->getIp());
            rd["node_name"]      = make_pair(TC_Mysql::DB_STR, current->getIp());
            rd["thread_id"]      = make_pair(TC_Mysql::DB_STR, info.sThreadId);

            if (!info.sSet.empty())
            {
                vector<string> v = TC_Common::sepstr<string>(info.sSet, ".");
                if (v.size() != 3 || (v.size() == 3 && (v[0] == "*" || v[1] == "*")))
                {
                    TLOGERROR("NotifyImp::reportNotifyInfo bad set name:" << info.sSet << endl);
                }
                else
                {
                    rd["set_name"]  = make_pair(TC_Mysql::DB_STR, v[0]);
                    rd["set_area"]  = make_pair(TC_Mysql::DB_STR, v[1]);
                    rd["set_group"] = make_pair(TC_Mysql::DB_STR, v[2]);
                }
                
            }
            else
            {
                rd["set_name"]  = make_pair(TC_Mysql::DB_STR, "");
                rd["set_area"]  = make_pair(TC_Mysql::DB_STR, "");
                rd["set_group"] = make_pair(TC_Mysql::DB_STR, "");
            }
            
            rd["result"]     = make_pair(TC_Mysql::DB_STR, info.sMessage);
            rd["notifytime"] = make_pair(TC_Mysql::DB_INT, "now()");
            string sTable = "t_server_notifys";
            try
            {
                _mysqlConfig.insertRecord(sTable, rd);
            }
            catch (TC_Mysql_Exception& ex)
            {
                string err = string(ex.what());
                if (std::string::npos != err.find("doesn't exist"))
                {
                    creatTb(sTable);
                }
                else
                {
                    string sInfo = string("insert2Db exception") + "|" + ServerConfig::LocalIp + "|" + ServerConfig::Application + "." + ServerConfig::ServerName;
                    TARS_NOTIFY_ERROR(sInfo);
                }
                TLOGERROR("NotifyImp::reportNotifyInfo insert2Db exception:" << ex.what() << endl);
            }
            catch (exception& ex)
            {
                TLOGERROR("NotifyImp::reportNotifyInfo insert2Db exception:" << ex.what() << endl);
                string sInfo = string("insert2Db exception") + "|" + ServerConfig::LocalIp + "|" + ServerConfig::Application + "." + ServerConfig::ServerName;
                //TARS_NOTIFY_ERROR(sInfo);
            }
        }
        case (NOTIFY):
        {
            TLOGDEBUG("NotifyImp::reportNotifyInfo  notifyServer:" << info.sApp + "." + info.sServer << "|sSet:" << info.sSet << "|sContainer:" << info.sContainer << "|ip:" << current->getIp() 
                << "|eLevel:" << tars::etos(info.eLevel) << "|sMessage:" << info.sMessage << endl);

#if 0 //修正notify打日志逻辑，避免磁盘被打满
            if (info.eLevel == NOTIFYERROR)
            {
                FDLOG("NOTIFYERROR") << "NotifyImp::reportNotifyInfo  notifyServer:" << info.sApp + "." + info.sServer << "|sSet:" << info.sSet << "|sContainer:" << info.sContainer << "|ip:" << current->getIp() 
                    << "|eLevel:" << tars::etos(info.eLevel) << "|sMessage:" << info.sMessage << endl;
            }
            else
            {
                DLOG << "NotifyImp::reportNotifyInfo  notifyServer:" << info.sApp + "." + info.sServer << "|sSet:" << info.sSet << "|sContainer:" << info.sContainer << "|ip:" << current->getIp() 
                    << "|eLevel:" << tars::etos(info.eLevel) << "|sMessage:" << info.sMessage << endl;
            }
#endif

            string sServerId = info.sApp + info.sServer + "_" + current->getIp();

            NotifyKey stKey0;
            stKey0.name = info.sApp + info.sServer;
            stKey0.ip   = current->getIp();
            stKey0.page = 0;

            NotifyInfo stInfo0;

            NotifyItem stItem;
            stItem.sTimeStamp = TC_Common::now2str("%Y-%m-%d %H:%M:%S");
            stItem.sServerId  = sServerId;
            stItem.iLevel     = info.eLevel;
            stItem.sMessage   = info.sMessage;

            int iRet;
            iRet = g_notifyHash->get(stKey0, stInfo0);
            TLOGDEBUG("get " << sServerId << " page " << stKey0.page << " info return :" << iRet << endl);
            if (iRet == TC_HashMap::RT_LOAD_DATA_ERR)
            {
                return;
            }

            if (stInfo0.notifyItems.size() < _maxPageSize)
            {
                stInfo0.notifyItems.push_back(stItem);
                iRet = g_notifyHash->set(stKey0, stInfo0);
                TLOGDEBUG("NotifyImp::reportNotifyInfo set sServerId:" << sServerId << "|page:" << stKey0.page << "|iRet:" << iRet << endl);
                return;
            }

            //0页置换出去
            NotifyKey stKeyReplPage = stKey0;
            stKeyReplPage.page = (stInfo0.nextpage + 1) % _maxPageNum;
            if (stKeyReplPage.page == 0)
            {
                stKeyReplPage.page = 1;
            }
            iRet = g_notifyHash->set(stKeyReplPage, stInfo0);

            //修改0页
            stInfo0.nextpage = stKeyReplPage.page;
            stInfo0.notifyItems.clear();
            stInfo0.notifyItems.push_back(stItem);
            iRet = g_notifyHash->set(stKey0, stInfo0);
        }
        default:
            break;
    }

    return;
}