<?php
namespace app\common\service;

use Predis\Client as RedisClient;

class QueueService
{
    protected $redis;

    public function __construct()
    {
        $this->redis = new RedisClient([
            'host' => '127.0.0.1',
            'port' => 6379,
            // 'password' => '', // 如果有密码
        ]);
    }

    // 推送消息到队列
    public function push($queueName, $message)
    {
        $this->redis->rpush($queueName, json_encode($message));
    }

    // 从队列中取出消息
    public function pop($queueName)
    {
        $message = $this->redis->lpop($queueName);
        return $message ? json_decode($message, true) : null;
    }

    /**
     * 阻塞式取出消息：队列空时挂在 Redis 上等，有消息立刻返回。
     *
     * 相比 pop() + sleep(1) 的轮询，取到消息的延迟从「最多 1 秒」降到接近 0，
     * 对「下单 -> 调上游发放」这种链路首段能实打实省掉一截等待。
     *
     * @param string $queueName
     * @param int    $timeout 最长阻塞秒数（到点返回 null，便于外层做兜底巡检）
     * @return array|string|null
     */
    public function bpop($queueName, $timeout = 5)
    {
        // BLPOP 会把连接挂住 $timeout 秒，必须短于 Redis 连接的读超时
        // （未显式配置时取 PHP default_socket_timeout，通常 60 秒），
        // 否则会抛读超时异常而不是正常返回 null。这里夹到 30 秒以内。
        $timeout = max(1, min(30, (int) $timeout));
        // Predis 的 blpop 返回 [队列名, 消息]，超时返回 null
        $res = $this->redis->blpop([$queueName], $timeout);
        if (empty($res[1])) {
            return null;
        }
        $decoded = json_decode($res[1], true);
        return $decoded === null ? $res[1] : $decoded;
    }

    /**
     * 延迟推送（ZSET）
     * @param string $queueName
     * @param array|string $data
     * @param int $delay 延迟秒数
     */
    public function laterPush($queueName, $data, $delay)
    {
        $score = time() + $delay;
        $payload = is_string($data) ? $data : json_encode($data, JSON_UNESCAPED_UNICODE);
        return $this->redis->zadd($queueName . ':delayed', [$payload => $score]);
    }

    /**
     * 检查延迟队列，将到期任务迁移到普通队列
     * 并发安全：先 zrem，只有「移除成功（返回>0）」的进程才把任务放进就绪队列，
     * 避免多个消费者同时迁移导致同一任务被投递多次。
     * @param string $queueName
     */
    public function migrateDelayedToReady($queueName)
    {
        $now = time();
        $tasks = $this->redis->zrangebyscore($queueName . ':delayed', 0, $now);
        if (empty($tasks)) {
            return;
        }
        foreach ($tasks as $task) {
            // 先抢占式移除，再入队
            if ($this->redis->zrem($queueName . ':delayed', $task) > 0) {
                $this->push($queueName, $task);
            }
        }
    }
}