<?php

namespace onespace\tools\rabbit\commands;

use Enqueue\AmqpLib\AmqpConsumer;
use Exception;
use Interop\Amqp\Impl\AmqpMessage;
use onespace\tools\rabbit\components\Rabbit;
use onespace\tools\rabbit\helpers\CLIHelper;
use onespace\tools\rabbit\helpers\Log;
use onespace\tools\rabbit\models\RabbitOutbox;
use onespace\tools\rabbit\models\RabbitTransactionErrorLog;
use PhpAmqpLib\Exception\AMQPChannelClosedException;
use PhpAmqpLib\Exception\AMQPHeartbeatMissedException;
use Yii;
use yii\base\Module;

abstract class BaseRabbitController extends \yii\console\Controller {
    use CLIHelper;

    /** Overwrite in the child class to change this value */
    public const DEFAULT_QUEUE_COUNT = 1;
    /** Overwrite in the child class to change this value */
    public const AVERAGE_SECS_PER_MESSAGE = 0.1;

    /**
     * The path of your rabbit service classes. Overwrite in the parent class if a different path is
     * required - for example in yii2-advanced split.
     *
     * @var string  $servicePath    Default value: `\app\services\rabbit\`
     *
     * @access  protected
     */
    protected string $servicePath = '\app\services\rabbit\\';

    /**
     * Sets the prefix of the specific queue series of queues & exchanges
     * used by this class.
     */
    abstract protected function appPrefix(): string;

    /**
     * To be run on launch - ensure required exchanges and queues are set up
     *
     * @access  public
     */
    abstract public function actionPrepareQueues(): void;

    /**
     * Returns a standardized general prefix used in the system.
     *
     * @return  string
     *
     * @static
     * @access  private
     */
    private static function getPrefix(): string {
        $prefix = '';
        if (YII_DEBUG) {
            // $prefix = 'dev_';
        }
        return $prefix;
    }

    /**
     * Get a definitive standard queue name from a parsed queue name.
     *
     * @param   string  $name   The raw queue name to be used.
     *
     * @return  string
     *
     * @static
     * @access  protected
     * @deprecated
     */
    protected static function getQueueName(string $name): string {
        $prefix = self::getPrefix();
        $name = str_replace('-', '_', $name);
        return $prefix . ((new static('id', new Module('id')))->appPrefex()) . $name;
    }

    /**
     * Get a definitive standard queue name from a parsed queue name.
     *
     * Replaces the static method `getQueueName`.
     */
    protected function queueName(string $name): string {
        $prefix = self::getPrefix();
        $name = str_replace('-', '_', $name);
        return $prefix . $this->appPrefix() . $name;
    }

    /**
     * Get a definitive standard exchange name from a parsed echange name.
     *
     * @param   string  $name   The raw echanged name to be used.
     *
     * @return  string
     *
     * @static
     * @access  protected
     * @deprecated
     */
    protected static function getExchangeName(string $name): string {
        $prefix = self::getPrefix();
        $name = str_replace('-', '_', $name);
        return $prefix . (new static('id', new Module('id')))->appPrefex() . $name;
    }

    /**
     * Get a definitive standard exchange name from a parsed echange name.
     *
     * Replaces the static method `getExchangeName`.
     */
    protected function exchangeName(string $name): string {
        $prefix = self::getPrefix();
        $name = str_replace('-', '_', $name);
        return $prefix . $this->appPrefix() . $name;
    }

    /**
     * Returns the queue count to spin up on a typical service. Either from a parsed queue name in
     * the .env or the default defined in the parent class.
     *
     * @param   string  $name   The identifier used to get the .env variable. Usually the entity name.
     *
     * @return  int
     *
     * @static
     * @access  protected
     * @deprecated
     */
    protected static function getQueueCount(string $name): int {
        $name = strtoupper($name);
        $name = str_replace('-', '_', $name);
        $count = getenv($name . '_QUEUE_COUNT');
        if ($count !== false) {
            return intval($count);
        }
        return static::DEFAULT_QUEUE_COUNT;
    }

    /**
     * Returns the queue count to spin up on a typical service. Either from a parsed queue name in
     * the .env or the default defined in the parent class.
     *
     * Replaces the static method `getQueueCount`.
     */
    protected function queueCount(string $name): int {
        $name = strtoupper($name);
        $name = str_replace('-', '_', $name);
        $count = getenv($name . '_QUEUE_COUNT');
        if ($count !== false) {
            return intval($count);
        }
        return static::DEFAULT_QUEUE_COUNT;
    }

    /**
     * Helper method intended to be used in the method `actionPrepareQueues`.
     *
     * Replaces the static method `handlePrepareQueues`.
     *
     * * Replaces the static method `getQueueCount`.
     *
     * ## Example Usage
     *
     * ```php
     * $entities = [
     *     'amq.topic' => ['provision_queue', 'gw_queue'],
     *     'onedevice.gateway.down' => 'gateway_down',
     * ];
     * $customBinds = [
     *     'provision_queue' => 'v1.pb.prov.*.up',
     *     'gw_queue' => 'v1.pb.gw.*.up',
     *     'gateway_down' => 1,
     * ];
     * $sac = [
     *     'gateway_down' => false,
     * ]
     * $this->prepareQueues($entities, $customBinds);
     * ```
     */
    protected function prepareQueues(array $entities, array $customBinds = [], array $sac = []): void {
        /** @var Rabbit $rabbit */
        $rabbit = Yii::$app->rabbit;

        foreach ($entities as $exchange => $entityGroup) {
            if (is_string($entityGroup)) {
                $entityGroup = [$entityGroup];
            }

            $prefix = self::getPrefix();
            // passively declare the entity exchange
            // if it doesn't exist yet, we should ignore it
            // this does mean that if a new entity is added, the service will need to be restarted
            $entityExchange = $rabbit->loadExistingExchange($prefix . str_replace('-', '_', $exchange));

            foreach ($entityGroup as $entity) {
                if ($entityExchange !== null) {
                    // create a single consistent hash exchange, named for this app and entity
                    $this->writeln("Declaring app exchange...");
                    $appExchange = $rabbit->declareExchange(
                        $this->exchangeName($entity),
                        Rabbit::EXCHANGE_TYPE_CONSISTENT_HASH,
                        [Rabbit::EXCHANGE_FLAG_DURABLE, Rabbit::EXCHANGE_FLAG_INTERNAL],
                        ['hash-header' => 'hash-on']
                    );

                    // bind the exchange to the entities we're interested in
                    $this->writeln("Binding {$exchange} app exchange to entity exchange...");

                    // bind our app exchange to the entity exchange, but only to receive entity (CRUD) events
                    if (isset($customBinds[$entity])) {
                        // If a custom bind is defined
                        $bind = $customBinds[$entity];
                        if (!is_array($bind)) {
                            $bind = [$bind];
                        }
                        foreach ($bind as $key) {
                            $rabbit->bind($entityExchange, $appExchange, $key);
                        }
                    } else {
                        // Custom not defined or specific entry not set.
                        $rabbit->bind($entityExchange, $appExchange, 'entity.#');
                        $rabbit->bind($entityExchange, $appExchange, $entity . '.#');
                    }
                    $this->writeln("\tBinding to {$prefix}{$exchange} exchange succeeded!");
                } else {
                    $this->writeln("\tBinding to {$prefix}{$exchange} exchange failed!");
                    continue;
                }

                // create the required number of queues and bind them to the app exchange
                $baseQueueName = $this->queueName($entity) . '_';
                $this->writeln("Loading {$this->queueCount($entity)} queues...");
                for ($x = 1; $x <= $this->queueCount($entity); $x++) {
                    $this->write("\t");
                    // create a 4 digit number out of the index
                    $stringNum = str_pad(strval($x), 4, '0', STR_PAD_LEFT);

                    // try to load the queue if it already exists
                    $queue = $rabbit->loadExistingQueue($baseQueueName . $stringNum);
                    if ($queue === null) {
                        // create the queue if it doesn't exist
                        $queue = $rabbit->declareQueue(
                            $baseQueueName . $stringNum,
                            [Rabbit::QUEUE_FLAG_DURABLE],
                            ['x-single-active-consumer' => $sac[$entity] ?? true],
                        );
                        $this->write("Creating " . (($sac[$entity] ?? true) ? '(SAC)' : '(Not SAC)') . "...");
                    }

                    // bind the queue to the app exchange
                    $rabbit->bind($appExchange, $queue, 1);
                    $this->writeln("Loaded and bound queue {$baseQueueName}{$stringNum}!");
                }

                // Reduce number of queues if needed
                $hasExcess = true;
                $this->writeln("Checking for excess queues...");
                while ($hasExcess) {
                    $stop = false;
                    $excessQueueNames = [];
                    $populatedQueues = [];
                    $waitingMessages = 0;

                    $index = $this->queueCount($entity);
                    // attempt to connect to queues with a higher number than the current number, until we find one that doesn't exist
                    while (!$stop) {
                        $index++;
                        // create a 4 digit number out of the index
                        $stringNum = str_pad(strval($index), 4, '0', STR_PAD_LEFT);

                        // try to load the queue if it exists
                        $this->write("\tSearching for queue {$stringNum}... ");
                        $queue = $rabbit->loadExistingQueue($baseQueueName . $stringNum);

                        if ($queue !== null) {
                            $this->write("Found queue {$stringNum}... ");
                            // unbind each one from the app exchange, ensuring they won't receive more messages
                            $rabbit->unbind($queue, $appExchange);
                            $this->writeln("Unbound from app exchange...");
                            $excessQueueNames[] = $queue->getQueueName();

                            // may as well check counts at this point, save doing it again later
                            // since we've already unbound the queue, no data should enter after this point
                            if (($messageCount = $rabbit->getQueueStatus($queue)) > 0) {
                                $populatedQueues[] = $queue;
                                $waitingMessages += $messageCount;
                            }
                        } else {
                            // we found a queue that doesn't exist, time to stop
                            $stop = true;
                            $this->writeln("Found the end of the queues at {$stringNum}...");
                        }
                    }

                    // if there are no messages waiting any more, go onto the deletion step
                    if ($waitingMessages == 0) {
                        $hasExcess = false;
                    } else {
                        $this->writeln("\tProcessing {$waitingMessages} outstanding messages...");
                        // process the data we can out of the populated queues
                        $consumer = $rabbit->subscribeToMessages($populatedQueues, function (AmqpMessage $message, AmqpConsumer $consumer) {
                            /**
                             * This will probably not work, because the full context is not available (entity name) which
                             * would then call a specific class.
                             */
                            return false; // Cannot succeed
                            // return $this->processMessage($message, $consumer);
                        });
                        // run a consume for approximately enough time to process all the jobs pending, then run again
                        $consumer->consume(ceil($waitingMessages * static::AVERAGE_SECS_PER_MESSAGE));
                        $this->writeln("\tDone!\n");
                        $this->writeln("\tChecking if queues are empty again...");
                    }
                }
                // work backwards, deleting all excess queues
                // we go backwards in case there is an issue - it should prevent any kind of hole of numbers developing
                $this->writeln("Deleting " . count($excessQueueNames) . " excess queues...");
                for ($x = count($excessQueueNames); $x > 0; $x--) {
                    // try to load the queue if it exists
                    $queue = $rabbit->loadExistingQueue($excessQueueNames[$x - 1]);
                    if ($queue !== null) {
                        $rabbit->deleteQueue($queue);
                    }
                }
            }
        }
        $this->writeln("!!!DONE!!!");
    }

    /**
     * Helper method for use in the method `actionPrepareQueues`.
     *
     * ## Example Usage
     *
     * ```php
     * $entities = [
     *     'amq.topic' => ['provision_queue', 'gw_queue'],
     *     'onedevice.gateway.down' => 'gateway_down',
     * ];
     * $customBinds = [
     *     'provision_queue' => 'v1.pb.prov.*.up',
     *     'gw_queue' => 'v1.pb.gw.*.up',
     *     'gateway_down' => 1,
     * ];
     * self::handlePrepareQueues($entities, $customBinds);
     * ```
     *
     * @param   array   $entities
     * @param   array   $customBinds
     *
     * @static
     * @access  protected
     * @deprecated
     */
    protected static function handlePrepareQueues(array $entities, array $customBinds = []): void {
        /** @var Rabbit $rabbit */
        $rabbit = Yii::$app->rabbit;

        foreach ($entities as $exchange => $entityGroup) {
            if (is_string($entityGroup)) {
                $entityGroup = [$entityGroup];
            }

            $prefix = self::getPrefix();
            // passively declare the entity exchange
            // if it doesn't exist yet, we should ignore it
            // this does mean that if a new entity is added, the service will need to be restarted
            $entityExchange = $rabbit->loadExistingExchange($prefix . str_replace('-', '_', $exchange));

            foreach ($entityGroup as $entity) {
                if ($entityExchange !== null) {
                    // create a single consistent hash exchange, named for this app and entity
                    echo "Declaring app exchange...\n";
                    $appExchange = $rabbit->declareExchange(
                        self::getExchangeName($exchange),
                        Rabbit::EXCHANGE_TYPE_CONSISTENT_HASH,
                        [Rabbit::EXCHANGE_FLAG_DURABLE, Rabbit::EXCHANGE_FLAG_INTERNAL],
                        ['hash-header' => 'hash-on']
                    );

                    // bind the exchange to the entities we're interested in
                    echo "Binding $exchange app exchange to entity exchanges...\n";

                    // bind our app exchange to the entity exchange, but only to receive entity (CRUD) events
                    if (isset($customBinds[$entity])) {
                        // If a custom bind is defined
                        $bind = $customBinds[$entity];
                        if (!is_array($bind)) {
                            $bind = [$bind];
                        }
                        foreach ($bind as $key) {
                            $rabbit->bind($entityExchange, $appExchange, $key);
                        }
                    } else {
                        // Custom not defined or specific entry not set.
                        $rabbit->bind($entityExchange, $appExchange, 'entity.#');
                        $rabbit->bind($entityExchange, $appExchange, $entity . '.#');
                    }
                    echo "\tBinding to {$prefix}{$exchange} exchange succeeded!\n";
                } else {
                    echo "\tBinding to {$prefix}{$exchange} exchange failed!\n";
                    continue;
                }

                // create the required number of queues and bind them to the app exchange
                $baseQueueName = self::getQueueName($entity) . '_';
                echo "Loading " . self::getQueueCount($entity) . " queues...\n";
                for ($x = 1; $x <= self::getQueueCount($entity); $x++) {
                    echo "\t";
                    // create a 4 digit number out of the index
                    $stringNum = str_pad(strval($x), 4, '0', STR_PAD_LEFT);

                    // try to load the queue if it already exists
                    $queue = $rabbit->loadExistingQueue($baseQueueName . $stringNum);
                    if ($queue === null) {
                        // create the queue if it doesn't exist
                        $queue = $rabbit->declareQueue(
                            $baseQueueName . $stringNum,
                            [Rabbit::QUEUE_FLAG_DURABLE],
                            ['x-single-active-consumer' => (self::getQueueCount($entity) > 1)]
                        );
                        echo "Creating...";
                    }

                    // bind the queue to the app exchange
                    $rabbit->bind($appExchange, $queue, 1);
                    echo "Loaded and bound queue " . $baseQueueName . $stringNum . "!\n";
                }

                // Reduce number of queues if needed
                $hasExcess = true;
                echo "Checking for excess queues...\n";
                while ($hasExcess) {
                    $stop = false;
                    $excessQueueNames = [];
                    $populatedQueues = [];
                    $waitingMessages = 0;

                    $index = self::getQueueCount($entity);
                    // attempt to connect to queues with a higher number than the current number, until we find one that doesn't exist
                    while (!$stop) {
                        $index++;
                        // create a 4 digit number out of the index
                        $stringNum = str_pad(strval($index), 4, '0', STR_PAD_LEFT);

                        // try to load the queue if it exists
                        echo "\tSearching for queue $stringNum... ";
                        $queue = $rabbit->loadExistingQueue($baseQueueName . $stringNum);

                        if ($queue !== null) {
                            echo "Found queue $stringNum... ";
                            // unbind each one from the app exchange, ensuring they won't receive more messages
                            $rabbit->unbind($queue, $appExchange);
                            echo "Unbound from app exchange...\n";
                            $excessQueueNames[] = $queue->getQueueName();

                            // may as well check counts at this point, save doing it again later
                            // since we've already unbound the queue, no data should enter after this point
                            if (($messageCount = $rabbit->getQueueStatus($queue)) > 0) {
                                $populatedQueues[] = $queue;
                                $waitingMessages += $messageCount;
                            }
                        } else {
                            // we found a queue that doesn't exist, time to stop
                            $stop = true;
                            echo "Found the end of the queues at $stringNum...\n";
                        }
                    }

                    // if there are no messages waiting any more, go onto the deletion step
                    if ($waitingMessages == 0) {
                        $hasExcess = false;
                    } else {
                        echo "\tProcessing $waitingMessages outstanding messages...\n";
                        // process the data we can out of the populated queues
                        $consumer = $rabbit->subscribeToMessages($populatedQueues, function (AmqpMessage $message, AmqpConsumer $consumer) {
                            return $this->processMessage($message, $consumer);
                        });
                        // run a consume for approximately enough time to process all the jobs pending, then run again
                        $consumer->consume(ceil($waitingMessages * static::AVERAGE_SECS_PER_MESSAGE));
                        echo "\tDone!\n\n";
                        echo "\tChecking if queues are empty again...\n";
                    }
                }
                // work backwards, deleting all excess queues
                // we go backwards in case there is an issue - it should prevent any kind of hole of numbers developing
                echo "Deleting " . count($excessQueueNames) . " excess queues...\n";
                for ($x = count($excessQueueNames); $x > 0; $x--) {
                    // try to load the queue if it exists
                    $queue = $rabbit->loadExistingQueue($excessQueueNames[$x - 1]);
                    if ($queue !== null) {
                        $rabbit->deleteQueue($queue);
                    }
                }
            }
        }
        echo "!!!DONE!!!\n";
    }

    /**
     * Monitor the Rabbit outbox, if anything is there, send them.
     */
    public function actionMonitorOutbox(): never {
        $this->writeln("Monitoring outbox...");
        $this->writeln("Connecting to Rabbit server...");
        /** @var Rabbit $rabbit */
        $rabbit = Yii::$app->rabbit;
        $rabbit->connect();
        $this->writeln("Connected. Monitoring...");
        while (true) {
            // check if there is anything in the outbox table
            $messages = RabbitOutbox::find();
            if ($messages->count() > 0) {
                // if so, send them on Rabbit
                foreach ($messages->each() as $message) {
                    /** @var RabbitOutbox $message */
                    try {
                        $rabbit->sendMessage(
                            destination: $rabbit->declareExchange(
                                $message->exchange,
                                Rabbit::EXCHANGE_TYPE_TOPIC,
                                [Rabbit::EXCHANGE_FLAG_DURABLE, Rabbit::EXCHANGE_FLAG_PASSIVE]
                            ),
                            message: $message->message,
                            routingKey: $message->routingKey,
                            customHeaders: $message->headers,
                            timestamp: $message->timestamp,
                            contentType: $message->contentType,
                            persistent: ($message->persistent ? true : false),
                            priority: $message->priority,
                            ttl: $message->ttl
                        );

                        $message->delete();
                        $this->writeln("Sent message #{$message->id}!");
                    } catch (AMQPChannelClosedException | AMQPHeartbeatMissedException $ex) {
                        $this->writeln("Failed to send message #{$message->id}!");
                        $this->writeln("\t" . $ex::class . " - {$ex->getMessage()}");
                        // wait a second to prevent overloading the server
                        sleep(1);
                        $rabbit->reconnect();
                        break;
                    } catch (Exception $ex) {
                        $this->writeln("Failed to send message #{$message->id}!");
                        $this->writeln("\t{$ex->getMessage()}");
                        // wait a second to prevent overloading the server
                        sleep(1);
                        // break at this point - need to maintain FIFO
                        break;
                    }
                }
            } else {
                // if there is nothing, wait a second before trying again
                sleep(1);
            }
        }

        throw new Exception("This script is never intended to finish");
    }

    /**
     * Listen for entity events on the queue, and process them as they come in
     *
     * @param   string  $entityName The identifier of the EventService to to execute. The file should be `Process{$entityName}EventService.php`.
     * @param   int     $timeout    When the run should time out. Typically passed as 0 for unlimited time.
     * @param   int     $prefetch   Set the prefetch of each consumer. This is the number of messages pulled at a time per consumer.
     *
     * @access  public
     */
    public function actionListenEntityEvents(string $entityName, int $timeout, int $prefetch = 1): void {
        $this->writeln("Starting...");
        /** @var Rabbit $rabbit */
        $rabbit = Yii::$app->rabbit;

        $className = $this->servicePath . 'Process' . str_replace(' ', '', ucwords(str_replace(['-', '.'], ' ', $entityName))) . 'EventService';
        if (!class_exists($className)) {
            $this->writeln("Failed to find class {$className}");
            return;
        }

        $queues = [];
        $baseQueueName = $this->queueName($entityName) . '_';
        // prep the names of all the queues to connect to
        $this->writeln("Connecting to queues...");
        for ($x = 1; $x <= $this->queueCount($entityName); $x++) {
            // create a 4 digit number out of the index
            $stringNum = str_pad(strval($x), 4, '0', STR_PAD_LEFT);

            // try to load the queue if it already exists
            // if the queue doesn't exist for any reason
            if ($queue = $rabbit->loadExistingQueue($baseQueueName . $stringNum)) {
                $this->writeln("\tFound queue {$baseQueueName}{$stringNum}...");
                $queues[] = $queue;
            }
        }
        // create a consumer that will listen to all the queues
        $consumer = $rabbit->subscribeToMessages($queues, function (AmqpMessage $message, AmqpConsumer $consumer) use ($className) {
            /** @var \onespace\tools\rabbit\services\base\BaseRabbitService @service */
            $service = new $className($message);

            $result = $service->run();

            /**
             * This is here for the unusual case where a transaction is not correctly closed
             * off (either committed or rolled back). This checks for an active transaction,
             * and rolls back any changes if found.
             *
             * Finally it logs the whole thing for later analysis
             *
             * Signed: Justin, Jason, Chris & Gareth (2025-08-27)
             */
            if ($transaction = Yii::$app->db->getTransaction()) {
                $this->writeln("Invalid transaction found, closing and logging...", Log::WARNING);
                $transaction->rollBack();
                RabbitTransactionErrorLog::logError(
                    $className,
                    $message->getBody(),
                    $message->getHeaders(),
                    $message->getRoutingKey(),
                );
            }

            // depending on the result, either acknowledge or reject the message
            if ($result) {
                $this->writelnTs();
                $this->drawStarLine();
                $this->writeln("Acknowledged!");
                $consumer->acknowledge($message);
            } else {
                $this->writeData($service->getErrorMessages());
                $consumer->reject($message, true);
            }

            return true;
        }, prefetch: $prefetch);
        $this->writeln("Listening to queues...");
        // start listening
        while (true) {
            $consumer->consume($timeout);
        }
    }

    /**
     * Listen for user events on the queue, and process them as they come in
     *
     * @param   int     $timeout    When the run should time out. Typically passed as 0 for unlimited time.
     * @param   string  $queueNames A comma seperated string with a list of queues to monitor.
     * @param   string  $methodName The method to call when one of the queues that are monitored receives a message.
     *                              This method should be in the parent class.
     * @param   int     $prefetch   Set the prefetch of each consumer. This is the number of messages pulled at a time per consumer.
     *
     * @access  public
     */
    public function actionListenArbitraryEvents(int $timeout, string $queueNames, string $methodName, int $prefetch = 1): void {
        $this->writeln("Starting...");
        /** @var Rabbit $rabbit */
        $rabbit = Yii::$app->rabbit;

        // prep the names of all the queues to connect to
        $this->writeln("Connecting to queues...");
        $queues = [];
        foreach (explode(',', $queueNames) as $queueName) {
            // try to load the queue if it already exists
            // if the queue doesn't exist for any reason
            if ($queue = $rabbit->loadExistingQueue($queueName)) {
                $this->writeln("\tFound queue $queueName...");
                $queues[] = $queue;
            }
        }
        // create a consumer that will listen to all the queues
        $consumer = $rabbit->subscribeToMessages($queues, function (AmqpMessage $message, AmqpConsumer $consumer) use ($methodName) {
            return $this->{$methodName}($message, $consumer);
        }, prefetch: $prefetch);
        $this->writeln("Listening to queues...");
        // start listening
        while (true) {
            $consumer->consume($timeout);
        }
    }
}
