CLOAK/lib/Cloak/VisitorLogQueue.php

<?php
/**
 * 访客日志异步写入队列（Redis 优先，文件降级）
 */
class VisitorLogQueue
{
    const REDIS_KEY = 'cloak:visitor_log_queue';
    const FILE_DIR  = 'visitor_log_queue';

    /** @var bool|null */
    private static $redisAvailable = null;

    /**
     * @param array $job 需含 type: insert_full|enrich
     * @return bool
     */
    public static function push(array $job)
    {
        $job['queued_at'] = date('Y-m-d H:i:s');
        $payload          = json_encode($job, JSON_UNESCAPED_UNICODE);
        if ($payload === false) {
            return false;
        }

        $redis = self::redisClient();
        if ($redis) {
            try {
                return (bool) $redis->lPush(self::REDIS_KEY, $payload);
            } catch (Throwable $e) {
                self::logError('Redis push failed: ' . $e->getMessage());
            }
        }

        return self::pushFile($payload);
    }

    /**
     * @return array<int, array>
     */
    public static function popBatch(int $limit = 100)
    {
        $limit = max(1, min(500, $limit));
        $jobs  = [];

        $redis = self::redisClient();
        if ($redis) {
            try {
                for ($i = 0; $i < $limit; $i++) {
                    $raw = $redis->rPop(self::REDIS_KEY);
                    if ($raw === false || $raw === null) {
                        break;
                    }
                    $decoded = json_decode((string) $raw, true);
                    if (is_array($decoded)) {
                        $jobs[] = $decoded;
                    }
                }
                if ($jobs !== []) {
                    return $jobs;
                }
            } catch (Throwable $e) {
                self::logError('Redis pop failed: ' . $e->getMessage());
            }
        }

        return self::popFileBatch($limit);
    }

    /**
     * 请求 shutdown 时轻量 drain（最多 $limit 条）
     */
    public static function drain(int $limit = 5)
    {
        if (!class_exists('VisitorLogWorker', false)) {
            require_once __DIR__ . '/VisitorLogWorker.php';
        }
        VisitorLogWorker::processBatch($limit);
    }

    /**
     * @return Redis|null
     */
    private static function redisClient()
    {
        if (self::$redisAvailable === false) {
            return null;
        }
        if (!extension_loaded('redis') || !defined('REDIS_ENABLED') || strtoupper(REDIS_ENABLED) !== 'ON') {
            self::$redisAvailable = false;
            return null;
        }
        try {
            $r = new Redis();
            $timeout = defined('REDIS_TIMEOUT') ? (float) REDIS_TIMEOUT : 1.0;
            if (!@$r->connect(REDIS_HOST, (int) REDIS_PORT, $timeout)) {
                self::$redisAvailable = false;
                return null;
            }
            if (defined('REDIS_PASSWORD') && REDIS_PASSWORD !== '') {
                $r->auth(REDIS_PASSWORD);
            }
            self::$redisAvailable = true;
            return $r;
        } catch (Throwable $e) {
            self::$redisAvailable = false;
            return null;
        }
    }

    private static function queueDir()
    {
        $dir = dirname(__DIR__, 2) . '/storage/' . self::FILE_DIR;
        if (!is_dir($dir)) {
            @mkdir($dir, 0755, true);
        }
        return $dir;
    }

    private static function pushFile($payload)
    {
        $dir  = self::queueDir();
        $file = $dir . '/pending.ndjson';
        $fp   = @fopen($file, 'ab');
        if (!$fp) {
            return false;
        }
        if (flock($fp, LOCK_EX)) {
            fwrite($fp, $payload . "\n");
            fflush($fp);
            flock($fp, LOCK_UN);
        }
        fclose($fp);
        return true;
    }

    /**
     * @return array<int, array>
     */
    private static function popFileBatch(int $limit)
    {
        $dir  = self::queueDir();
        $file = $dir . '/pending.ndjson';
        if (!is_file($file)) {
            return [];
        }

        $fp = @fopen($file, 'c+');
        if (!$fp) {
            return [];
        }

        $jobs = [];
        if (!flock($fp, LOCK_EX)) {
            fclose($fp);
            return [];
        }

        $lines = [];
        while (($line = fgets($fp)) !== false) {
            $line = trim($line);
            if ($line !== '') {
                $lines[] = $line;
            }
        }

        if ($lines === []) {
            flock($fp, LOCK_UN);
            fclose($fp);
            return [];
        }

        $take    = array_slice($lines, 0, $limit);
        $remain  = array_slice($lines, $limit);
        ftruncate($fp, 0);
        rewind($fp);
        foreach ($remain as $line) {
            fwrite($fp, $line . "\n");
        }
        fflush($fp);
        flock($fp, LOCK_UN);
        fclose($fp);

        foreach ($take as $line) {
            $decoded = json_decode($line, true);
            if (is_array($decoded)) {
                $jobs[] = $decoded;
            }
        }

        return $jobs;
    }

    /**
     * @param array $job
     * @param string $error
     */
    public static function moveToDead(array $job, $error)
    {
        $dir = self::queueDir() . '/dead';
        if (!is_dir($dir)) {
            @mkdir($dir, 0755, true);
        }
        $job['dead_at']  = date('Y-m-d H:i:s');
        $job['dead_error'] = (string) $error;
        $name = $dir . '/' . date('Ymd_His') . '_' . substr(md5(json_encode($job)), 0, 8) . '.json';
        @file_put_contents($name, json_encode($job, JSON_UNESCAPED_UNICODE | JSON_PRETTY_PRINT));
        self::logError('Dead letter: ' . $error);
    }

    public static function logError($message)
    {
        $handle = @fopen(dirname(__DIR__, 2) . '/err.txt', 'a+');
        if ($handle) {
            fwrite($handle, date('Y-m-d H:i:s') . ' | VisitorLogQueue | ' . $message . "\n");
            fclose($handle);
        }
    }
}