///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoLogManager.cc
// -----------------
// Cego database log manager implementation
//                                                         
// Design and Implementation by Bjoern Lemke               
//                                                         
// (C)opyright 2000-2010 Bjoern Lemke                        
//
// 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, 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; see the file COPYING.  If not, write to
// the Free Software Foundation, 59 Temple Place - Suite 330,
// Boston, MA 02111-1307, USA.
//
// IMPLEMENTATION MODULE
//
// Class: CegoLogManager
// 
// Description: Cego database logging
//
// Status: QG-2.6
//
///////////////////////////////////////////////////////////////////////////////

// base includes
#include <lfc/Exception.h>
#include <lfc/File.h>
#include <lfc/Host.h>
#include <lfc/Net.h>
#include <lfc/Datetime.h>

// cego includes
#include "CegoLogManager.h"
#include "CegoXMLdef.h"
#include "CegoDefs.h"

#include <string.h>
#include <stdlib.h>

CegoLogManager::CegoLogManager(const Chain& xmlDef, const Chain& logFile) : CegoFileHandler(logFile), CegoXMLSpace(xmlDef)
{
    for (int i=0;i< TABMNG_MAXTABSET;i++)
    {
	_lsn[i]=0;
	_pLog[i]=0;
	_logActive[i] = false;
	_pLogHandler[i] = 0;
	_pNetHandler[i] = 0;
    }
    _modId = getModId("CegoLogManager");
}

CegoLogManager::~CegoLogManager()
{
    for (int i=0;i< TABMNG_MAXTABSET;i++)
    {
	if (_pLog[i])
	{
	    _pLog[i]->close();
	    delete _pLog[i];
	}
    }
}

void CegoLogManager::setLSN(int tabSetId, long lsn)
{
    _lsn[tabSetId] = lsn;
}

long CegoLogManager::getLSN(int tabSetId)
{
    return _lsn[tabSetId];
}

Chain CegoLogManager::getArchiveLogName(const Chain& tableSet, long lsn)
{
    Chain lsnStr = Chain("000000000000") + Chain(lsn);
    Chain lsnFix = lsnStr.subChain(lsnStr.length() - 12, lsnStr.length());
    return tableSet + "-" + lsnFix + ".dbf";
}

void CegoLogManager::setLogFile(int tabSetId, const Chain& logFile, bool readOnly)
{

    if (_pLog[tabSetId])
    {
	_pLog[tabSetId]->close();
	delete _pLog[tabSetId];
    }

    _logFile[tabSetId] = logFile;
    
    _pLog[tabSetId] = new File(logFile);

    if ( readOnly )
	_pLog[tabSetId]->open(File::READ);
    else
	_pLog[tabSetId]->open(File::READWRITE);

    _logSize[tabSetId] = _pLog[tabSetId]->Size();    
    int c = _pLog[tabSetId]->readByte((char*)&_logOffset[tabSetId], sizeof(int));	

    _logPos[tabSetId] = sizeof(int);
    _logActive[tabSetId] = false;

}

void CegoLogManager::allocateLogConnection(int tabSetId, const Chain& tableSet, const Chain& logHost, int logPort)
{

    Net n ( NETMNG_MSG_BUFLEN, NETMNG_SIZEBUFLEN );
    
    log(_modId, Logger::NOTICE, Chain("Connecting to loghost ") + logHost + Chain(":") + Chain(logPort) + Chain(" ..."));    

    try  {

	_pNetHandler[tabSetId] = n.connect(logHost, logPort);
        
    }
    catch ( Exception e )
    {
	Chain msg = Chain("Cannot connect to loghost ") + logHost + Chain(":") + Chain(logPort);
	log(_modId, Logger::LOGERR, msg); 
	throw Exception(EXLOC, msg); 
    }
    _pLogHandler[tabSetId] = new CegoLogHandler(this, _pNetHandler[tabSetId]);
    _pLogHandler[tabSetId]->requestLogSession(tableSet);
}

bool CegoLogManager::hasLogConnection(int tabSetId)
{
    if ( _pLogHandler[tabSetId] )
    {
	return true;
    }
    return false;
}

void CegoLogManager::releaseLogConnection(int tabSetId)
{

    if ( _pLogHandler[tabSetId] )
    { 
	
	try 
	{
	    _pLogHandler[tabSetId]->closeSession();
	}
	catch ( Exception e )
	{
	    log(_modId, Logger::LOGERR, Chain("Cannot close session to loghost, ignoring"));
	}
	delete _pLogHandler[tabSetId];
	_pLogHandler[tabSetId] = 0;
	
	delete _pNetHandler[tabSetId];
    }
}


void CegoLogManager::initLog(int tabSetId, int size)
{
    
    if (_pLog[tabSetId] == 0)
    {
	Chain msg = "No logfile set up for tableset";
	throw Exception(EXLOC, msg);
    }

    _pLog[tabSetId]->seek(0);

    char wBuf[LOGMNG_WBUFSIZE];

    _logOffset[tabSetId] = sizeof(int);
    
    _pLog[tabSetId]->writeByte((char*)&_logOffset[tabSetId], sizeof(int));
    
    int wBytes = sizeof(int);
    while (wBytes < size )
    {
	int n = wBytes + LOGMNG_WBUFSIZE > size ? size - wBytes : LOGMNG_WBUFSIZE;
	_pLog[tabSetId]->writeByte(wBuf, n);
	wBytes += n;
    }

    _pLog[tabSetId]->close();
    delete _pLog[tabSetId];
    _pLog[tabSetId]=0;

    
}

int CegoLogManager::getLogOffset(int tabSetId)
{
    return _logOffset[tabSetId];
}

void CegoLogManager::resetLog(int tabSetId)
{
    _logOffset[tabSetId] = sizeof(int);
    _pLog[tabSetId]->seek(0);
    _pLog[tabSetId]->writeByte((char*)&_logOffset[tabSetId], sizeof(int));    
}

void CegoLogManager::stopLog(int tabSetId)
{
    _logActive[tabSetId] = false;
}

void CegoLogManager::startLog(int tabSetId)
{
    if ( _pLog[tabSetId] != 0 )
    {
	_pLog[tabSetId]->seek(0);
	_pLog[tabSetId]->readByte((char*)&_logOffset[tabSetId], sizeof(int));
	_pLog[tabSetId]->seek(_logOffset[tabSetId]);
    }
    _logActive[tabSetId] = true;
}

CegoLogManager::LogResult CegoLogManager::logAction(int tabSetId, CegoLogRecord& logRec)
{

    if ( _logActive[tabSetId] == false )
	return LOG_SUCCESS;

    // cout << "Logging " <<  _lsn[tabSetId] << " " << logRec << endl;

    logRec.setLSN(_lsn[tabSetId]);

    Datetime ts;
    logRec.setTS( ts.asInt());
    
    int len = logRec.getEncodingLength();
	
    char* pBuf = (char*)malloc(len);
        
    logRec.encode(pBuf);
    
    if ( _pLogHandler[tabSetId] )
    {
#ifdef CGDEBUG
	log(_modId, Logger::DEBUG, Chain("Sending log msg ( lsn=") + Chain(_lsn[tabSetId]) + Chain(", size=") + Chain(len) + Chain(")"));    
#endif
	if ( _pLogHandler[tabSetId]->sendLogEntry(pBuf, len) == false )
	{
	    free(pBuf);
	    return LOG_ERROR;
	}
    }
    else
    {

#ifdef CGDEBUG
	log(_modId, Logger::DEBUG, Chain("Logging local ( lsn=") + Chain(_lsn[tabSetId]) + Chain(", size=") + Chain(len) + Chain(")"));    
#endif

	if ( _logOffset[tabSetId] + len > _logSize[tabSetId] )
	{
	    free (pBuf);
	    return LOG_FULL;
	}
        
	_pLog[tabSetId]->writeByte((char*)&len, sizeof(int));       
	_pLog[tabSetId]->writeByte(pBuf, len);		
	_logOffset[tabSetId] += len + sizeof(int);

	_pLog[tabSetId]->seek(0);
	_pLog[tabSetId]->writeByte((char*)&_logOffset[tabSetId], sizeof(int));
	_pLog[tabSetId]->seek(_logOffset[tabSetId]);
    }

    _lsn[tabSetId]++;

    free(pBuf);
    return LOG_SUCCESS;
}

long CegoLogManager::getMinLSN(int tabSetId)
{

    if (_pLog[tabSetId] == 0)
    {
	Chain msg = "No logfile set up for tableset";
	throw Exception(EXLOC, msg);
    }

    _pLog[tabSetId]->seek(0); 
    int c = _pLog[tabSetId]->readByte((char*)&_logOffset[tabSetId], sizeof(int));	
    
    _logPos[tabSetId] = sizeof(int);

    long minlsn=0;

    if ( _logPos[tabSetId] < _logOffset[tabSetId] )
    {

	int len;
	c = _pLog[tabSetId]->readByte((char*)&len, sizeof(int));	

	if ( len > LOGMNG_RECBUFSIZE )
	{
	    Chain msg = "Log read buffer exceeded";
	    throw Exception(EXLOC, msg);		    
	}
	char buf[LOGMNG_RECBUFSIZE];		
	
	_pLog[tabSetId]->readByte(buf, len);
	    
	CegoLogRecord lr;
	lr.decode(buf);
	
	minlsn = lr.getLSN();

	_logPos[tabSetId] += len + sizeof(int);

    }
    return minlsn;
}


long CegoLogManager::getMaxLSN(int tabSetId)
{

    if (_pLog[tabSetId] == 0)
    {
	Chain msg = "No logfile set up for tableset";
	throw Exception(EXLOC, msg);
    }

    _pLog[tabSetId]->seek(0);    
    int c = _pLog[tabSetId]->readByte((char*)&_logOffset[tabSetId], sizeof(int));	

#ifdef CGDEBUG
    log(_modId, Logger::DEBUG, Chain("Read logoffset ") + Chain(_logOffset[tabSetId]));
#endif
        
    _logPos[tabSetId] = sizeof(int);

    long maxlsn=0;

    while (_logPos[tabSetId] < _logOffset[tabSetId])
    {

	int len;
	c = _pLog[tabSetId]->readByte((char*)&len, sizeof(int));	

	if ( len > LOGMNG_RECBUFSIZE )
	{
	    Chain msg = "Log read buffer exceeded";
	    throw Exception(EXLOC, msg);		    
	}
	char buf[LOGMNG_RECBUFSIZE];		
	
	_pLog[tabSetId]->readByte(buf, len);
	    
	CegoLogRecord lr;
	lr.decode(buf);
	
	maxlsn = lr.getLSN();

	_logPos[tabSetId] += len + sizeof(int);

    }
    return maxlsn;
}

void CegoLogManager::seekToStart(int tabSetId)
{
    if (_pLog[tabSetId] == 0)
    {
	Chain msg = "No logfile set up for tableset";
	throw Exception(EXLOC, msg);
    }

    _pLog[tabSetId]->seek(0);    
    int c = _pLog[tabSetId]->readByte((char*)&_logOffset[tabSetId], sizeof(int));    
    _logPos[tabSetId] = sizeof(int);
}


bool CegoLogManager::logSeek(int tabSetId, long lsn)
{
    if (_pLog[tabSetId] == 0)
    {
	Chain msg = "No logfile set up for tableset";
	throw Exception(EXLOC, msg);
    }

    _pLog[tabSetId]->seek(0);    
    int c = _pLog[tabSetId]->readByte((char*)&_logOffset[tabSetId], sizeof(int));	
    
    _logPos[tabSetId] = sizeof(int);

    long loglsn=0;

    while (_logPos[tabSetId] < _logOffset[tabSetId])
    {

	int len;
	c = _pLog[tabSetId]->readByte((char*)&len, sizeof(int));	

	if ( len > LOGMNG_RECBUFSIZE )
	{
	    Chain msg = "Log read buffer exceeded";
	    throw Exception(EXLOC, msg);		    
	}
	char buf[LOGMNG_RECBUFSIZE];		
	
	
	_pLog[tabSetId]->readByte(buf, len);
	    
	CegoLogRecord lr;
	lr.decode(buf);
	
	loglsn = lr.getLSN();

	if (  loglsn == lsn ) 
	{
	    _pLog[tabSetId]->seek(_logPos[tabSetId]);
	    return true;
	}
	else if ( loglsn > lsn ) 
	{

	    Chain msg = "LSN=" + Chain(lsn) + " too small, not found in log";
	    throw Exception(EXLOC, msg);	
	} 

	_logPos[tabSetId] += len + sizeof(int);

    }

    if ( loglsn + 1 == lsn )
    {	
	return false;	
    } 
    else
    {
	Chain msg = "LSN=" + Chain(lsn) + " too high, log delta is missing";
	throw Exception(EXLOC, msg);	
	
    }
}

bool CegoLogManager::logRead(int tabSetId, CegoLogRecord& logRec)
{

    if (_logPos[tabSetId] < _logOffset[tabSetId])
    {
	int len;
	int c = _pLog[tabSetId]->readByte((char*)&len, sizeof(int));	

	if ( len > LOGMNG_RECBUFSIZE )
	{
	    Chain msg = "Log read buffer exceeded";
	    throw Exception(EXLOC, msg);		    
	}
	
	char buf[LOGMNG_RECBUFSIZE];
	
	_pLog[tabSetId]->readByte(buf, len);
	
	logRec.decode(buf);

	_logPos[tabSetId] += len + sizeof(int);
	
	return true;
    }
    return false;

}


bool CegoLogManager::switchLogFile(int tabSetId)
{    

    if ( _pLogHandler[tabSetId] == 0 )
    {

	Chain tableSet = getTabSetName(tabSetId);
	
	ListT<Chain> lfList;
	ListT<int> sizeList;
	ListT<Chain> statusList;
	
	getLogFileInfo(tableSet, lfList, sizeList, statusList);
	
	Chain *pLogFile = lfList.First();
	Chain *pStatus = statusList.First();
	
	bool found = false;
	while ( pLogFile && pStatus && found == false) 
	{
	    
	    if ( *pStatus == Chain(XML_ACTIVE_VALUE))
	    {
	    	
		Chain *pNextLogFile = lfList.Next();
		Chain *pNextLogStatus = statusList.Next();
		if ( pNextLogFile == 0)
		{
		    pNextLogFile = lfList.First();
		    pNextLogStatus = statusList.First();
		}
		
		if ( isArchiveMode(tabSetId) )
		{	    
		    if ( *pNextLogStatus == Chain(XML_OCCUPIED_VALUE))
		    {
			return false;
		    }
		    setLogFileStatus(tableSet, *pLogFile, XML_OCCUPIED_VALUE); 
		}
		else
		    setLogFileStatus(tableSet, *pLogFile, XML_FREE_VALUE); 
		
		
		setLogFileStatus(tableSet, *pNextLogFile, XML_ACTIVE_VALUE);
		
		
		log(_modId, Logger::NOTICE, Chain("Logfile switch to logfile ") + *pNextLogFile + Chain(" for tableSet ") +  tableSet);
		
		setLogFile(tabSetId, *pNextLogFile, false);
		
		found = true;
	    }
	    else
	    {
		pLogFile = lfList.Next();
		pStatus = statusList.Next();
	    }	
	}
	doc2Xml();
	resetLog(tabSetId);
	startLog(tabSetId);
	
    }

    CegoLogRecord lr;
    lr.setAction(CegoLogRecord::LOGREC_SYNC);    
    logAction(tabSetId, lr);
	
    return true;
}

void CegoLogManager::setActiveLogFile(const Chain& tableSet)
{

    int tabSetId = getTabSetId(tableSet);

    ListT<Chain> lfList;
    ListT<int> sizeList;
    ListT<Chain> statusList;

    getLogFileInfo(tableSet, lfList, sizeList, statusList);
    
    Chain *pLogFile = lfList.First();
    Chain *pStatus = statusList.First();
    
    while ( pLogFile && pStatus ) 
    {
	
	if ( *pStatus == Chain(XML_ACTIVE_VALUE))
	{
	    log(_modId, Logger::NOTICE, Chain("Setting active logfile to ") + Chain(*pLogFile) + Chain(" ..."));	
	    setLogFile(tabSetId, *pLogFile, false);	    
	    return;
	}
	else
	{
	    pLogFile = lfList.Next();
	    pStatus = statusList.Next();
	}	
    }   
}
