<?php

namespace onespace\tools\rabbit\components;

use Closure;
use Enqueue\AmqpLib\AmqpConnectionFactory;
use Enqueue\AmqpLib\AmqpConsumer;
use Enqueue\AmqpLib\AmqpContext;
use Interop\Amqp\AmqpDestination;
use Interop\Amqp\Impl\AmqpBind;
use Interop\Amqp\Impl\AmqpMessage;
use Interop\Amqp\Impl\AmqpQueue;
use Interop\Amqp\Impl\AmqpTopic;
use Interop\Queue\Destination;
use Interop\Queue\SubscriptionConsumer;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Exception\AMQPProtocolChannelException;
use PhpAmqpLib\Exception\AMQPProtocolException;
use PhpAmqpLib\Wire\AMQPTable;
use Yii;
use yii\base\Component;
use yii\helpers\ArrayHelper;
use yii\httpclient\Client;

class Rabbit extends Component {
    // duplicate underlying flags to make naming and referencing more consistent
    public const EXCHANGE_TYPE_DIRECT = AmqpTopic::TYPE_DIRECT;
    public const EXCHANGE_TYPE_FANOUT = AmqpTopic::TYPE_FANOUT;
    public const EXCHANGE_TYPE_TOPIC = AmqpTopic::TYPE_TOPIC;
    public const EXCHANGE_TYPE_HEADERS = AmqpTopic::TYPE_HEADERS;
    public const EXCHANGE_TYPE_CONSISTENT_HASH = 'x-consistent-hash';

    public const EXCHANGE_FLAG_INTERNAL = AmqpTopic::FLAG_INTERNAL;
    public const EXCHANGE_FLAG_NOPARAM = AmqpTopic::FLAG_NOPARAM;
    public const EXCHANGE_FLAG_PASSIVE = AmqpTopic::FLAG_PASSIVE;
    public const EXCHANGE_FLAG_DURABLE = AmqpTopic::FLAG_DURABLE;
    public const EXCHANGE_FLAG_AUTODELETE = AmqpTopic::FLAG_AUTODELETE;
    public const EXCHANGE_FLAG_NOWAIT = AmqpTopic::FLAG_NOWAIT;
    public const EXCHANGE_FLAG_IFUNUSED = AmqpTopic::FLAG_IFUNUSED;

    public const QUEUE_FLAG_NOPARAM = AmqpQueue::FLAG_NOPARAM;
    public const QUEUE_FLAG_PASSIVE = AmqpQueue::FLAG_PASSIVE;
    public const QUEUE_FLAG_DURABLE = AmqpQueue::FLAG_DURABLE;
    public const QUEUE_FLAG_AUTODELETE = AmqpQueue::FLAG_AUTODELETE;
    public const QUEUE_FLAG_NOWAIT = AmqpQueue::FLAG_NOWAIT;
    public const QUEUE_FLAG_IFUNUSED = AmqpQueue::FLAG_IFUNUSED;
    public const QUEUE_FLAG_EXCLUSIVE = AmqpQueue::FLAG_EXCLUSIVE;
    public const QUEUE_FLAG_IFEMPTY = AmqpQueue::FLAG_IFEMPTY;

    public const BIND_FLAG_NOPARAM = AmqpBind::FLAG_NOPARAM;
    public const BIND_FLAG_NOWAIT = AmqpBind::FLAG_NOWAIT;

    public $serverUrl = '';
    public $serverPort = 5672;
    public $vHost = '/';
    public $username = 'testApp';
    public $password = 'testPassword';
    public $heartbeat = 10;

    public $outboxModel = null;

    /** @var AmqpContext $connection */
    private $connection = null;

    private $factory;

    public $uiEventConfig = [
        'exchange' => "onespace.message.ui",
    ];

    protected readonly string $apiUrl;

    public function init() {
        parent::init();
        $this->apiUrl = "http://{$this->serverUrl}:1{$this->serverPort}/api";
    }

    /**
     * Connecting is a relatively costly operation. It should only be done once, but only when the system actually needs to send something
     *
     * Best way to do this is to call this method
     */
    public function connect() {
        if ($this->connection == null) {
            $this->factory = new AmqpConnectionFactory([
                'host' => $this->serverUrl,
                'port' => $this->serverPort,
                'vhost' => $this->vHost,
                'user' => $this->username,
                'pass' => $this->password,
                'heartbeat' => $this->heartbeat,
            ]);

            $this->connection = $this->factory->createContext();
        }
        return $this;
    }

    /**
     * Forcibly reconnect to the server
     * @return $this
     */
    public function reconnect() {
        if ($this->connection != null) {
            $this->connection->close();
            $this->connection = null;
        }
        return $this->connect();
    }

    /**
     * Load an existing queue from the server
     *
     * @param string $name unique name of the queue
     *
     * @return AmqpTopic|null loaded queue or null if creation failed
     */
    public function loadExistingExchange(string $name) {
        return $this->declareExchange($name, AmqpTopic::TYPE_DIRECT, [AmqpTopic::FLAG_PASSIVE]);
    }

    /**
     * Create an exchange if it doesn't exist
     *
     * @param string $name name of the exchange
     * @param string $type type of exchange
     * @param int[] $flags array of integer flags
     * @param array $arguments array of arguments in form key => value
     *
     * @return AmqpTopic|null created exchange or null if creation failed
     */
    public function declareExchange(string $name, string $type, $flags = [], $arguments = []): ?AmqpTopic {
        $this->connect();
        $exchange = $this->connection->createTopic($name);
        $exchange->setType($type);

        // collapse the flag array into an integer
        if ($flags == null) {
            $flagInt = AmqpTopic::FLAG_NOPARAM;
        } else {
            $flagInt = array_sum(array_unique($flags, SORT_NUMERIC));
        }
        $exchange->setFlags($flagInt);
        $exchange->setArguments($arguments);

        try {
            $this->connection->declareTopic($exchange);
            return $exchange;
        } catch (AMQPProtocolChannelException $ex) {
            // when this error occurs, the system tears the channel down - next item will need to recreate it
            $this->connection->close();
            $this->connection = null;
            return null;
        } catch (AMQPProtocolException $ex) {
            // if any kind of protocol error occurs, assume the channel needs to be torn down
            $this->connection->close();
            $this->connection = null;
            return null;
        }
    }

    /**
     * Load an existing queue from the server
     *
     * @param string $name unique name of the queue
     *
     * @return AmqpQueue|null loaded queue or null if creation failed
     */
    public function loadExistingQueue(string $name) {
        return $this->declareQueue($name, [AmqpQueue::FLAG_PASSIVE]);
    }

    /**
     * Create a queue if it doesn't exist
     *
     * @param string $name unique name of the queue
     * @param int[] $flags array of integer flags
     * @param array $arguments array of arguments in form key => value
     *
     * @return AmqpQueue|null created queue or null if creation failed
     */
    public function declareQueue(string $name, $flags = [], $arguments = []) {
        $this->connect();
        $queue = $this->connection->createQueue($name);

        // collapse the flag array into an integer
        if ($flags == null) {
            $flagInt = AmqpQueue::FLAG_NOPARAM;
        } else {
            $flagInt = array_sum(array_unique($flags, SORT_NUMERIC));
        }
        $queue->setFlags($flagInt);
        $queue->setArguments($arguments);

        try {
            $this->connection->declareQueue($queue);
            return $queue;
        } catch (AMQPProtocolChannelException $ex) {
            // when this error occurs, the system tears the channel down - next item will need to recreate it
            $this->connection->close();
            $this->connection = null;
            return null;
        } catch (AMQPProtocolException $ex) {
            // if any kind of protocol error occurs, assume the channel needs to be torn down
            $this->connection->close();
            $this->connection = null;
        }
    }

    /**
     * Get the status of a queue on the server
     *
     * @param string|AmqpQueue $queue instance or name of the queue
     *
     * @return int|false if queue does not exist on server false; otherwise, number of pending messages in the queue
     */
    public function getQueueStatus($queue) {
        $this->connect();
        // if we were given the string
        if (is_string($queue)) {
            $queue = $this->connection->createQueue($queue);
        }

        // by setting the passive flag, we're checking the server, not actually creating a queue
        $queue->setFlags(AmqpQueue::FLAG_PASSIVE);

        try {
            return $this->connection->declareQueue($queue);
        } catch (AMQPProtocolChannelException $ex) {
            // when this error occurs, the system tears the channel down - next item will need to recreate it
            $this->connection->close();
            $this->connection = null;
            return false;
        } catch (AMQPProtocolException $ex) {
            // if any kind of protocol error occurs, assume the channel needs to be torn down
            $this->connection->close();
            $this->connection = null;
        }
    }

    /**
     * Delete an existing queue
     *
     * @param string|AmqpQueue $queue instance or name of the queue
     */
    public function deleteQueue($queue) {
        $this->connect();
        // if we were given the string
        if (is_string($queue)) {
            $queue = $this->connection->createQueue($queue);
        }

        $this->connection->deleteQueue($queue);
    }

    /**
     * Bind a queue or exchange to an exchange
     *
     * @param AmqpDestination $item queue/exchange that should RECEIVE messages
     * @param AmqpDestination $destination exchange that will be SENDING messages
     */
    public function bind(AmqpDestination $item, AmqpDestination $destination, $routingKey = null, $flags = [], $arguments = []) {
        $this->connect();
        // collapse the flag array into an integer
        if ($flags == null) {
            $flagInt = AmqpBind::FLAG_NOPARAM;
        } else {
            $flagInt = array_sum(array_unique($flags, SORT_NUMERIC));
        }
        $bind = new AmqpBind($destination, $item, $routingKey, $flagInt, $arguments);
        $this->connection->bind($bind);
    }

    /**
     * Unbind a queue or exchange from an exchange
     */
    public function unbind(AmqpDestination $item, AmqpDestination $destination, $routingKey = null, $flags = [], $arguments = []) {
        $this->connect();
        // collapse the flag array into an integer
        if ($flags == null) {
            $flagInt = AmqpBind::FLAG_NOPARAM;
        } else {
            $flagInt = array_sum(array_unique($flags, SORT_NUMERIC));
        }
        $bind = new AmqpBind($destination, $item, $routingKey, $flagInt);
        $this->connection->unbind($bind);
    }

    /**
     * Subscribe to events on given queue(s), and run the listener on each event
     *
     * @param Destination|Destination[] $sources list of queues to listen to
     * @param Closure $listener code to run when an event is emitted
     * @param int[] $flags array of integer flags
     * @param int $prefetch The number of messages the consumer should fetch at a time
     *
     * @return SubscriptionConsumer populated subscription consumer that can be listened to
     */
    public function subscribeToMessages($sources, Closure $listener, $flags = [], int $prefetch = 1) {
        $this->connect();
        // make sure sources is an array
        if (!is_array($sources)) {
            $sources = [$sources];
        }
        // collapse the flag array into an integer
        if ($flags == null) {
            $flagInt = AmqpConsumer::FLAG_NOPARAM;
        } else {
            $flagInt = array_sum(array_unique($flags, SORT_NUMERIC));
        }

        // Set the prefetch count. If 1, leave settings as default, which is 1 prefetch
        if ($prefetch > 1) {
            $channel = $this->connection->getLibChannel();
            $channel->basic_qos(null, $prefetch, null);
        }

        $subConsumer = $this->connection->createSubscriptionConsumer();

        foreach ($sources as $source) {
            $consumer = $this->connection->createConsumer($source);
            $consumer->setFlags($flagInt);
            $subConsumer->subscribe($consumer, $listener);
        }

        return $subConsumer;
    }

    /**
     * Publish an event to a given exchange or queue
     */
    public function sendMessage(
        Destination $destination,
        string $message,
        $routingKey = null,
        $customHeaders = [],
        $timestamp = null,
        $contentType = 'application/json',
        $persistent = true,
        $priority = null,
        $ttl = null
    ) {
        $this->connect();
        if ($timestamp === null) {
            $timestamp = time();
        }
        $producer = $this->connection->createProducer();
        $producer->setPriority($priority);
        $producer->setTimeToLive($ttl);

        $message = $this->connection->createMessage($message);
        $message->setHeader('application_headers', new AMQPTable($customHeaders));

        if ($persistent) {
            $message->setDeliveryMode(AmqpMessage::DELIVERY_MODE_PERSISTENT);
        } else {
            $message->setDeliveryMode(AmqpMessage::DELIVERY_MODE_NON_PERSISTENT);
        }
        $message->setRoutingKey($routingKey);
        $message->setContentType($contentType);
        if ($timestamp !== false) {
            $message->setTimestamp($timestamp);
        }
        $producer->send($destination, $message);
    }

    /**
     * Publish an event to a given exchange, using an outbox pattern
     *
     * Note, if [[outboxModel]] is not set, the message will be sent immediately using sendMessage()
     */
    public function outboxSendMessage(
        string $destinationName,
        string $message,
        $routingKey = null,
        $customHeaders = [],
        $timestamp = null,
        $contentType = 'application/json',
        $persistent = true,
        $priority = null,
        $ttl = null
    ) {
        if ($timestamp === null) {
            $timestamp = time();
        }
        if ($this->outboxModel != null) {
            $outbox = new $this->outboxModel();
            $outbox->exchange = $destinationName;
            $outbox->message = $message;
            $outbox->routingKey = $routingKey;
            $outbox->headers = $customHeaders;
            $outbox->timestamp = $timestamp;
            $outbox->contentType = $contentType;
            $outbox->persistent = $persistent ? 1 : 0;
            $outbox->priority = $priority;
            $outbox->ttl = $ttl;
            $outbox->save();
        } else {
            // send the message immediately, since there is no outbox defined
            $this->sendMessage(
                $this->loadExistingExchange($destinationName),
                $message,
                $routingKey,
                $customHeaders,
                $timestamp,
                $contentType,
                $persistent,
                $priority,
                $ttl
            );
        }
    }

    public function createUser($username, $password, $tags = 'none') {
        $client = new Client();
        $response = $client->put("{$this->apiUrl}/users/{$username}", ['password' => $password, 'tags' => $tags])
            ->setFormat(Client::FORMAT_JSON)
            ->addHeaders(['Authorization' => 'Basic ' . base64_encode($this->username . ':' . $this->password)])
            ->send();
        if ($response->isOk) {
            Yii::info($response->data, __METHOD__);
        } else {
            Yii::error($response->content, __METHOD__);
        }
        return $response->isOk;
    }

    public function deleteUser(string $username): bool {
        $client = new Client();
        $response = $client->delete("{$this->apiUrl}/users/{$username}")
            ->setFormat(Client::FORMAT_JSON)
            ->addHeaders(['Authorization' => 'Basic ' . base64_encode($this->username . ':' . $this->password)])
            ->send();
        if ($response->isOk) {
            Yii::info($response->data, __METHOD__);
        } else {
            Yii::error($response->content, __METHOD__);
        }
        return $response->isOk;
    }

    public function getUser($username) {
        $client = new Client();
        $response = $client->get("{$this->apiUrl}/users/{$username}")
            ->setFormat(Client::FORMAT_JSON)
            ->addHeaders(['Authorization' => 'Basic ' . base64_encode($this->username . ':' . $this->password)])
            ->send();
        if (!$response->isOk) {
            Yii::error($response->content, __METHOD__);
            return null;
        } else {
            Yii::info($response->data, __METHOD__);
            return $response->data;
        }
    }

    public function setUserPermissions(string $username, string $configure = ".*", string $write = ".*", string $read = ".*"): bool {
        $vhost = urlencode($this->vHost);
        $client = new Client();
        $response = $client->put("{$this->apiUrl}/permissions/{$vhost}/{$username}", ['configure' => $configure, 'write' => $write, 'read' => $read])
            ->setFormat(Client::FORMAT_JSON)
            ->addHeaders(['Authorization' => 'Basic ' . base64_encode($this->username . ':' . $this->password)])
            ->send();
        if ($response->isOk) {
            Yii::info($response->data, __METHOD__);
        } else {
            Yii::error($response->content, __METHOD__);
        }
        return $response->isOk;
    }

    public function setUserTopicPermissions(string $username, string $exchange = 'amq.topic', string $write = '.*', string $read = '.*'): bool {
        $vhost = urlencode($this->vHost);
        $client = new Client();
        $response = $client->put("{$this->apiUrl}/topic-permissions/{$vhost}/{$username}", ['exchange' => $exchange, 'write' => $write, 'read' => $read])
            ->setFormat(Client::FORMAT_JSON)
            ->addHeaders(['Authorization' => 'Basic ' . base64_encode($this->username . ':' . $this->password)])
            ->send();
        if ($response->isOk) {
            Yii::info($response->data, __METHOD__);
        } else {
            Yii::error($response->content, __METHOD__);
        }
        return $response->isOk;
    }

    /**
     * This purges the parsed queue on the rabbit server.
     *
     * ⚠️⚠️⚠️⚠️ WARNING THIS WILL RESULT IN DATA LOSS.
     *
     * Do not use without significant consideration.
     */
    public function purgeQueue(string $queueName): bool {
        $vhost = urlencode($this->vHost);
        $client = new Client();

        $response = $client->delete("{$this->apiUrl}/queues/{$vhost}/{$queueName}/contents")
            ->setFormat(Client::FORMAT_JSON)
            ->addHeaders(['Authorization' => 'Basic ' . base64_encode($this->username . ':' . $this->password)])
            ->send();

        if ($response->isOk) {
            Yii::info($response->data, __METHOD__);
        } else {
            Yii::error($response->content, __METHOD__);
        }
        return $response->isOk;
    }

    /** Fire and forget a message to the UI exchange - meant for SSE/Websocket events */
    public function sendUiMessage(string $type, string $routingKey, mixed $messageData){
        $exchange = ArrayHelper::getValue($this->uiEventConfig, 'exchange');
        $timestamp = time();

        $message = [
            'event' => [
                'type' => $type,
                'timestamp' => $timestamp
            ],
            'data' => ArrayHelper::toArray($messageData)
        ];

        $this->sendMessage(
            $this->declareExchange($exchange, Rabbit::EXCHANGE_TYPE_TOPIC, [Rabbit::EXCHANGE_FLAG_DURABLE, Rabbit::EXCHANGE_FLAG_PASSIVE]),
            json_encode($message),
            $routingKey,
            [],
            $timestamp
        );
    }

    /**
     * Get any information about a rabbit queue
     *
     * @param string $name The name of the queue to query
     */
    public function getQueueInfo(string $name): array {
        $vhost = urlencode($this->vHost);
        $client = new Client();

        $url = "{$this->apiUrl}/queues/{$vhost}/{$name}";
        $response = $client->get($url)
            ->setFormat(Client::FORMAT_JSON)
            ->addHeaders(['Authorization' => 'Basic ' . base64_encode($this->username . ':' . $this->password)])
            ->send();

        if ($response->isOk) {
            Yii::info($response->data, __METHOD__);
            return $response->data;
        } else {
            Yii::error($response->content, __METHOD__);
            return [];
        }
    }

    /**
     * Gets the underlying php-amqplib channel for advanced operations
     * like heartbeat-aware waiting during long operations.
     */
    public function getLibChannel(): AMQPChannel {
        $this->connect();
        return $this->connection->getLibChannel();
    }
}
