<?php

namespace XZD\AI\V5\Core;

/**
 * 轻量级待处理队列
 *
 * 用 option 替代 wp_schedule_single_event + spawn_cron，
 * Action Scheduler 每 30 秒轮询处理，消除 spawn_cron() DNS 阻塞。
 */
class PendingQueue
{
    private const OPTION_KEY = 'xzd_pending_queue';
    private const MAX_ITEMS = 500;

    /**
     * 添加流水线任务到队列
     * 新发布文章始终入队，其他按总积压限制
     */
    public static function addPipeline(int $postId, string $event): void
    {
        if (class_exists(\XZD\AI\V5\Tools\QueueManager::class)
            && !\XZD\AI\V5\Tools\QueueManager::shouldAcceptTask($event)) {
            return;
        }

        self::add([
            'type'      => 'pipeline',
            'post_id'   => $postId,
            'event'     => $event,
            'queued_at' => time(),
        ]);
    }

    /**
     * 添加规则评估任务到队列
     *
     * 上下文数据（可能较大）单独存 transient，队列中只存引用 key
     */
    public static function addRuleEvaluation(string $event, array $context): void
    {
        // 大上下文存 transient，避免 option 膨胀
        $ctxKey = 'xzd_rctx_' . substr(md5($event . serialize(array_keys($context)) . microtime(true)), 0, 12);
        set_transient($ctxKey, $context, 300);

        self::add([
            'type'        => 'rule_evaluate',
            'event'       => $event,
            'context_key' => $ctxKey,
            'queued_at'   => time(),
        ]);
    }

    /**
     * 取出并清空队列的前 N 条
     *
     * @return array 队列条目数组
     */
    public static function drain(int $batchSize = 10): array
    {
        // 原子 Lock 自旋保护读-改-写，防止并发 drain 丢数据
        $lock = self::acquireLock();
        if (!$lock) {
            return [];
        }

        try {
            $queue = Config::getArrayOption(self::OPTION_KEY);
            if (empty($queue)) {
                return [];
            }

            $batch     = array_slice($queue, 0, $batchSize);
            $remaining = array_slice($queue, $batchSize);

            if (empty($remaining)) {
                delete_option(self::OPTION_KEY);
            } else {
                update_option(self::OPTION_KEY, array_values($remaining), false);
            }

            return $batch;
        } finally {
            $lock->release();
        }
    }

    /**
     * 新文章优先drain — 双通道分离
     * 通道A（优先）：publish_post / rule_evaluate，不限量
     * 通道B（节流）：历史文章，按limit限制
     *
     * @param int $oldLimit 通道B最大数量（0=只取新文章）
     * @return array ['priority' => [...], 'normal' => [...]]
     */
    public static function drainNewFirst(int $oldLimit = 5): array
    {
        // 原子 Lock 自旋，与 add() 共用同一把锁防止竞态
        $lock = self::acquireLock();
        if (!$lock) {
            return ['priority' => [], 'normal' => []];
        }

        try {
            $queue = Config::getArrayOption(self::OPTION_KEY);
            if (empty($queue)) {
                return ['priority' => [], 'normal' => []];
            }

            $priority = []; // 新文章
            $normal   = []; // 历史文章
            $remaining = [];

            foreach ($queue as $item) {
                $event = $item['event'] ?? '';
                $type = $item['type'] ?? '';
                $isPriority = ($event === 'publish_post' || $type === 'rule_evaluate');

                if ($isPriority) {
                    $priority[] = $item;
                } elseif (count($normal) < $oldLimit) {
                    $normal[] = $item;
                } else {
                    $remaining[] = $item;
                }
            }

            if (empty($remaining)) {
                delete_option(self::OPTION_KEY);
            } else {
                update_option(self::OPTION_KEY, array_values($remaining), false);
            }

            return ['priority' => $priority, 'normal' => $normal];
        } finally {
            $lock->release();
        }
    }

    /**
     * 队列当前大小
     */
    public static function size(): int
    {
        $queue = Config::getArrayOption(self::OPTION_KEY);
        return count($queue);
    }

    /**
     * 清空队列
     */
    public static function clear(): void
    {
        delete_option(self::OPTION_KEY);
    }

    // 内部方法

    /**
     * 原子自旋获取队列锁
     * 基于 Lock::for + token 校验，确保真正互斥
     *
     * @return Lock|null 获取失败返回 null
     */
    private static function acquireLock(int $maxWaits = 20, int $ttl = 5): ?Lock
    {
        $lock = Lock::for('xzd_pq_lock', $ttl);
        for ($waits = 0; $waits < $maxWaits; $waits++) {
            if ($lock->acquire()) {
                return $lock;
            }
            usleep(100000); // 100ms后重试
        }
        return null;
    }

    private static function add(array $item): void
    {
        // 原子 Lock 自旋，锁获取失败时放弃写入，防止与 drain() 并发导致 option 交错覆盖
        $lock = self::acquireLock();
        if (!$lock) {
            Logger::info('pending_queue', 'PendingQueue 锁获取超时，放弃写入以保护数据一致性', [
                'post_id' => $item['post_id'] ?? 0,
                'event'   => $item['event'] ?? '',
            ]);
            return; // 放弃写入，由上游调用者重试
        }

        try {
            $queue = Config::getArrayOption(self::OPTION_KEY);

            // 去重：同一类型 + 同一主体 只保留最新一条
            $dedupKey = self::dedupKey($item);
            $queue    = array_values(array_filter($queue, function ($existing) use ($dedupKey) {
                return self::dedupKey($existing) !== $dedupKey;
            }));

            $queue[] = $item;

            // 防止异常积压撑爆 option
            $queueSize = count($queue);
            if ($queueSize > self::MAX_ITEMS) {
                $dropped = $queueSize - self::MAX_ITEMS;
                $queue = array_slice($queue, -self::MAX_ITEMS);

                // 节流日志（30秒内不重复）
                $logThrottle = get_transient('xzd_pending_queue_overflow_log');
                if ($logThrottle === false) {
                    set_transient('xzd_pending_queue_overflow_log', time(), 30);
                    Logger::warningOnce('pending_queue', "PendingQueue 溢出，丢弃 {$dropped} 个旧任务", [
                        'queue_size' => $queueSize, 'max' => self::MAX_ITEMS, 'dropped' => $dropped,
                    ]);
                }

                // 管理员通知（每天最多3次）
                \XZD\AI\V5\Core\NotificationThrottle::send(
                    'pending_overflow',
                    'PendingQueue 队列溢出',
                    sprintf(
                        "待处理队列(PendingQueue)溢出，丢弃 %d 个旧任务。\n当前队列大小: %d (上限: %d)\n请检查: %s",
                        $dropped, $queueSize, self::MAX_ITEMS,
                        admin_url('admin.php?page=xzdbk-ai')
                    )
                );
            }

            update_option(self::OPTION_KEY, array_values($queue), false);
        } finally {
            if ($lock) {
                $lock->release();
            }
        }

        // 入队即触发处理 — 5秒节流，通过REST heartbeat端点触发
        $throttleKey = 'xzd_process_trigger_throttle';
        if (!get_transient($throttleKey) && isset($queue) && count($queue) > 0) {
            set_transient($throttleKey, time(), 5);
            // 异步非阻塞触发心跳tick，不拖慢当前请求
            wp_remote_post(rest_url('xzdbk-ai/v1/heartbeat/tick'), [
                'timeout'   => 0.5,
                'blocking'  => false,
                'sslverify' => false,
            ]);
        }
    }

    private static function dedupKey(array $item): string
    {
        if ($item['type'] === 'pipeline') {
            return "pipe_{$item['post_id']}_{$item['event']}";
        }
        return "rule_{$item['event']}_" . ($item['context_key'] ?? '');
    }
}
