298 lines
10 KiB
PHP
298 lines
10 KiB
PHP
<?php
|
|
|
|
namespace App\Commands;
|
|
|
|
use CodeIgniter\CLI\BaseCommand;
|
|
use CodeIgniter\CLI\CLI;
|
|
|
|
use App\Models\Entities\NaverWorkerLogModel; // 새로 만든 테이블용 모델
|
|
|
|
// 헬퍼 로드 (app/Helpers/log_helper.php 가 있어야 함 autoload 설정 넣어놓았음)
|
|
|
|
class NaverWorker extends BaseCommand
|
|
{
|
|
protected $group = 'Workers';
|
|
protected $name = 'naver:worker';
|
|
protected $description = 'Redis에서 데이터를 꺼내 DB에 저장하고 네이버 API를 호출합니다.';
|
|
|
|
// DB 객체를 담을 변수 선언
|
|
protected $db;
|
|
protected $redisHost;
|
|
protected $redisPort;
|
|
protected $redisDatabase;
|
|
|
|
public function run(array $params)
|
|
{
|
|
helper(['log', 'redis']); // redis helper 추가
|
|
|
|
$this->db = \Config\Database::connect();
|
|
|
|
// 워커 시작 시점에 선제적으로 연결 상태를 보정
|
|
try {
|
|
$this->db->initialize();
|
|
} catch (\Throwable $e) {
|
|
CLI::error('Database connection init failed: ' . $e->getMessage());
|
|
}
|
|
|
|
$logModel = model(NaverWorkerLogModel::class);
|
|
$naverService = new \App\Services\NaverService(); // 서비스 생성
|
|
|
|
// Redis 연결 (실패해도 계속 진행 - 파일 모드로 동작 가능)
|
|
$redis = get_redis_connection('worker');
|
|
$config = get_redis_config('worker');
|
|
|
|
if ($redis) {
|
|
CLI::write(CLI::color('🟢 Naver Worker running... (Redis: ' . $config['host'] . ':' . $config['port'] . ' DB:' . $config['database'] . ')', 'green'));
|
|
} else {
|
|
CLI::write(CLI::color('⚠️ Naver Worker running in FILE-ONLY mode (Redis unavailable)', 'yellow'));
|
|
}
|
|
|
|
|
|
|
|
while (true) {
|
|
|
|
// Redis 또는 폴백 파일에서 데이터 읽기
|
|
$rawData = null;
|
|
$source = 'redis'; // 데이터 소스 추적
|
|
|
|
// 1. Redis에서 데이터 읽기 시도 (Redis가 있을 경우만)
|
|
if ($redis) {
|
|
$maxRetries = 2;
|
|
$retryCount = 0;
|
|
|
|
while ($retryCount < $maxRetries) {
|
|
try {
|
|
$result = $redis->brPop(['naver:raw_queue'], 5); // 5초 타임아웃
|
|
if ($result) {
|
|
$rawData = $result[1];
|
|
$source = 'redis';
|
|
}
|
|
break; // 성공하면 루프 탈출
|
|
} catch (\Exception $e) {
|
|
$retryCount++;
|
|
CLI::write(CLI::color("⚠️ Redis error (attempt {$retryCount}/{$maxRetries}): " . $e->getMessage(), 'yellow'));
|
|
|
|
if ($retryCount >= $maxRetries) {
|
|
CLI::write(CLI::color("⚠️ Redis unavailable, switching to file mode", 'yellow'));
|
|
$redis = null; // Redis를 비활성화
|
|
break;
|
|
}
|
|
|
|
// Redis 재연결 시도
|
|
try {
|
|
CLI::write(CLI::color('🔄 Reconnecting to Redis...', 'yellow'));
|
|
$redis->close();
|
|
$redis = get_redis_connection('worker');
|
|
if ($redis) {
|
|
CLI::write(CLI::color('✅ Redis reconnected', 'green'));
|
|
}
|
|
} catch (\Exception $reconnectError) {
|
|
CLI::error("Redis reconnection error: " . $reconnectError->getMessage());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// 2. Redis에서 데이터 없으면 폴백 파일 확인
|
|
if (!$rawData) {
|
|
$rawData = $this->readFromFallbackFile();
|
|
if ($rawData) {
|
|
$source = 'file';
|
|
}
|
|
}
|
|
|
|
// 3. 데이터 없으면 다음 루프
|
|
if (!$rawData) {
|
|
continue;
|
|
}
|
|
|
|
// 4. 데이터 소스 로깅
|
|
CLI::write(CLI::color("📥 Data received from: " . strtoupper($source), 'cyan'));
|
|
|
|
// [1] 꺼내자마자 DB에 원문 저장 (2차 임시 저장) - 실패 시 재시도
|
|
try {
|
|
$logId = $logModel->insert([
|
|
'raw_payload' => $rawData,
|
|
'status' => 'INIT'
|
|
]);
|
|
} catch (\Throwable $e) {
|
|
// MySQL 연결 계열 에러 시 재연결 후 재시도
|
|
if ($this->isMySqlConnectionError($e)) {
|
|
CLI::write(CLI::color('⚠️ MySQL gone away, reconnecting...', 'yellow'));
|
|
$this->db->close();
|
|
$this->db = \Config\Database::connect();
|
|
$this->db->initialize();
|
|
$logModel = model(NaverWorkerLogModel::class);
|
|
|
|
// 재시도
|
|
$logId = $logModel->insert([
|
|
'raw_payload' => $rawData,
|
|
'status' => 'INIT'
|
|
]);
|
|
} else {
|
|
throw $e; // 다른 에러면 그대로 throw
|
|
}
|
|
}
|
|
|
|
try {
|
|
$responseJson = json_decode($rawData, true);
|
|
$payload = $responseJson['request_data'] ?? [];
|
|
|
|
if (empty($payload)) {
|
|
throw new \Exception("빈 페이로드 데이터");
|
|
}
|
|
|
|
// 서비스의 함수 하나로 모든 처리 완료
|
|
$insertId = $naverService->processArticle($payload);
|
|
|
|
// [3] 성공 시 로그 업데이트 (재연결 처리 포함)
|
|
$this->safeUpdateLog($logModel, $logId, [
|
|
'atcl_no' => $payload['articleNumber'] ?? null,
|
|
'status' => 'SUCCESS',
|
|
'target_db_id' => $insertId
|
|
]);
|
|
|
|
CLI::write("✅ Success! DB ID: $insertId | Source: $source", 'cyan');
|
|
|
|
} catch (\Exception $e) {
|
|
CLI::error("❌ Task Failed: " . $e->getMessage());
|
|
|
|
// payload에서 매물번호 추출 시도
|
|
$atclNo = null;
|
|
try {
|
|
if (!empty($rawData)) {
|
|
$responseJson = json_decode($rawData, true);
|
|
$payload = $responseJson['request_data'] ?? [];
|
|
$atclNo = $payload['articleNumber'] ?? null;
|
|
}
|
|
} catch (\Exception $parseEx) {
|
|
// JSON 파싱 실패는 무시
|
|
}
|
|
|
|
// 실패 로그는 여기서 남김
|
|
// 1. DB 상태를 FAIL로 업데이트 (필수) (재연결 처리 포함)
|
|
$this->safeUpdateLog($logModel, $logId, [
|
|
'atcl_no' => $atclNo,
|
|
'status' => 'FAIL',
|
|
'error_msg' => $e->getMessage()
|
|
]);
|
|
|
|
// 2. Redis 실패 큐에 백업 (선택 - Redis가 있을 경우만)
|
|
if ($redis) {
|
|
try {
|
|
$redis->lPush('naver:failed_queue', $rawData);
|
|
} catch (\Exception $redisEx) {
|
|
// Redis 실패 시에도 에러 처리하지 않음 (이미 DB에 FAIL 로그 남김)
|
|
CLI::write(CLI::color('⚠️ Failed to push to failed_queue: ' . $redisEx->getMessage(), 'yellow'));
|
|
}
|
|
}
|
|
|
|
helper('log');
|
|
write_custom_log("FAILED_DATA | Error: " . $e->getMessage(), 'ERROR', 'failed');
|
|
|
|
// 루프 과부하 방지 (연속 에러 시)
|
|
sleep(1);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* MySQL 연결 계열 에러 발생 시 재연결 후 재시도하는 안전한 update
|
|
*/
|
|
protected function safeUpdateLog($logModel, $logId, $data)
|
|
{
|
|
try {
|
|
return $logModel->update($logId, $data);
|
|
} catch (\Throwable $e) {
|
|
if ($this->isMySqlConnectionError($e)) {
|
|
CLI::write(CLI::color('⚠️ MySQL gone away on update, reconnecting...', 'yellow'));
|
|
$this->db->close();
|
|
$this->db = \Config\Database::connect();
|
|
$this->db->initialize();
|
|
$logModel = model(\App\Models\Entities\NaverWorkerLogModel::class);
|
|
|
|
// 재시도
|
|
return $logModel->update($logId, $data);
|
|
} else {
|
|
throw $e;
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* MySQL 연결 끊김 계열 에러 여부 판별
|
|
*/
|
|
protected function isMySqlConnectionError(\Throwable $e): bool
|
|
{
|
|
$message = strtolower($e->getMessage());
|
|
|
|
return str_contains($message, 'mysql server has gone away')
|
|
|| str_contains($message, 'lost connection to mysql server')
|
|
|| str_contains($message, 'server has gone away');
|
|
}
|
|
|
|
/**
|
|
* 폴백 파일에서 데이터 읽기 (Redis 장애 시 파일에서 직접 처리)
|
|
*
|
|
* @return string|null JSON 데이터 또는 null
|
|
*/
|
|
protected function readFromFallbackFile()
|
|
{
|
|
$fallbackDir = ROOTPATH . 'worker/fallback_queue';
|
|
|
|
// 폴백 디렉토리가 없으면 null 반환
|
|
if (!is_dir($fallbackDir)) {
|
|
return null;
|
|
}
|
|
|
|
// 폴백 파일 목록 가져오기 (오래된 순서대로)
|
|
$files = glob($fallbackDir . '/*.json');
|
|
|
|
if (empty($files)) {
|
|
return null;
|
|
}
|
|
|
|
sort($files); // 파일명(타임스탬프) 기준 정렬
|
|
|
|
// 가장 오래된 파일 하나 처리
|
|
$filePath = $files[0];
|
|
|
|
try {
|
|
// 파일 락을 사용하여 읽기 (동시 접근 방지)
|
|
$fp = fopen($filePath, 'r');
|
|
if (!$fp) {
|
|
CLI::write(CLI::color("⚠️ Failed to open fallback file: " . basename($filePath), 'yellow'));
|
|
return null;
|
|
}
|
|
|
|
// 배타적 락 획득 시도
|
|
if (!flock($fp, LOCK_EX | LOCK_NB)) {
|
|
// 락 획득 실패 (다른 프로세스가 처리 중)
|
|
fclose($fp);
|
|
return null;
|
|
}
|
|
|
|
// 파일 내용 읽기
|
|
$content = stream_get_contents($fp);
|
|
|
|
// 파일 삭제 (처리 완료로 간주)
|
|
flock($fp, LOCK_UN);
|
|
fclose($fp);
|
|
unlink($filePath);
|
|
|
|
CLI::write(CLI::color("📂 Processing fallback file: " . basename($filePath), 'green'));
|
|
|
|
return $content;
|
|
|
|
} catch (\Exception $e) {
|
|
CLI::write(CLI::color("❌ Error reading fallback file " . basename($filePath) . ": " . $e->getMessage(), 'red'));
|
|
if (isset($fp) && is_resource($fp)) {
|
|
flock($fp, LOCK_UN);
|
|
fclose($fp);
|
|
}
|
|
return null;
|
|
}
|
|
}
|
|
|
|
|
|
} |