<?php

namespace Php\Queue;

use Redis;

/**
 * 后续
 * 1.提取不同模式的队列代码
 * 2.win单进程
 * 3.liunx多进程模式/单进程
 * 4.swoole模式下多进程使用swoole提供的进程管理
 */
class Query
{
	/**
	 * 主队列
	 *
	 * @var int
	 */
	const READY = 2;
	
	/**
	 * 延时队列
	 *
	 * @var int
	 */
	const DELAY = 1;
	
	/**
	 * 工作队列
	 *
	 * @var int
	 */
	const WROK = 3;
	
	/**
	 * 失败队列
	 *
	 * @var int
	 */
	const FAIL = 4;
	
	/**
	 * 队列后缀
	 *
	 * @var string[]
	 */
	const LISTNAME = [
		1 => '--delay',
		2 => '--list',
		3 => '--work',
		4 => '--fail',
	];

	/**
	 * 队列名称
	 *
	 * @var string
	 */
	public $queryName = 'default';
	
	/**
	 * 是否是swoole环境
	 *
	 * @var bool
	 */
	public $isSoole = false;
	
	/**
	 * redis配置
	 *
	 * @var RedisConfig
	 */
	public $config;
	
	/**
	 * 重试次数默认3次
	 *
	 * @var int
	 */
	private $retryNum = 3;
	
	public function __construct(RedisConfig $config)
	{
		$this->config = $config;
		if (extension_loaded('swoole')) {
			$this->isSoole = true;
		}
	}
	
	/**
	 * 设置重试次数
	 *
	 * @author chenhuan 2023年1月9日
	 * @param int $num
	 * @return $this
	 */
	public function setRetryNum(int $num)
	{
		$this->retryNum = $num;
		return $this;
	}
	
	/**
	 * 加入队列
	 *
	 * @author chenhuan 2023年1月6日
	 * @param string $job 执行的文件
	 * @param array $data 数据
	 * @param int $delay 延时时间
	 * @return int
	 */
	public function push(string $job, array $data, int $delay = 0): int
	{
		$redis = RedisDriver::connect($this->config, $this->isSoole);
		
		if ($delay > 0) {
			$args = [
				$this->getQueueName(self::DELAY),
				$this->getMessageKeyPrefix(),
				$this->getMessageIdKey(),
				microtime(true) + $delay,
				date('Ymd'),
			];
			foreach ($data as $k => $v) {
				$args[] = $k;
				$args[] = $v;
			}
			$args[] = 'job';
			$args[] = $job;
			
			$result = $redis->eval(<<<'LUA'
 local queueKey = KEYS[1]
 local messageKeyPrefix = KEYS[2]
 local messageIdKey = KEYS[3]
 local delayTo = ARGV[1]
 local date = ARGV[2]
 -- 创建消息id
 local messageId = redis.call('hIncrby', messageIdKey, date, 1);
 if messageId > 0 then
 messageId = date .. messageId
 else
 return false
 end
 -- 创建消息
 local messageKey = messageKeyPrefix .. messageId;
 local ARGVLength = table.getn(ARGV)
 for i=3,ARGVLength,2 do
 redis.call('hset', messageKey, ARGV[i], ARGV[i + 1])
 end
 redis.call('hset', messageKey, 'messageId', messageId)
 redis.call('hset', messageKey, 'messageId', messageId)
 -- 加入延时队列
 redis.call('zadd', queueKey, delayTo, messageId);
 return messageId
 LUA, $args, 3);
		} else {
			$args = [
				$this->getQueueName(self::READY),
				$this->getMessageKeyPrefix(),
				$this->getMessageIdKey(),
				date('Ymd'),
			];
			foreach ($data as $k => $v) {
				$args[] = $k;
				$args[] = $v;
			}
			$args[] = 'job';
			$args[] = $job;
			$result = $redis->eval(<<<'LUA'
 local queueKey = KEYS[1]
 local messageKeyPrefix = KEYS[2]
 local messageIdKey = KEYS[3]
 local date = ARGV[1]
 -- 创建消息id
 local messageId = redis.call('hIncrby', messageIdKey, date, 1);
 if messageId > 0 then
 messageId = date .. messageId
 else
 return false
 end
 -- 创建消息
 local messageKey = messageKeyPrefix .. messageId;
 local ARGVLength = table.getn(ARGV)
 for i=2,ARGVLength,2 do
 redis.call('hset', messageKey, ARGV[i], ARGV[i + 1])
 end
 redis.call('hset', messageKey, 'messageId', messageId)
 --重试次数
 redis.call('hset', messageKey, 'retryNum', 0)
 -- 加入队列
 redis.call('rpush', queueKey, messageId);
 return messageId
 LUA, $args, 3);
		}
		if ($this->isSoole) {
			RedisDriver::giveBackRedis($redis);
		}
		return $result;
	}
	
	/**
	 * 获取队列名
	 *
	 * @author chenhuan 2023年1月6日
	 * @param int $type
	 * @return string
	 */
	private function getQueueName(int $type = 1): string
	{
		return 'queue:' . $this->queryName . self::LISTNAME[$type];
	}
	
	/**
	 * 获取消息键前缀
	 *
	 * @author chenhuan 2023年1月6日
	 * @return string
	 */
	private function getMessageKeyPrefix(): string
	{
		return 'queue:message:';
	}
	
	/**
	 * 获取消息 ID 的键
	 *
	 * @author chenhuan 2023年1月6日
	 * @return string
	 */
	private function getMessageIdKey(): string
	{
		return 'queue:message:id';
	}
	
	/**
	 * 消费
	 *
	 * @author chenhuan 2023年1月9日
	 * @param string $queueName
	 * @return mixed
	 */
	public function work(string $queueName = 'default')
	{
		$redis = RedisDriver::connect($this->config, $this->isSoole);
		$this->getQueue($queueName);
		
		while (1) {
			// 将达到指定时间的消息加入到队列.
			$this->migrateExpiredJobs($redis, 100);
			$result = $this->pop($redis);
			if ($result) {
				if ($this->isSoole) {
					go(function () use ($result, $queueName) {
						$class = $result['job'];
						unset($result['job']);
						$class = new $class;
						$class->job($this, $result);
					});
				} else {
					$class = $result['job'];
					unset($result['job']);
					$class = new $class;
					$class->job($this, $result);
				}
			} else {
				sleep(2);
				// 当队列中没有消息时,将没有标记成功的消息恢复到队列中
				$this->recoverMessage($redis, 100);
			}
		}
	}
	
	public function getQueue($queryName)
	{
		$this->queryName = $queryName;
		return $this;
	}
	
	/**
	 * 将消息恢复到队列中
	 *
	 * @author chenhuan 2023年1月6日
	 * @return mixed
	 */
	public function recoverMessage($redis, $count = 100)
	{
		$argv = [
			$this->getQueueName(self::WROK),
			$this->getQueueName(self::READY),
			$this->getQueueName(self::FAIL),
			$this->getMessageKeyPrefix(),
			$this->retryNum,
			microtime(true),
			$count,
		];
//		local messageIds = redis.call('ZREVRANGE', KEYS[1], 0, 100)
		return $redis->eval(<<<'LUA'
 -- 查询消息ID
 local messageIds = redis.call('zrevrangebyscore', KEYS[1], ARGV[2], 0, 'limit', 0, ARGV[3])
 local messageIdCount = table.getn(messageIds)
 if 0 == messageIdCount then
 return 0
 end
 for i, v in pairs(messageIds) do
 local retryNum = redis.call('hget', KEYS[4] .. v, 'retryNum')
 if retryNum < ARGV[1] then
 redis.call('hIncrby', KEYS[4] .. v, 'retryNum', 1);
 redis.call('rpush', KEYS[2], v)
 redis.call('zrem', KEYS[1], v)
 else
 -- 超过重试次数后,从工作队列中删除,加入失败队列
 redis.call('zadd', KEYS[3], ARGV[2], v)
 redis.call('zrem', KEYS[1], v)
 end
 end
 return messageIds
 LUA, $argv, 4);
	}
	
	/**
	 * 将达到指定时间的消息加入到队列.
	 *
	 * @author chenhuan 2023年1月6日
	 * @param Redis $redis
	 * @param int $count
	 * @return int
	 */
	public function migrateExpiredJobs(\Redis $redis, int $count = 100)
	{
		$argv = [
			$this->getQueueName(self::READY),
			$this->getQueueName(self::DELAY),
			microtime(true),
			$count,
		];
		
		return $redis->eval(<<<'LUA'
 -- 查询消息ID
 local messageIds = redis.call('zrevrangebyscore', KEYS[2], ARGV[1], 0, 'limit', 0, ARGV[2])
 local messageIdCount = table.getn(messageIds)
 if 0 == messageIdCount then
 return 0
 end
 for i, v in pairs(messageIds) do
 redis.call('rpush', KEYS[1], v)
 redis.call('zrem', KEYS[2], v)
 end
 return messageIds
 LUA, $argv, 2);
	}
	
	/**
	 * 从队列中弹出消息
	 *
	 * @author chenhuan 2023年1月6日
	 * @param Redis $redis
	 * @return array|null
	 */
	public function pop(Redis $redis): ?array
	{
		$result = $redis->eval(<<<'LUA'
 -- 从列表弹出
 local messageId = redis.call('lpop', KEYS[1])
 if false == messageId then
 return -1
 end
 -- 获取消息内容
 local hashResult = redis.call('hgetall', KEYS[3] .. messageId)
 redis.call('zadd', KEYS[2], ARGV[1], messageId)
 return hashResult
 LUA, [
			$this->getQueueName(self::READY),
			$this->getQueueName(self::WROK),
			$this->getMessageKeyPrefix(),
			microtime(true)+600,
		], 3);
		
		
		if ($result && \is_array($result)) {
			$data = [];
			$length = count($result);
			for ($i = 0; $i < $length; $i += 2) {
				$data[$result[$i]] = $result[$i + 1];
			}
			return $data;
		}
		return null;
	}
	
	/**
	 * 标记消息成功
	 *
	 * @author chenhuan 2023年1月6日
	 * @param int $messageId
	 * @return void
	 */
	public function success(int $messageId = 0)
	{
		$redis = RedisDriver::connect($this->config, $this->isSoole);
		
		$result = $redis->eval(<<<'LUA'
 -- 从工作队列删除
 redis.call('zrem', KEYS[1], ARGV[1])
 -- 删除消息
 redis.call('del', KEYS[2] .. ARGV[1])
 return true
 LUA, [
			$this->getQueueName(self::WROK),
			$this->getMessageKeyPrefix(),
			$messageId,
		], 2);
		if ($this->isSoole) {
			RedisDriver::giveBackRedis($redis);
		}
		return $result;
	}
	
	/**
	 * 将失败的消息恢复到主队列中
	 *
	 * @author chenhuan 2023年1月9日
	 * @return int
	 */
	public function recoverFailMessage(): int
	{
		$redis = RedisDriver::connect($this->config, $this->isSoole);
		$result = $redis->eval(<<<'LUA'
 -- 获取失败队列中的消息
 local messageIds = redis.call('ZREVRANGE', KEYS[2], 0, 100)
 local messageIdCount = table.getn(messageIds)
 if 0 == messageIdCount then
 return 0
 end
 for i, v in pairs(messageIds) do
 redis.call('hset', KEYS[3] .. v, 'retryNum', 0);
 redis.call('rpush', KEYS[1], v)
 redis.call('zrem', KEYS[2], v)
 end
 return messageIdCount
 LUA, [
			$this->getQueueName(self::READY),
			$this->getQueueName(self::FAIL),
			$this->getMessageKeyPrefix(),
		], 3);
		if ($this->isSoole) {
			RedisDriver::giveBackRedis($redis);
		}
		return $result;
	}
}