#include "GetContentListThread.h"

#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <limits.h>
#include <sys/time.h>

using namespace std;



#define DB_INFORMATION_UPDATE_INTERVAL      300 // sec
//#define DB_INFORMATION_UPDATE_INTERVAL      1 // sec
#define DEFAULT_QUERY_BUFFER_SIZE           1024

#define GET_LAST_BOUNDARY_DAY      15 // Day
#define USEC_PER_SEC    (1000000LL)
#define USEC_PER_DAY    (24L*60L*60L*USEC_PER_SEC)
#define QUERY_LIMIT_COUNT    2000

/// @brief  현재 시간 정보를 micro-second 단위로 계산하여 반환
/// @return 1970년 1월 1일 이후부터의 현재시간을 micro-second(1/1,000,000) 단위로 반환처리, 오류발생시에는 -1 값을 반환.
long long longtime_now()
{
	struct timeval tv;
	
	if( gettimeofday(&tv, NULL) == 0 )
	{
		// 성공시 micro-second 단위의 값을 반환
		return tv.tv_sec * USEC_PER_SEC + tv.tv_usec;	
	}
	else
	{
		// 실패시 -1 값을 반환.
		return -1;
	}
}

CGetContentListThread::CGetContentListThread(CQueue<CFileInfo>* pQueueFileInfo
		, CQueue<CFileInfo>* pQueueDeleteJob
		, CQueue<CFileInfo>* pQueueCompleteJob)
	:m_pQueueFileInfo(pQueueFileInfo)
	, m_pQueueDeleteJob(pQueueDeleteJob)
	, m_pQueueCompleteJob(pQueueCompleteJob)
{
	m_pPgSQL = NULL;
	m_bExistCheck = false;
	m_bCurrent = false;
	m_bServiceTableSeparate = false;

	// 멤버 변수 초기화
	m_threadHandle = 0;
}

CGetContentListThread::~CGetContentListThread()
{
	DbClose();
	// Thread 동작 정지 처리
	// - 만약 Thread 가 이미 종료된 경우 m_threadHandle 이 다른 Thread Handle 일 수 있으므로 
	//   업무 Flow 수정시 주의할 것
	if( m_threadHandle != 0 )
		pthread_cancel( m_threadHandle );
}

/// @brief DB 연결 함수.
bool CGetContentListThread::DbConnect()
{
    if( m_pPgSQL != NULL )
    {
        DbClose();
    }
    m_pPgSQL = new DataBase;
	if( m_pPgSQL == NULL )
	{
		LOG(LERR, "Creating new Database has failed.");
		return false;
	}
	if( m_pPgSQL->PgOpenDB(m_szHost, m_nPort, m_szDBName, m_szAcct, m_szPasswd) == NULL )
	{
		LOG(LERR, "Connecting to Database has failed.");
		DbClose();
		return false;
	}
    return true;
}

/// @breif DB close 함수.
void CGetContentListThread::DbClose()
{
	if( m_pPgSQL != NULL )
	{
		delete m_pPgSQL;
        m_pPgSQL = NULL;
	}
}

bool CGetContentListThread::ThreadInit()
{
    FUNC_BEGIN();

	// 조회 정보를 저장할 Queue 에 대한 참조 포인터가 NULL 경우 오류 처리
	if( m_pQueueFileInfo == NULL )
	{
		LOG(LERR, "FileInfo Queue Pointer is NULL");
		return false;
	}
    
	// RCDB 접속 처리를 위한 객체 생성
	DataBase rcdb;

	m_szHost = CProcessStatus::GetInstance()->GetRcdbIp();
	m_nPort = CProcessStatus::GetInstance()->GetRcdbPort(); 
	m_szDBName = CProcessStatus::GetInstance()->GetRcdbName(); 
	m_szAcct = CProcessStatus::GetInstance()->GetRcdbAcct(); 
	m_szPasswd = CProcessStatus::GetInstance()->GetRcdbAcctPw();
	
	// 최초 초기화 시에는 DB연결 정보들이 제대로 되는것인지에 대한 
	// 확인만 처리 한다.
    if (rcdb.PgOpenDB(m_szHost, m_nPort, m_szDBName, m_szAcct, m_szPasswd) == NULL )
	{
		LOG(LERR, "Connecting to Database has failed.");
		return false;
	}
	
	// 접속 성공한 경우...
	LOG( LINF, "RCDB connection test ok..." );

	// 서비스 table 분리된 신규 형상인지.. 구형 형상인지 체크하여
	// m_bServiceTableSeparate 변수 설정 처리.
	// 해당 변수 초기화는 생성자에서 1차로 수행함.
	char szQuery[1024];
	snprintf( szQuery, 1023,"SELECT * FROM pg_tables WHERE schemaname = 'public' AND tablename = 't_dav_resource' " );

	rcdb.PgDoExec( szQuery );
	if( rcdb.PgResult( DataBase::NOT_CLEAR ) < 0 )
	{
		// Query 수행 결과 오류 발생시...
		// - pg_tables 가 존재하지 않을 경우 오류 발생 가능...
		// - 이는 RCDB 가 아직 적절하게 구성되지 않았다는 의미이므로... Init()함수를 오류로 처리한다.
		LOG( LERR, "RCDB t_dav_resource exist check failed.[%s][%s]", rcdb.GetErrorMessage().c_str(), szQuery );

		rcdb.PgClear();
		rcdb.PgCloseDB();

		return false;
	}
	else
	{
		// Query 수행이 정상인 경우.
		int  nResult = rcdb.GetNoTuples();
		if( nResult > 0 )
		{
			// t_dav_resource 테이블이 존재하는 경우...
			m_bServiceTableSeparate = false;
			_LOG( LINF, "CGetContentListThread: t_dav_resource table exist.[Old RCDB Type]" );
		}
		else
		{
			// t_dav_resource 테이블이 존재하지 않는 서비스별 테이블 분리 형상인 경우.
			m_bServiceTableSeparate = true;
			_LOG( LINF, "CGetContentListThread: t_dav_resource table not exist.[New RCDB Type]" );
		}
	}

	// RCDB 조회 결과 set 를 clear 처리 후 RCDB 접속 해제.
	rcdb.PgClear();
	rcdb.PgCloseDB();

	FUNC_END();
	return true;
}

bool CGetContentListThread::GetContentList()
{
	FUNC_BEGIN();
	
	std::list<CPreserve> listPreserve;
	std::list<CPreserve>::iterator it;
	
	// 현재 시간정보를 구한다.( micro-second 단위)
	long long timeNow = longtime_now(); 
	long long timeStart;
	long long timeEnd, timeRealEnd;
	long long timeRealEndMin = LLONG_MAX;
	timeRealEnd = 0;
	
	int nMaxPreserve = 0;
	
	time_t tmStart, tmEnd;
	
	if( m_pPgSQL == NULL )
	{
		LOG(LWAR, "Not initialized.");
		return false;
	}
	
	// Query 문장 저장을 위한 버퍼 생성
	char szQuery[DEFAULT_QUERY_BUFFER_SIZE];
	int nResult = 0;
	
	// 조회 구간을 이동 할 것인지 아닌지 확인하는 flag 이다.
	// 기본은 구간을 이동하지만 특정 조건이 될때는 구간을 옮기지 않는다.
	// limit 수 만큼 쿼리가 된경우 false 로 세팅
	bool bSlide = true;
	
	// 조회 구간의 시작시간이 현재 시간에서 보존기간보다 뺀 기간보다 짧은 경우
	// 다시 동일 시작 시점부터 조회 해야 하는지 확인 기간이다.
	bool bCheckSkip = true;
	
	timeStart = m_timeStartPeriod;
	timeEnd = timeStart + (GET_LAST_BOUNDARY_DAY * USEC_PER_DAY);
	// GET_LAST_BOUNDARY_DAY 보다 작으면 timeEnd 는 현재시간으로...
	// 그렇지 않으면 timeStart에 GET_LAST_BOUNDARY_DAY 만큼 더해서 timeEnd 를 설정한다.
	if( (timeNow - timeStart) >= (GET_LAST_BOUNDARY_DAY * USEC_PER_DAY) )
	{
		LOG(LDBG, "Past......" );
	} else
	{
		LOG(LDBG, "Now...... now[%lld] start[%lld] ",timeNow, timeStart  );
		timeEnd = timeNow;
		m_bCurrent = true;
	}
	
	listPreserve = CProcessStatus::GetInstance()->GetPreserveList();
	for ( it=listPreserve.begin(); it != listPreserve.end(); it++ )
	{
		CPreserve objPreserve = *it;
		int nLimit = QUERY_LIMIT_COUNT;
		char buff1[20];
		char buff2[20];
		
		//LOG( LINF, "start  : %lld, time end : %lld, realend : %lld,  now - preserve : %lld", timeStart, timeEnd, timeRealEnd, ( timeNow - (objPreserve.m_nPreserveDay * USEC_PER_DAY)) );
		
		LOG( LDBG, "svc[%s] start[%lld] end[%lld] real_end[%lld], now - preserve =[%lld]"
			, objPreserve.m_szServiceID.c_str()
			, timeStart, timeEnd, timeRealEnd
			, (timeNow - (objPreserve.m_nPreserveDay * USEC_PER_DAY)) );


		//!! 조건에 따라 EndTime 을 구한다.
		//!! 또는 조건에 따라 쿼리를 하지 않기도 한다.
		
		//1. 시작시간에서 보존 기간을 뺀 값이 마이너스인경우 해당 서비스는 스킵.
		if ( timeStart > (timeNow - (objPreserve.m_nPreserveDay * USEC_PER_DAY))  )
		{
			// skip 된 서비스가 하나라도 있으면 Start Period 를 옮기지 않는다.
			bCheckSkip = false;
			continue;
		}
		//2. 현재시간에서 보존기간을 뺀 값이 시작과 끝 시간 사이라면 timeEnd 를 조정해 준다.
		else if ( timeEnd > ( timeNow - (objPreserve.m_nPreserveDay * USEC_PER_DAY))  )
		{
			timeRealEnd = timeNow - (objPreserve.m_nPreserveDay * USEC_PER_DAY);
			// 테스트를 위함 -- 3분
			//timeRealEnd = timeNow - (180 * USEC_PER_SEC);
		}
		else
		{
			timeRealEnd = timeEnd;
		}
		//3. 그렇지 않으면 그냥 기존 시간대로 진행... ( 현재시간과 가깝지 않다는 뜻..)
		
		// 가장 작은 실제 엔드 값을 구한다.
		// 이는 조회 구간 조정시 사용 된다.
		if ( timeRealEnd < timeRealEndMin )
		{
			timeRealEndMin = timeRealEnd;
		}
		//4. 시간값 로깅 처리
		tmStart = timeStart/USEC_PER_SEC;
		tmEnd = timeRealEnd/USEC_PER_SEC;
        strftime(buff1, 20, "%Y-%m-%d %H:%M:%S", localtime(&tmStart));
        strftime(buff2, 20, "%Y-%m-%d %H:%M:%S", localtime(&tmEnd));
		
		_LOG( LINF, "content check: svc[%s] start[%s] end[%s]", objPreserve.m_szServiceID.c_str(), buff1, buff2);
	    LOG( LDEV, "GetList Query : [%s]", szQuery );
	    
		//!! webting 수정 필요
		if(m_bServiceTableSeparate == false)
		{
			// 구형 DB 스키마 용
			// 실제 쿼리
			snprintf(szQuery, DEFAULT_QUERY_BUFFER_SIZE - 1,
					"SELECT sp_svc_tran_id, host_name, resource_type, uri, filename_hash, get_content_length, get_lastmodified "
					"FROM t_dav_resource "
					"WHERE sp_svc_tran_id = %s AND get_lastmodified >= %lld AND get_lastmodified < %lld AND deleted_yn = 'Y'  LIMIT %d; "
					, objPreserve.m_szServiceID.c_str()
					, timeStart
					, timeRealEnd
					, nLimit );
		}
		else
		{
			// 실제 쿼리
			snprintf(szQuery, DEFAULT_QUERY_BUFFER_SIZE - 1,
					"SELECT sp_svc_tran_id, host_name, resource_type, uri, filename_hash, get_content_length, get_lastmodified "
					"FROM t_meta_%s "
					"WHERE sp_svc_tran_id = %s AND get_lastmodified >= %lld AND get_lastmodified < %lld AND deleted_yn = 'Y'  LIMIT %d; "
					, objPreserve.m_szServiceID.c_str()
					, objPreserve.m_szServiceID.c_str()
					, timeStart
					, timeRealEnd
					, nLimit );
		}
			    
		// Query 실행 
		m_pPgSQL->PgDoExec(szQuery);
		nResult = m_pPgSQL->PgResult( DataBase::NOT_CLEAR );
		if( nResult < 0 )
		{
			if( m_pPgSQL != NULL )
			{
				LOG(LWAR, "RCDB Query failed.[%d][%s]", nResult, m_pPgSQL->GetErrorMessage().c_str() );
	            
				m_pPgSQL->PgClear();
	            DbClose();
			}
			
			// 실패 해도 변경 주기 만큼 슬립한다.
			return false;
		}
	    
		nResult = m_pPgSQL->GetNoTuples();
		
		if( nResult >= 1 )
		{
			m_bExistCheck = true;
		}
		
		if( nResult >= QUERY_LIMIT_COUNT )
		{
			bSlide = false;
		}
		
	    _LOG( LINF, "delete target: svc[%s] count[%d]", objPreserve.m_szServiceID.c_str(), nResult);
		// Replication 을 수행할 고객사가 존재하는 경우.
		// 조회된 정보를 내부 멤버변수에 저장처리.
		for( int i =0 ; i< nResult; i++ )
		{
			CFileInfo objFileInfo;
			objFileInfo.m_szSvcTranid = m_pPgSQL->GetValue(i, 0);
			objFileInfo.m_szHostName = m_pPgSQL->GetValue(i, 1);
			objFileInfo.m_nType =atoi(m_pPgSQL->GetValue(i, 2));
			objFileInfo.m_szUri = m_pPgSQL->GetValue(i, 3);
			objFileInfo.m_szFileNameHash = m_pPgSQL->GetValue(i, 4);
			objFileInfo.m_nContentLength = atoll(m_pPgSQL->GetValue(i, 5));
			objFileInfo.m_nLastModified = atoll(m_pPgSQL->GetValue(i, 6));
			
			m_pQueueFileInfo->Push(objFileInfo);
		}
		
		// Query Result set Clear.
		m_pPgSQL->PgClear();
	}
	
	//!! 쿼리 결과가 limit 수 보다 모두 적은 경우는 startPeriod 를 옮겨 준다.
	if( bSlide == true && bCheckSkip == true )
	{
		m_timeStartPeriod = timeRealEndMin;
	}
	
	FUNC_END();
	return true;
}


///@brief 이 함수에서 컨트롤 로직이 표현된다.
void CGetContentListThread::Execute()
{
	FUNC_BEGIN();
	
	// 시작 모드를 결정한다.

	std::string szStartPeriod = CProcessStatus::GetInstance()->GetStartPeriod();
	
	// 설정 값이 있다면 그 시간을 시작으로 잡고...
	if ( szStartPeriod.size() == 8 )
	{ 
		struct tm sTime= {0,};
		sTime.tm_year = atoi(szStartPeriod.substr(0,4).c_str()) - 1900;
		sTime.tm_mon = atoi(szStartPeriod.substr(4,2).c_str()) - 1;
		sTime.tm_mday = atoi(szStartPeriod.substr(6,2).c_str());
		
		time_t time_tee = mktime(&sTime);
		m_timeStartPeriod = time_tee * USEC_PER_SEC;
	}
	else
	{
		// 설정값이 없다면 기본 boundary 의 두배 만큼 현재 시간에서 과거로 돌려서.. 시작.
		m_timeStartPeriod = longtime_now() - (GET_LAST_BOUNDARY_DAY * USEC_PER_DAY *2);
	}
	
	while(1)
	{
		std::list<CPreserve> listPreserve;
		listPreserve = CProcessStatus::GetInstance()->GetPreserveList();
		m_bExistCheck = false;
		m_bCurrent = false;
		if ( listPreserve.size() <= 0 )
		{
			LOG(LWAR, "listPreserve is Not initialized.");
			sleep(1);
			continue;
		}
		if ( DbConnect() == false)
		{
			// 한 번더 정리한 후 
			DbClose();
			
			sleep( DB_INFORMATION_UPDATE_INTERVAL );
			// 연결 재시도
			continue;
		}
		// 아래 함수에서 return false 가 되면 기존 세션에대해 정리한 후
		// DB Connection을 다시 요청한다.
		if( GetContentList() == false )
		{
			// 한 번더 정리한 후 
			DbClose();
			sleep( DB_INFORMATION_UPDATE_INTERVAL );
			// 연결 재시도
			continue;
		}
		
		// 일단 DB 연결을 해제한다.
		DbClose();
		// 끝날때까지 대기 한다.
		while(1)
		{
			_LOG(LINF, "Job cheking ...");
			_LOG(LINF, "    List Queue[%d], Delete Queue[%d], Delete Thread[%d], Complete Queue[%d]"
				, m_pQueueFileInfo->GetSize(), m_pQueueDeleteJob->GetSize()
			    , CProcessStatus::GetInstance()->GetDeleteThreadCount()
			    , m_pQueueCompleteJob->GetSize() );


			if( m_pQueueFileInfo->GetSize() <= 0 &&
				m_pQueueDeleteJob->GetSize() <= 0 &&
			  	m_pQueueCompleteJob->GetSize() <= 0 &&
			  	CProcessStatus::GetInstance()->GetDeleteThreadCount() <= 0  )
			{
				break;
			}
			sleep(1);
		}
		//sleep(DB_INFORMATION_UPDATE_INTERVAL);
		
		// 
		if( m_bExistCheck == true )
		{
			sleep(1);
		} else {
			if( m_bCurrent == false )
			{
				sleep(1);
			} else
			{
				sleep(DB_INFORMATION_UPDATE_INTERVAL);
			}
		}
	}
	
	// 쓰레드 종료 될 때 DB 세션 정리 한다.
	DbClose();
	
	LOG(LERR, "Thread create failed.: errno:%d errmsg:%s", errno, strerror(errno));
	FUNC_END();
}

void* CGetContentListThread::EntryPoint(void* arg)
{
    CGetContentListThread* pObject = reinterpret_cast<CGetContentListThread *>(arg);
	pthread_detach( pthread_self() );
    
	pthread_testcancel();
	pObject->Execute();
	pthread_testcancel();
	
	// Thread 종료시 m_threadHandle 값을 초기화 처리.
	// - 소멸자에서 잘못된 Thread 를 종료할 수 있기 때문.
	pObject->m_threadHandle = 0;

	delete pObject;
	return 0;
}

bool CGetContentListThread::Start()
{
	int nRet = ::pthread_create(&m_threadHandle, 0, CGetContentListThread::EntryPoint, this);

	if( nRet )
	{
		LOG(LERR, "Thread create failed.: errno:%d errmsg:%s", errno, strerror(errno));
		return false;
	}
    LOG(LDBG, "Thread create succeed");
	sleep(0);
	return true;
}
