CLOAK/lib/Cloak/VisitorLogWorker.php

<?php
/**
 * 访客日志队列消费（INSERT / enrich UPDATE）
 */
require_once dirname(__DIR__) . '/DbHelper.php';
require_once __DIR__ . '/VisitorLogSchema.php';
require_once __DIR__ . '/VisitorLogQueue.php';
require_once __DIR__ . '/VisitorLogVtimes.php';

class VisitorLogWorker
{
    /**
     * @return array{processed:int, failed:int}
     */
    public static function processBatch(int $limit = 100)
    {
        $jobs      = VisitorLogQueue::popBatch($limit);
        $processed = 0;
        $failed    = 0;

        foreach ($jobs as $job) {
            $ok = false;
            for ($attempt = 1; $attempt <= 3; $attempt++) {
                if (self::processJob($job)) {
                    $ok = true;
                    break;
                }
                usleep(100000 * $attempt);
            }
            if ($ok) {
                $processed++;
            } else {
                $failed++;
                VisitorLogQueue::moveToDead($job, 'process failed after retries');
            }
        }

        return ['processed' => $processed, 'failed' => $failed];
    }

    /**
     * @param array $job
     */
    public static function processJob(array $job)
    {
        $type = $job['type'] ?? '';
        if ($type === 'insert_full') {
            return self::insertFull($job['logs'] ?? []) !== null;
        }
        if ($type === 'enrich') {
            return self::enrich((int) ($job['log_id'] ?? 0), $job['logs'] ?? []);
        }
        if ($type === 'update') {
            return self::updateFromLogs((int) ($job['log_id'] ?? 0), $job['logs'] ?? []);
        }
        return false;
    }

    /**
     * @param array $logs
     * @return int|null insert_id
     */
    public static function insertFull(array $logs)
    {
        $conn = self::connect();
        if (!$conn) {
            return null;
        }
        VisitorLogSchema::ensureColumns($conn);
        VisitorLogSchema::ensureIndexes($conn);

        $logs = VisitorLogVtimes::mergeIntoLogs($logs, $conn);
        $data = self::escapeLogs($conn, $logs);
        $d    = static function ($key) use ($data, $conn) {
            return array_key_exists($key, $data) ? $data[$key] : cloak_db_escape($conn, '');
        };
        list($simCols, $simVals) = self::simulationInsertFragment($data);

        $sql = "INSERT INTO `visitor_logs`(`campagin_id`, `visit_md5_code`, `domain`, `visit_date`, `IP`, `country`, `result`, `reason`, `referer`, `vtimes`, `client`, `browser`, `page`, `language`, `user_agent`, `http_referer`, `accept_language_raw`, `judge_timing`, `fp_url`, `device`, `api_by_request`, `api_by_response`, `api_risk_request`, `api_risk_response`{$simCols}) VALUES ('"
            . $d('campagin_id') . "','"
            . $d('visit_md5_code') . "','"
            . $d('site_name') . "','"
            . ($data['visit_date'] ?? date('Y-m-d H:i:s')) . "','"
            . $d('customers_ip') . "','"
            . $d('country') . "','"
            . $d('result') . "','"
            . $d('reason') . "','"
            . $d('v_referer') . "','"
            . (int) ($data['vtimes'] ?? 1) . "','"
            . $d('Client') . "','"
            . $d('v_Browser') . "','"
            . $d('v_PageURL') . "','"
            . $d('accept_language') . "','"
            . $d('user_agent') . "','"
            . $d('http_referer') . "','"
            . $d('accept_language_raw') . "','"
            . $d('judge_timing') . "','"
            . $d('fp_url') . "','"
            . $d('device') . "','"
            . $d('api_by_request') . "','"
            . $d('api_by_response') . "','"
            . $d('api_risk_request') . "','"
            . $d('api_risk_response') . "'"
            . $simVals . ')';

        if ($conn->query($sql) !== true) {
            VisitorLogQueue::logError('insert_full: ' . $conn->error . ' | ' . $sql);
            $conn->close();
            return null;
        }
        $id = (int) $conn->insert_id;
        $conn->close();
        return $id > 0 ? $id : null;
    }

    /**
     * 同步最小 INSERT（二次风控 wait，需立即返回 log_id）
     *
     * @param array $logs
     * @return int|null
     */
    public static function insertMinimal(array $logs)
    {
        $conn = self::connect();
        if (!$conn) {
            return null;
        }
        VisitorLogSchema::ensureColumns($conn);
        VisitorLogSchema::ensureIndexes($conn);

        $logs = VisitorLogVtimes::mergeIntoLogs($logs, $conn);
        $data = self::escapeLogs($conn, $logs);
        $d    = static function ($key) use ($data, $conn) {
            return array_key_exists($key, $data) ? $data[$key] : cloak_db_escape($conn, '');
        };
        list($simCols, $simVals) = self::simulationInsertFragment($data);

        $sql = "INSERT INTO `visitor_logs`(`campagin_id`, `visit_md5_code`, `domain`, `visit_date`, `IP`, `country`, `result`, `reason`, `vtimes`, `client`, `browser`, `language`, `fp_url`, `device`{$simCols}) VALUES ('"
            . $d('campagin_id') . "','"
            . $d('visit_md5_code') . "','"
            . $d('site_name') . "','"
            . date('Y-m-d H:i:s') . "','"
            . $d('customers_ip') . "','"
            . $d('country') . "','"
            . $d('result') . "','"
            . $d('reason') . "','"
            . (int) ($data['vtimes'] ?? 1) . "','"
            . $d('Client') . "','"
            . $d('v_Browser') . "','"
            . $d('accept_language') . "','"
            . $d('fp_url') . "','"
            . $d('device') . "'"
            . $simVals . ')';

        if ($conn->query($sql) !== true) {
            VisitorLogQueue::logError('insertMinimal: ' . $conn->error);
            $conn->close();
            return null;
        }
        $id = (int) $conn->insert_id;
        $conn->close();
        return $id > 0 ? $id : null;
    }

    /**
     * @param int   $logId
     * @param array $logs
     */
    public static function enrich($logId, array $logs)
    {
        if ($logId <= 0) {
            return false;
        }
        $conn = self::connect();
        if (!$conn) {
            return false;
        }

        $data = self::escapeLogs($conn, $logs);
        $sets = [];
        $map  = [
            'referer'             => 'v_referer',
            'page'                => 'v_PageURL',
            'user_agent'          => 'user_agent',
            'http_referer'        => 'http_referer',
            'accept_language_raw' => 'accept_language_raw',
            'judge_timing'        => 'judge_timing',
            'api_by_request'      => 'api_by_request',
            'api_by_response'     => 'api_by_response',
            'api_risk_request'    => 'api_risk_request',
            'api_risk_response'   => 'api_risk_response',
        ];
        foreach ($map as $col => $logKey) {
            if (array_key_exists($logKey, $data)) {
                $sets[] = "`{$col}`='" . $data[$logKey] . "'";
            } elseif (array_key_exists($col, $data)) {
                $sets[] = "`{$col}`='" . $data[$col] . "'";
            }
        }
        if ($sets === []) {
            $conn->close();
            return true;
        }

        $sql = 'UPDATE `visitor_logs` SET ' . implode(', ', $sets) . ' WHERE `id`=' . (int) $logId;
        $ok  = $conn->query($sql) === true;
        if (!$ok) {
            VisitorLogQueue::logError('enrich: ' . $conn->error);
        }
        $conn->close();
        return $ok;
    }

    /**
     * 续步 UPDATE：合并同一次 visit 内的日志字段（不修改 vtimes）。
     *
     * @param int   $logId
     * @param array $logs
     */
    public static function updateFromLogs($logId, array $logs)
    {
        $logId = (int) $logId;
        if ($logId <= 0) {
            return false;
        }
        $conn = self::connect();
        if (!$conn) {
            return false;
        }
        VisitorLogSchema::ensureColumns($conn);

        $data = self::escapeLogs($conn, $logs);
        $sets = [];
        $map  = [
            'country'             => 'country',
            'referer'             => 'v_referer',
            'client'              => 'Client',
            'browser'             => 'v_Browser',
            'page'                => 'v_PageURL',
            'language'            => 'accept_language',
            'user_agent'          => 'user_agent',
            'http_referer'        => 'http_referer',
            'accept_language_raw' => 'accept_language_raw',
            'judge_timing'        => 'judge_timing',
            'fp_url'              => 'fp_url',
            'device'              => 'device',
            'api_by_request'      => 'api_by_request',
            'api_by_response'     => 'api_by_response',
            'api_risk_request'    => 'api_risk_request',
            'api_risk_response'   => 'api_risk_response',
        ];
        foreach (['result', 'reason', 'fp_url'] as $col) {
            if (array_key_exists($col, $logs)) {
                $key = array_key_exists($col, $data) ? $col : $col;
                $logKey = $col === 'fp_url' ? 'fp_url' : $col;
                $val = array_key_exists($logKey, $data) ? $data[$logKey] : cloak_db_escape($conn, $logs[$col]);
                $sets[] = "`{$col}`='" . $val . "'";
            }
        }
        foreach ($map as $col => $logKey) {
            if (!array_key_exists($logKey, $data)) {
                continue;
            }
            if ((string) $data[$logKey] === '') {
                continue;
            }
            $sets[] = "`{$col}`='" . $data[$logKey] . "'";
        }
        foreach (['fp_hdata', 'api_risk_request', 'api_risk_response'] as $col) {
            if (!array_key_exists($col, $logs)) {
                continue;
            }
            $val = array_key_exists($col, $data) ? $data[$col] : cloak_db_escape($conn, $logs[$col]);
            $sets[] = "`{$col}`='" . $val . "'";
        }
        if ($sets === []) {
            $conn->close();
            return true;
        }

        $sql = 'UPDATE `visitor_logs` SET ' . implode(', ', array_unique($sets)) . ' WHERE `id`=' . $logId;
        $ok  = $conn->query($sql) === true;
        if (!$ok) {
            VisitorLogQueue::logError('updateFromLogs: ' . $conn->error);
        }
        $conn->close();
        return $ok;
    }

    /**
     * @param mysqli $conn
     * @param array  $logs
     * @return array<string, string>
     */
    private static function escapeLogs($conn, array $logs)
    {
        $data = [];
        foreach ($logs as $key => $value) {
            $data[$key] = cloak_db_escape($conn, $value);
        }
        if (!isset($data['visit_date'])) {
            $data['visit_date'] = date('Y-m-d H:i:s');
        }
        return $data;
    }

    /**
     * @param array<string, string> $data escaped log fields
     * @return array{0:string,1:string}
     */
    private static function simulationInsertFragment(array $data)
    {
        if (empty($data['is_simulation']) && !array_key_exists('sim_source_log_id', $data)) {
            return ['', ''];
        }
        $cols = ', `is_simulation`, `sim_source_log_id`';
        $isSim = (int) ($data['is_simulation'] ?? 0);
        if (!empty($data['sim_source_log_id'])) {
            $vals = ", '{$isSim}', '" . (int) $data['sim_source_log_id'] . "'";
        } else {
            $vals = ", '{$isSim}', NULL";
        }
        return [$cols, $vals];
    }

    /**
     * @return mysqli|null
     */
    private static function connect()
    {
        if (!defined('DB_USERNAME') || !defined('DB_PASSWORD') || !defined('DB_NAME')) {
            return null;
        }
        $conn = @new mysqli('localhost', DB_USERNAME, DB_PASSWORD, DB_NAME);
        if ($conn->connect_error) {
            VisitorLogQueue::logError('DB connect: ' . $conn->connect_error);
            return null;
        }
        return $conn;
    }
}