<?php

/**
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; under version 2
 * of the License (non-upgradable).
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 *
 * Copyright (c) 2017-2021 (original work) Open Assessment Technologies SA (under the project TAO-PRODUCT);
 *
 */

namespace oat\taoTaskQueue\model\QueueBroker;

use Doctrine\DBAL\FetchMode;
use Doctrine\DBAL\Query\QueryBuilder;
use Doctrine\DBAL\Schema\SchemaException;
use Doctrine\DBAL\Schema\Schema;
use Doctrine\DBAL\Schema\AbstractSchemaManager;
use oat\tao\model\taskQueue\Queue\Broker\AbstractQueueBroker;
use oat\tao\model\taskQueue\Task\TaskInterface;
use oat\taoTaskQueue\model\Task\CallbackTaskDecorator;

/**
 * Storing messages/tasks in DB.
 *
 * @author Gyula Szucs <gyula@taotesting.com>
 */
class RdsQueueBroker extends AbstractQueueBroker
{
    public const ID ='rds';

    private $persistenceId;

    /**
     * @var \common_persistence_SqlPersistence
     */
    protected $persistence;

    /**
     * RdsQueueBroker constructor.
     *
     * @param string $persistenceId
     * @param int $receiveTasks
     */
    public function __construct($persistenceId, $receiveTasks = 1)
    {
        parent::__construct($receiveTasks);

        if (empty($persistenceId)) {
            throw new \InvalidArgumentException("Persistence id needs to be set for " . __CLASS__);
        }

        $this->persistenceId = $persistenceId;
    }

    public function __toPhpCode()
    {
        return 'new ' . get_called_class() . '('
            . \common_Utils::toHumanReadablePhpString($this->persistenceId)
            . ', '
            . \common_Utils::toHumanReadablePhpString($this->getNumberOfTasksToReceive())
            . ')';
    }

    /**
     * @return \common_persistence_SqlPersistence
     */
    protected function getPersistence()
    {
        if (is_null($this->persistence)) {
            $this->persistence = $this->getServiceLocator()
                ->get(\common_persistence_Manager::SERVICE_ID)
                ->getPersistenceById($this->persistenceId);
        }

        return $this->persistence;
    }

    /**
     * @return string
     */
    protected function getTableName()
    {
        return strtolower($this->getQueueNameWithPrefix());
    }

    /**
     * Note: this method can be run multiple times because only the migrate queries (result of getMigrateSchemaSql) will be run.
     *
     * @inheritdoc
     */
    public function createQueue()
    {
        $persistence = $this->getPersistence();
        /** @var AbstractSchemaManager $schemaManager */
        $schemaManager = $persistence->getDriver()->getSchemaManager();

        /** @var Schema $schema */
        $schema = $schemaManager->createSchema();
        $fromSchema = clone $schema;
        try {
            if (in_array($this->getTableName(), $schemaManager->getTables())) {
                $schema->dropTable($this->getTableName());
            }
            $table = $schema->createTable($this->getTableName());
            $table->addOption('engine', 'InnoDB');
            $table->addColumn('id', 'integer', ["autoincrement" => true, "notnull" => true, "unsigned" => true]);
            $table->addColumn('message', 'text', ["notnull" => true]);
            $table->addColumn('visible', 'boolean', ["default" => 1]);
            $table->addColumn('created_at', 'datetime', ['notnull' => true]);
            $table->setPrimaryKey(['id']);
            $table->addIndex(['created_at', 'visible'], 'IDX_created_at_visible_' . $this->getQueueName());
        } catch (SchemaException $e) {
            $this->logDebug('Schema of ' . $this->getTableName() . ' table already up to date.');
        }

        $queries = $persistence->getPlatForm()->getMigrateSchemaSql($fromSchema, $schema);

        foreach ($queries as $query) {
            $persistence->exec($query);
        }

        if ($queries) {
            $this->logDebug('Queue ' . $this->getTableName() . ' created/updated in RDS.');
        }
    }

    /**
     * Insert a new task into the queue table.
     *
     * @param TaskInterface $task
     * @return bool
     */
    public function push(TaskInterface $task)
    {
        return (bool) $this->getPersistence()->insert($this->getTableName(), [
            'message' => $this->serializeTask($task),
            'created_at' => $this->getPersistence()->getPlatForm()->getNowExpression()
        ]);
    }

    /**
     * Does the DBAL specific pop mechanism.
     */
    protected function doPop()
    {
        $this->getPersistence()->getPlatform()->beginTransaction();

        $logContext = [
            'Queue' => $this->getQueueNameWithPrefix()
        ];

        try {
            $qb = $this->getQueryBuilder()
                ->select('id, message')
                ->from($this->getTableName())
                ->andWhere('visible = :visible')
                ->orderBy('created_at')
                ->setMaxResults($this->getNumberOfTasksToReceive());

            /**
             * SELECT ... FOR UPDATE is used for locking
             *
             * @see https://dev.mysql.com/doc/refman/5.6/en/innodb-locking-reads.html
             */
            $sql = $qb->getSQL() . ' ' . $this->getPersistence()->getPlatForm()->getWriteLockSQL();

            if ($dbResult = $this->getPersistence()->query($sql, ['visible' => 1])->fetchAll(\PDO::FETCH_ASSOC)) {
                // set the received messages to invisible for other workers
                $qb = $this->getQueryBuilder()
                    ->update($this->getTableName())
                    ->set('visible', ':visible')
                    ->where('id IN (' . implode(',', array_column($dbResult, 'id')) . ')')
                    ->setParameter('visible', 0);

                $qb->execute();

                foreach ($dbResult as $row) {
                    if ($task = $this->unserializeTask($row['message'], $row['id'], $logContext)) {
                        $task->setMetadata('RdsMessageId', $row['id']);
                        $this->pushPreFetchedMessage($task);
                    }
                }
            } else {
                $this->logDebug('No task in the queue.', $logContext);
            }

            $this->getPersistence()->getPlatform()->commit();
        } catch (\Exception $e) {
            $this->getPersistence()->getPlatform()->rollBack();
            $this->logError('Popping tasks failed with MSG: ' . $e->getMessage(), $logContext);
        }
    }

    /**
     * Delete the message after being processed by the worker.
     *
     * @param TaskInterface $task
     */
    public function delete(TaskInterface $task)
    {
        $this->doDelete($task->getMetadata('RdsMessageId'), [
            'InternalMessageId' => $task->getId(),
            'RdsMessageId' => $task->getMetadata('RdsMessageId')
        ]);
    }

    /**
     * @param string $id
     * @param array  $logContext
     * @return int
     */
    protected function doDelete($id, array $logContext = [])
    {
        try {
            $this->getQueryBuilder()
                ->delete($this->getTableName())
                ->where('id = :id')
                ->andWhere('visible = :visible')
                ->setParameter('id', (int) $id)
                ->setParameter('visible', 0)
                ->execute();
        } catch (\Exception $e) {
            $this->logError('Deleting task failed with MSG: ' . $e->getMessage(), $logContext);
        }
    }

    /**
     * @TODO Make queue broker open/closed: https://oat-sa.atlassian.net/browse/ADF-556
     */
    public function getTaskByTaskLogId(string $taskLogId): ?CallbackTaskDecorator
    {
        $logId = substr($taskLogId, strpos($taskLogId, '#'));

        $row = $this->getQueryBuilder()
            ->select('id, message, visible, created_at')
            ->from($this->getTableName())
            ->andWhere('message LIKE :taskLogId')
            ->setParameter('taskLogId', "%$logId%")
            ->setMaxResults(1)
            ->execute()
            ->fetch(FetchMode::ASSOCIATIVE);

        if (!$row) {
            return null;
        }

        $task = $this->unserializeTask(
            $row['message'],
            $row['id'],
            [
                'Queue' => $this->getQueueNameWithPrefix()
            ]
        );

        if (!$task) {
            return null;
        }

        return new CallbackTaskDecorator($task, $row['id']);
    }

    /**
     * @TODO Make queue broker open/closed: https://oat-sa.atlassian.net/browse/ADF-556
     */
    public function changeTaskVisibility(string $taskId, bool $visible): void
    {
        $this->getQueryBuilder()
            ->update($this->getTableName())
            ->set('visible', ':visible')
            ->where('id = :id')
            ->setParameter('visible', (int)$visible)
            ->setParameter('id', $taskId)
            ->execute();
    }

    /**
     * @return int
     */
    public function count()
    {
        try {
            $qb = $this->getQueryBuilder()
                ->select('COUNT(id)')
                ->from($this->getTableName())
                ->andWhere('visible = :visible')
                ->setParameter('visible', 1);

            return (int) $qb->execute()->fetchColumn();
        } catch (\Exception $e) {
            $this->logError('Counting tasks failed with MSG: ' . $e->getMessage());
        }

        return 0;
    }

    /**
     * @return QueryBuilder
     */
    private function getQueryBuilder()
    {
        return $this->getPersistence()->getPlatform()->getQueryBuilder();
    }
}