#include "CacheInsert.h"

#include <errno.h>
#include <string.h>

#include "Database.h"
#include "Logger.h"
#include "Util.h"


// 처리 loop 간의 sleep time 정의
#define DEFAULT_SLEEP_TIME  3
// RCDB 접속 실패 또는 Query 실패 등의 상황 발생시 
#define MAX_SLEEP_TIME     30

// Query 문장 저장을 위한 Buffer 크기
#define DEFAULT_QUERY_BUFFER_SIZE 2048


// 생성자
// param [IN] pQueueCacheComplete Cache 생성 완료 Queue 에 대한 포인터
// param [IN] pListCacheJob Cache 생성 관련 Job List 객체에 대한 포인터
CCacheInsert::CCacheInsert( CDataQueue<CCacheData>* pQueueCacheComplete, CCacheJobList* pCacheJob )
	: m_pQueueCacheComplete( pQueueCacheComplete )
	, m_pCacheJob( pCacheJob )
	
{
	m_threadHandle = 0;

	m_uSleep = DEFAULT_SLEEP_TIME;

	// NEW 2014-07-07 신규 멤버 변수 초기화 코드 추가.
	// Init() 에서 RCDB 접속 실패가 발생 가능하므로.. 구형 RCDB 형상을 사용토록 설정.
	m_bServiceTableSeparate = false;
	}

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

// 객체 초기화 수행
bool CCacheInsert::Init()
{
	// 생성자로 전달받은 각 객체에 대한 포인터는 유효하다가 판단하고....
	// 각 멤버변수에 대한 검사는 수행하지 않는다.

	// RCDB 접속 정보 Load.
	m_rcdbInfo.Load();

	// RCDB 접속 처리를 위한 객체 생성
	DataBase rcdb;

	// RCDB 접속 시도 및 접속 실패시
	if( rcdb.PgOpenDB( m_rcdbInfo.m_strRcdbIp, m_rcdbInfo.m_nRcdbPort
		, m_rcdbInfo.m_strRcdbName, m_rcdbInfo.m_strRcdbAcct
		, m_rcdbInfo.m_strRcdbAcctPw ) == NULL )
	{
		LOG( LERR, "RCDB connection failed." );
		return false;
	}
	else
	{
		// 접속 성공한 경우...
		LOG( LINF, "RCDB connection test ok..." );

		// NEW 2014-06-23 huibong
		// 서비스 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, "CCacheInsert: t_dav_resource table exist.[Old RCDB Type]" );
			}
			else
			{
				// t_dav_resource 테이블이 존재하지 않는 서비스별 테이블 분리 형상인 경우.
				m_bServiceTableSeparate = true;
				_LOG( LINF, "CCacheInsert: t_dav_resource table not exist.[New RCDB Type]" );
			}
		}

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

		return true;
	}	
}

// thread를 생성하여 업무 flow 를 시작.
bool CCacheInsert::Start()
{
	int nRet = ::pthread_create( &m_threadHandle, 0, CCacheInsert::threadFunc, this );
	if( nRet != 0 )
	{
		// Thread 생성 실패시
		int errorNum = errno;
		LOG( LERR, "Thread create failed.[%d][%s]", errorNum, strerror( errorNum ) );

		return false;
	}
	
	return true;
}

// 스레드 함수
void* CCacheInsert::threadFunc( void* arg )
{
	CCacheInsert* pObject = reinterpret_cast<CCacheInsert *>( arg );
	pthread_detach( pthread_self() );

	while( 1 )
	{
		pthread_testcancel();
		pObject->Execute();
		pthread_testcancel();

		// 1회 수행 후 지정된 시간 만큼 대기
		sleep( pObject->m_uSleep );
	}

	// NEW 2014-07-07 huibong
	// Thread 종료시 m_threadHandle 값을 초기화 처리.
	// - 소멸자에서 잘못된 Thread 를 종료할 수 있기 때문.
	pObject->m_threadHandle = 0;
		
	return NULL;
}

// 스레드 내에서 각 loop 단위 처리 호출 함수.
void CCacheInsert::Execute()
{
	// Cache complete queue 상에 처리할 Data 가 존재하는 경우.
	if( m_pQueueCacheComplete->GetSize() > 0 )
	{
		bool bResult;

		// t_dav_resource 테이블 사용 형상인 경우.
		if( m_bServiceTableSeparate == false )
		{
			bResult = InsertToOldDB();
		}
		else
		{
			// 서비스별 Table 이 분린된 신규 형상인 경우.
			bResult = InsertToDB();
		}
				
		// 처리 결과에 따라 loop sleep time 조정
		if( bResult == true )
			m_uSleep = DEFAULT_SLEEP_TIME;
		else
		{
			// RCDB 연결 실패, Queury 수행 실패 등의 오류 발생시
			m_uSleep = MAX_SLEEP_TIME;
		}
	}
	else
	{
		// 처리할 Data 가 존재하지 않는 경우.
		m_uSleep = DEFAULT_SLEEP_TIME;
	}
}

// Cache 생성이 완료된 Job 내역을 구형 RCDB 형상에 INSERT 처리
// - RCDB 에 저장이 완료된 후
// - 해당 내역을 Job 관리 항목에서 삭제 처리해야 추가 요청을 수신할 수 있다.
bool CCacheInsert::InsertToOldDB()
{
	// 1. RCDB 에 접속	
	DataBase rcdb;
	if( rcdb.PgOpenDB( m_rcdbInfo.m_strRcdbIp, m_rcdbInfo.m_nRcdbPort
		, m_rcdbInfo.m_strRcdbName, m_rcdbInfo.m_strRcdbAcct
		, m_rcdbInfo.m_strRcdbAcctPw ) == NULL )
	{
		// RCDB 접속이 실패
		LOG( LERR, "RCDB connection failed." );
		return false;
	}
	
	// 2. Queue Pop 처리	
	
	// queue 에서 꺼낸 정보를 저장하기 위한 변수
	CCacheData resultData;

	// Query 문장 저장을 위한 버퍼 생성
	char szQuery[DEFAULT_QUERY_BUFFER_SIZE];
	
	// Query 수행 결과값 저장을 위한 변수..	
	int nResult = 0;

	while( m_pQueueCacheComplete->GetSize() > 0 )	
	{
		// queue 에 저장된 첫번째 data 복사 (queue 에서 삭제된 상태 아님)		
		resultData = m_pQueueCacheComplete->Front();

		// 3. RCDB 에 저장된 해당 Content 의 원본 정보를 조회
		//    - 이미 삭제가 되었을 수도 있지만....
		//      RCDB 에 등록하지 않을 경우... 생성된 content 의 자동 삭제가 불가능하므로...
		//      조회시 삭제 여부는 조건에 포함시키지 않는다.
		//    - uri 정보를 이용하여 조회시 특수문자 포함으로 문제가 발생 가능하고..
		//      실제 작업을 수행한 원본이 아닐 수 있으므로... (RCDB key 값 문제)
		//      Source FHS 와 filname_hash 정보를 이용하여 조회 수행
		sprintf( szQuery,
			"SELECT "
			" sp_user_seq, sp_svc_tran_id, '%s', "        // 생성된 FHS host_name
			" resource_type, uri, filename_org, '%s', "   // 생성된 filename_hash
			" depth, display_name, get_content_language,  "
			" get_content_length, get_content_type, get_etag, "
			" %lld, creation_date, "                      // get_lastmodified
			" lock_discovery, source, supported_lock, deleted_yn, 'Y', file_lastmodified "
			"FROM t_dav_resource "
			"WHERE filename_hash = '%s' "                 // 생성시 사용한 filename_hash
			"AND get_content_length = %lld "              // 생성시 사용한 get_content_length
			"AND host_name = '%s' ;"                      // 생성시 사용한 FHS host_name
			, resultData.m_strTargetHost.c_str()             // 생성된 FHS host_name
			, resultData.m_strTargetFilenameHash.c_str()     // 생성된 filename_hash
			, longtime_now()                                 // get_lastmodified
			, resultData.m_strSourceFilenameHash.c_str()     // 생성시 사용한 filename_hash
			, resultData.m_requestData.m_uContentLength      // 생성시 사용한 get_content_length
			, resultData.m_strSourceHost.c_str()             // 생성시 사용한 FHS host_name
			);

		// Query 수행 및 결과 확인
		rcdb.PgDoExec( szQuery );
		nResult = rcdb.PgResult( DataBase::NOT_CLEAR );
		if( nResult < 0 )
		{
			// RCDB t_dav_resource 테이블 조회 실패시
			// 우선 DB 연결 문제일 수 있으므로... 
			// RCDB 재접속 시도.
			rcdb.PgClear();
			rcdb.PgCloseDB();
			sleep( 1 );

			if( rcdb.PgOpenDB( m_rcdbInfo.m_strRcdbIp, m_rcdbInfo.m_nRcdbPort
				, m_rcdbInfo.m_strRcdbName, m_rcdbInfo.m_strRcdbAcct
				, m_rcdbInfo.m_strRcdbAcctPw ) == NULL )
			{
				// RCDB 접속 실패 문제인 경우..
				// - 본 처리 함수 종료 처리하여 다음 loop 에서 재시도토록 처리.
				LOG( LERR, "select query fail because RCDB connection error. [%d][%s]"
					, nResult, rcdb.GetErrorMessage().c_str() );

				rcdb.PgCloseDB();
				return false;
			}
			else
			{
				// query 실패시				
				// - 관련 내역 로깅
				// - queue 에서 해당 정보를 삭제, Job 에서 해당 정보 삭제 후
				// - 루프 계속 수행한다.
				LOG( LERR, "select query fail. queue forced pop.[%d][%s][%s]"
					, nResult, rcdb.GetErrorMessage().c_str(), szQuery );
				
				rcdb.PgClear();

				m_pQueueCacheComplete->Pop();
				m_pCacheJob->UnRegister( resultData );								
				continue;
			}
		}

		// 결과 Row 수 확인.
		nResult = rcdb.GetNoTuples();
		if( nResult <= 0 )
		{
			// Cache 생성된 content 가 RCDB 에 존재하지 않는 경우...
			// - RCDB 에 저장하지 않고 그냥 버린다.
			// - 이로 인해 FHS 에 meta 기준으로 삭제 불가능한 파일이 남게 된다.			
			_LOG( LINF, "Cache source content[%s] not found in RCDB."
				, resultData.m_strSourceFilenameHash.c_str() );

			rcdb.PgClear();

			m_pQueueCacheComplete->Pop();
			m_pCacheJob->UnRegister( resultData );
			continue;			
		}

		// 조회 결과는 
		// COPY 처리로 인해 N 개의 URI 가 작업에 사용된 content 를 가르킬 수 있으므로...
		// N 개가 조회 될 수 있다.
		// 따라서 COPY 된 URI 각각에 대해 Cache 정보를 등록해야 한다.
						
		// 조회된 N개 row 정보를 저장하기 위한 vector
		std::vector< std::vector<string> > srcarry;
		std::string strResult;
		for( int j = 0; j < nResult; j++ )
		{
			// 조회된 1개의 Row 정보를 저장하기 위한 vector
			std::vector<string> srcinfo;
			for( int k = 0; k < 21; k++ )
			{
				strResult = rcdb.GetValue( j, k );
				strResult = stringreplace( strResult, "'", "''" );
				srcinfo.push_back( strResult );
			}

			srcarry.push_back( srcinfo );
		}
		
		// Select Query 의 결과를 Clear 
		rcdb.PgClear();
		
		// 4. 조회된 N 개의 정보를 바탕으로 RCDB 에 INSERT 처리 수행.
		while( srcarry.empty() == false )
		{
			std::vector<string> srcinfo = srcarry.back();
			srcarry.pop_back();

			sprintf( szQuery,
				"INSERT INTO t_dav_resource "
				"( "
				"  sp_user_seq, sp_svc_tran_id, "
				"  host_name, "
				"  resource_type, uri, filename_org, filename_hash, "
				"  depth, display_name, get_content_language, "
				"  get_content_length, get_content_type, "
				"  get_etag, get_lastmodified, creation_date, "
				"  lock_discovery, source, supported_lock, deleted_yn, is_cache ,file_lastmodified) "
				"VALUES "
				" ( "
				"  %s, %s, "
				"  '%s', "
				"  %s, '%s', '%s', '%s', "
				"  %s, '%s', '%s', "
				"  %s, '%s', "
				"  '%s', %s, %s, "
				"  '%s', %s, '%s', '%s', '%s', %s );", // is_cache = Y
				srcinfo[0].c_str(), srcinfo[1].c_str(), srcinfo[2].c_str(), srcinfo[3].c_str(),
				srcinfo[4].c_str(), srcinfo[5].c_str(), srcinfo[6].c_str(), srcinfo[7].c_str(),
				srcinfo[8].c_str(), srcinfo[9].c_str(), srcinfo[10].c_str(), srcinfo[11].c_str(),
				srcinfo[12].c_str(), srcinfo[13].c_str(), srcinfo[14].c_str(), srcinfo[15].c_str(),
				srcinfo[16].c_str(), srcinfo[17].c_str(), srcinfo[18].c_str(), srcinfo[19].c_str(), srcinfo[20].c_str()
				);

			// INSERT query 수행
			rcdb.PgDoExec( szQuery );
			nResult = rcdb.PgResult( DataBase::CLEAR );
			if( nResult < 0 )
			{
				// RCDB INSERT 작업시 오류가 발생한 경우...
				// - DB 연결 문제 역시 바로 앞에서 체크 했으므로 아닐 가능성이 큼.
				// - 가장 큰 오류는 Query 오류일 가능성이 큼.
				// - 루프 계속 수행
				
				LOG( LERR, "insert query fail. queue forced pop.[%d][%s][%s]"
					, nResult, rcdb.GetErrorMessage().c_str(), szQuery );				
			}
			else
			{
				// 성공시
				// PgResult 호출시 CLEAR 옵션을 사용했으므로....내부 결과 set 포인터가 NULL 처리됨. 
				// 따라서 수행 row 수 확인을 위해 GetCmdTuples() 함수를 호출해서는 안됨.

				_LOG( LINF, "Cache insert to RCDB. query result[%d] [%s][%llu]"
					, nResult, resultData.m_strTargetFilenameHash.c_str()
					, resultData.m_requestData.m_uContentLength );
			}
			
			// Query 재사용을 위한 clear 처리.
			rcdb.PgClear();
		}

		// 5. 해당 Content 에 대한 작업이 완료된 경우..
		// - Queue , Job 항목에서 삭제 처리.
		m_pQueueCacheComplete->Pop();
		m_pCacheJob->UnRegister( resultData );

	}

	// queue 에 저장된 정보 처리 작업이 완료된 경우...
	// - RCDB 연결 해제 처리.
	rcdb.PgClear();
	rcdb.PgCloseDB();
	
	return true;
}


// Cache 생성이 완료된 Job 내역을 신규 RCDB 형상에 INSERT 처리
// - RCDB 에 저장이 완료된 후
// - 해당 내역을 Job 관리 항목에서 삭제 처리해야 추가 요청을 수신할 수 있다.
bool CCacheInsert::InsertToDB()
{
	// 1. RCDB 에 접속	
	DataBase rcdb;
	if( rcdb.PgOpenDB( m_rcdbInfo.m_strRcdbIp, m_rcdbInfo.m_nRcdbPort
		, m_rcdbInfo.m_strRcdbName, m_rcdbInfo.m_strRcdbAcct
		, m_rcdbInfo.m_strRcdbAcctPw ) == NULL )
	{
		// RCDB 접속이 실패
		LOG( LERR, "RCDB connection failed." );
		return false;
	}

	// 2. Queue Pop 처리
	

	// queue 에서 꺼낸 정보를 저장하기 위한 변수
	CCacheData resultData;

	// Query 문장 저장을 위한 버퍼 생성
	char szQuery[DEFAULT_QUERY_BUFFER_SIZE];

	// Query 수행 결과값 저장을 위한 변수..	
	int nResult = 0;

	while( m_pQueueCacheComplete->GetSize() > 0 )	
	{
		// queue 에 저장된 첫번째 data 복사 (queue 에서 삭제된 상태 아님)		
		resultData = m_pQueueCacheComplete->Front();

		// 3. RCDB 에 저장된 해당 Content 의 원본 정보를 조회
		//    - 이미 삭제가 되었을 수도 있지만....
		//      RCDB 에 등록하지 않을 경우... 생성된 content 의 자동 삭제가 불가능하므로...
		//      조회시 삭제 여부는 조건에 포함시키지 않는다.
		//    - uri 정보를 이용하여 조회시 특수문자 포함으로 문제가 발생 가능하고..
		//      실제 작업을 수행한 원본이 아닐 수 있으므로... (RCDB key 값 문제)
		//      Source FHS 와 filname_hash 정보를 이용하여 조회 수행
		//    - 신규 RCDB 형상 지원 및 미사용 컬럼 제거 처리

		// CHG 2015-02-12 huibong 
		// 신규 RCDB 형상에 맞도록 처리 Flow 단순화 (#21354)
		// - RCDB 구조가 pgpool -> streaming replicaton 구조로 변경됨
		// - 이에 따라 기존 SELECT 후 INSERT, 검수, DELETE flow 에 대한 단순화 가능함.
		
		sprintf( szQuery,
				 "INSERT INTO t_meta_%s "
				 "( "
				 "  sp_user_seq, sp_svc_tran_id, host_name, "
				 "  resource_type, uri, filename_org, filename_hash, "
				 "  depth, display_name, "
				 "  get_content_length, get_content_type, "
				 "  get_lastmodified, creation_date, "
				 "  source, deleted_yn, is_cache, file_lastmodified "
				 ") "
				 "SELECT "
				 " sp_user_seq, sp_svc_tran_id, '%s', "               // 생성된 FHS host_name
				 " resource_type, uri, filename_org, '%s', "          // 생성된 filename_hash
				 " depth, display_name, "
				 " get_content_length, get_content_type, "
				 " %lld, creation_date, "                             // get_lastmodified
				 " source, deleted_yn, 'Y', file_lastmodified "
				 "FROM t_meta_%s "                                    // 신규 RCDB 형상
				 "WHERE filename_hash = '%s' "                        // 생성시 사용한 filename_hash
				 "AND get_content_length = %lld "                     // 생성시 사용한 get_content_length
				 "AND host_name = '%s' ;"                             // 생성시 사용한 FHS host_name
				 , resultData.m_requestData.m_strServiceSeq.c_str()   // 신규 RCDB 형상 지원
				 , resultData.m_strTargetHost.c_str()                 // 생성된 FHS host_name
				 , resultData.m_strTargetFilenameHash.c_str()         // 생성된 filename_hash
				 , longtime_now()                                     // get_lastmodified
				 , resultData.m_requestData.m_strServiceSeq.c_str()   // 신규 RCDB 형상 지원
				 , resultData.m_strSourceFilenameHash.c_str()         // 생성시 사용한 filename_hash
				 , resultData.m_requestData.m_uContentLength          // 생성시 사용한 get_content_length
				 , resultData.m_strSourceHost.c_str()                 // 생성시 사용한 FHS host_name
		);
		
		// Query 수행 및 결과 확인
		rcdb.PgDoExec( szQuery );
		nResult = rcdb.PgResult( DataBase::NOT_CLEAR );
		if( nResult < 0 )
		{
			// Query 수행 실패시
			// 우선 DB 연결 문제일 수 있으므로... 
			// RCDB 재접속 시도.

			std::string strPreviousErrorMessage = rcdb.GetErrorMessage();

			rcdb.PgClear();
			rcdb.PgCloseDB();
			sleep( 1 );

			if( rcdb.PgOpenDB( m_rcdbInfo.m_strRcdbIp, m_rcdbInfo.m_nRcdbPort
				, m_rcdbInfo.m_strRcdbName, m_rcdbInfo.m_strRcdbAcct
				, m_rcdbInfo.m_strRcdbAcctPw ) == NULL )
			{
				// RCDB 접속 실패 문제인 경우..
				// - 본 처리 함수 종료 처리하여 다음 loop 에서 재시도토록 처리.
				LOG( LERR, "Query fail because RCDB connection error. [%d][%s]"
					, nResult, rcdb.GetErrorMessage().c_str() );

				rcdb.PgCloseDB();
				return false;
			}
			else
			{
				// query 실패시				
				// - 관련 내역 로깅
				// - queue 에서 해당 정보를 삭제, Job 에서 해당 정보 삭제 후
				// - 루프 계속 수행한다.
				LOG( LERR, "insert-select query fail. queue forced pop.[%d][%s][%s]"
					 , nResult, strPreviousErrorMessage.c_str(), szQuery );

				rcdb.PgClear();

				m_pQueueCacheComplete->Pop();
				m_pCacheJob->UnRegister( resultData );
				continue;
			}
		}

		// PgResult 호출시 NOT_CLEAR 옵션을 사용했으므로....INSERT 수행 row 수 확인을 위해 GetCmdTuples() 함수를 호출가능.
		nResult = rcdb.GetCmdTuples();
		if( nResult > 0 )
		{
			// COPY 처리로 인해 N 개의 URI 가 작업에 사용된 content 를 가르킬 수 있으므로...
			// N 개가 나타날 수 있다.
			
			_LOG( LINF, "Cache insert to RCDB. [%d] row. [%s][%s][%llu]"
				  , nResult
				  , resultData.m_strTargetHost.c_str()
				  , resultData.m_strTargetFilenameHash.c_str()
				  , resultData.m_requestData.m_uContentLength 
			);
		}
		else
		{
			// INSERT 처리된 row 가 없는 경우....
			// - 생성된 cache 파일은 고아 파일이 됨.
			
			_LOG( LWAR, "cache rcdb Insert origin not found. orgin[%s|%s|%llu] orphan[%s|%s|%llu]"
				  , resultData.m_strSourceHost.c_str()
				  , resultData.m_strSourceFilenameHash.c_str()
				  , resultData.m_requestData.m_uContentLength
				  , resultData.m_strTargetHost.c_str()
				  , resultData.m_strTargetFilenameHash.c_str()
				  , resultData.m_requestData.m_uContentLength
			);
		}


		// Query 의 결과 Clear 
		rcdb.PgClear();


		// 해당 Content 에 대한 작업이 완료된 경우..
		// - Queue , Job 항목에서 삭제 처리.		
		m_pQueueCacheComplete->Pop();
		m_pCacheJob->UnRegister( resultData );
				
	}

	// queue 에 저장된 정보 처리 작업이 완료된 경우...
	// - RCDB 연결 해제 처리.
	rcdb.PgClear();
	rcdb.PgCloseDB();

	return true;
}
