<?php
// +----------------------------------------------------------------------
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
// +----------------------------------------------------------------------
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
// +----------------------------------------------------------------------
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
// +----------------------------------------------------------------------
// | Author: yunwuxin <448901948@qq.com>
// +----------------------------------------------------------------------

namespace think\queue\connector;

use think\exception\HttpException;
use think\queue\Connector;
use think\Request;
use think\queue\job\Topthink as TopthinkJob;
use think\Response;

class Topthink extends Connector
{
    protected $options = [
        'token'       => '',
        'project_id'  => '',
        'protocol'    => 'https',
        'host'        => 'qns.topthink.com',
        'port'        => 443,
        'api_version' => 1,
        'max_retries' => 3,
        'default'     => 'default'
    ];

    /** @var  Request */
    protected $request;

    protected $url;

    protected $curl = null;

    protected $last_status;

    protected $headers = [];

    public function __construct($options)
    {
        if (!empty($options)) {
            $this->options = array_merge($this->options, $options);
        }

        $this->url = "{$this->options['protocol']}://{$this->options['host']}:{$this->options['port']}/v{$this->options['api_version']}/";

        $this->headers['Authorization'] = "Bearer {$this->options['token']}";

        $this->request = Request::instance();
    }

    public function push($job, $data = '', $queue = null)
    {
        return $this->pushRaw(0, $queue, $this->createPayload($job, $data));
    }

    public function later($delay, $job, $data = '', $queue = null)
    {
        return $this->pushRaw($delay, $queue, $this->createPayload($job, $data));
    }

    public function release($queue, $job, $delay)
    {
        return $this->pushRaw($delay, $queue, $job->payload, $job->attempts);
    }

    public function marshal()
    {
        $job = new TopthinkJob($this, $this->marshalPushedJob(), $this->request->header('topthink-message-queue'));
        if ($this->request->header('topthink-message-status') == 'success') {
            $job->fire();
        } else {
            $job->failed();
        }
        return new Response('OK');
    }

    public function pushRaw($delay, $queue, $payload, $attempts = 0)
    {
        $queue_name = $this->getQueue($queue);
        $queue      = rawurlencode($queue_name);
        $url        = "project/{$this->options['project_id']}/queue/{$queue}/message";
        $message    = [
            'payload'  => $payload,
            'attempts' => $attempts,
            'delay'    => $delay
        ];

        return $this->apiCall('POST', $url, $message)->id;
    }

    public function deleteMessage($queue, $id)
    {
        $queue = rawurlencode($queue);
        $url   = "project/{$this->options['project_id']}/queue/{$queue}/message/{$id}";
        return $this->apiCall('DELETE', $url);
    }

    protected function apiCall($type, $url, $params = [])
    {
        $url = "{$this->url}$url";

        if ($this->curl == null) {
            $this->curl = curl_init();
        }

        switch ($type = strtoupper($type)) {
            case 'DELETE':
                curl_setopt($this->curl, CURLOPT_URL, $url);
                curl_setopt($this->curl, CURLOPT_CUSTOMREQUEST, $type);
                curl_setopt($this->curl, CURLOPT_POSTFIELDS, json_encode($params));
                break;
            case 'PUT':
                curl_setopt($this->curl, CURLOPT_URL, $url);
                curl_setopt($this->curl, CURLOPT_CUSTOMREQUEST, $type);
                curl_setopt($this->curl, CURLOPT_POSTFIELDS, json_encode($params));
                break;
            case 'POST':
                curl_setopt($this->curl, CURLOPT_URL, $url);
                curl_setopt($this->curl, CURLOPT_CUSTOMREQUEST, $type);
                curl_setopt($this->curl, CURLOPT_POST, true);
                curl_setopt($this->curl, CURLOPT_POSTFIELDS, $params);
                break;
            case 'GET':
                curl_setopt($this->curl, CURLOPT_POSTFIELDS, null);
                curl_setopt($this->curl, CURLOPT_CUSTOMREQUEST, $type);
                curl_setopt($this->curl, CURLOPT_HTTPGET, true);
                $url .= '?' . http_build_query($params);
                curl_setopt($this->curl, CURLOPT_URL, $url);
                break;
        }

        curl_setopt($this->curl, CURLOPT_SSL_VERIFYPEER, false);
        curl_setopt($this->curl, CURLOPT_RETURNTRANSFER, true);

        $headers = [];
        foreach ($this->headers as $k => $v) {
            if ($k == 'Connection') {
                $v = 'Close';
            }
            $headers[] = "$k: $v";
        }

        curl_setopt($this->curl, CURLOPT_HTTPHEADER, $headers);
        curl_setopt($this->curl, CURLOPT_CONNECTTIMEOUT, 10);

        return $this->callWithRetries();
    }

    protected function callWithRetries()
    {
        for ($retry = 0; $retry < $this->options['max_retries']; $retry++) {
            $out = curl_exec($this->curl);
            if ($out === false) {
                $this->reportHttpError(0, curl_error($this->curl));
            }
            $this->last_status = curl_getinfo($this->curl, CURLINFO_HTTP_CODE);

            if ($this->last_status >= 200 && $this->last_status < 300) {
                return self::jsonDecode($out);
            } elseif ($this->last_status >= 500) {
                self::waitRandomInterval($retry);
            } else {
                $this->reportHttpError($this->last_status, $out);
            }
        }
        $this->reportHttpError($this->last_status, "Service unavailable");
        return;
    }

    protected static function jsonDecode($response)
    {
        $data = json_decode($response);

        $json_error = json_last_error();
        if ($json_error != JSON_ERROR_NONE) {
            throw new \RuntimeException($json_error);
        }

        return $data;
    }

    protected static function waitRandomInterval($retry)
    {
        $max_delay = pow(4, $retry) * 100 * 1000;
        usleep(rand(0, $max_delay));
    }

    protected function reportHttpError($status, $text)
    {
        throw new HttpException($status, "http error: {$status} | {$text}");
    }

    /**
     * Marshal out the pushed job and payload.
     *
     * @return object
     */
    protected function marshalPushedJob()
    {
        return (object) [
            'id'       => $this->request->header('topthink-message-id'),
            'payload'  => $this->request->getContent(),
            'attempts' => $this->request->header('topthink-message-attempts')
        ];
    }

    public function __destruct()
    {
        if ($this->curl != null) {
            curl_close($this->curl);
            $this->curl = null;
        }
    }

    public function pop($queue = null)
    {
        throw new \RuntimeException('pop queues not support for this type');
    }
}