<commit_before><?php

/*
CREATE TABLE IF NOT EXISTS `jobs` (
  `id` mediumint(20) NOT NULL AUTO_INCREMENT,
  `queue` char(32) NOT NULL DEFAULT 'default',
  `data` mediumtext NOT NULL,
  `priority` int(1) NOT NULL DEFAULT '0',
  `expires_at` datetime DEFAULT NULL,
  `delay_until` datetime DEFAULT NULL,
  `locked` tinyint(1) NOT NULL DEFAULT '0',
  PRIMARY KEY (`id`),
  KEY `queue` (`queue`,`locked`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
*/

namespace josegonzalez\Queuesadilla\Engine;

use DateInterval;
use DateTime;
use PDO;
use PDOException;
use josegonzalez\Queuesadilla\Engine\Base;

class MysqlEngine extends Base
{
    protected $baseConfig = [
        'delay' => null,
        'database' => 'database_name',
        'expires_in' => null,
        'user' => null,
        'pass' => <PASSWORD>,
        'persistent' => true,
        'port' => 3306,
        'priority' => 0,
        'queue' => 'default',
        'host' => '127.0.0.1',
        'table' => 'jobs',
    ];


    /**
     * {@inheritDoc}
     */
    public function connect()
    {
        $config = $this->settings;
        if (empty($config['flags'])) {
            $config['flags'] = [];
        }

        $flags = [
            PDO::ATTR_PERSISTENT => $config['persistent'],
            PDO::MYSQL_ATTR_USE_BUFFERED_QUERY => true,
            PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION
        ] + $config['flags'];

        $dsn = "mysql:host={$config['host']};port={$config['port']};dbname={$config['database']}";

        try {
            $this->connection = new PDO(
                $dsn,
                $config['user'],
                $config['pass'],
                $flags
            );
        } catch (PDOException $e) {
            // TODO: Logging
            $this->connection = null;
        }

        return (bool)$this->connection;
    }

    /**
     * {@inheritDoc}
     */
    public function acknowledge($item)
    {
        if (!parent::acknowledge($item)) {
            return false;
        }

        $sql = sprintf('DELETE FROM `%s` WHERE id = ?', $this->config('table'));
        $sth = $this->connection()->prepare($sql);
        $sth->bindParam(1, $item['id'], PDO::PARAM_INT);
        $sth->execute();
        return $sth->rowCount() == 1;
    }

    /**
     * {@inheritDoc}
     */
    public function reject($item)
    {
        return $this->acknowledge($item);
    }

    /**
     * {@inheritDoc}
     */
    public function pop($options = [])
    {
        $queue = $this->setting($options, 'queue');
        $selectSql = implode(" ", [
            'SELECT `id`, `data` FROM `%s`',
            'WHERE `queue` = ? AND `locked` != 1',
            'AND (expires_at IS NULL OR expires_at > ?)',
            'AND (delay_until IS NULL OR delay_until < ?)',
            'ORDER BY priority ASC LIMIT 1 FOR UPDATE',
        ]);
        $selectSql = sprintf($selectSql, $this->config('table'));
        $updateSql = sprintf('UPDATE `%s` SET locked = 1 WHERE id = ?', $this->config('table'));

        $datetime = new DateTime;
        $dtFormatted = $datetime->format('Y-m-d H:i:s');

        try {
            $sth = $this->connection()->prepare($selectSql);
            $sth->bindParam(1, $queue, PDO::PARAM_STR);
            $sth->bindParam(2, $dtFormatted, PDO::PARAM_STR);
            $sth->bindParam(3, $dtFormatted, PDO::PARAM_STR);

            $this->connection()->beginTransaction();
            $sth->execute();
            $result = $sth->fetch(PDO::FETCH_ASSOC);

            if (!empty($result)) {
                $sth = $this->connection()->prepare($updateSql);
                $sth->bindParam(1, $result['id'], PDO::PARAM_INT);
                $sth->execute();
                $this->connection()->commit();
                if ($sth->rowCount() == 1) {
                    $data = json_decode($result['data'], true);
                    return [
                        'id' => $result['id'],
                        'class' => $data['class'],
                        'args' => $data['args'],
                        'queue' => $queue,
                        'options' => $data['options'],
                    ];
                }
            }
            $this->connection()->commit();
        } catch (PDOException $e) {
            $this->connection()->rollBack();
        }

        return null;
    }

    /**
     * {@inheritDoc}
     */
    public function push($item, $options = [])
    {
        if (!is_array($options)) {
            $options = ['queue' => $options];
        }

        $delay = $this->setting($options, 'delay');
        $expiresIn = $this->setting($options, 'expires_in');
        $queue = $this->setting($options, 'queue');
        $priority = $this->setting($options, 'priority');

        $delayUntil = null;
        if ($delay !== null) {
            $datetime = new DateTime;
            $delayUntil = $datetime->add(new DateInterval(sprintf('PT%sS', $delay)))->format('Y-m-d H:i:s');
        }

        $expiresAt = null;
        if ($expiresIn !== null) {
            $datetime = new DateTime;<commit_msg>Log issues in the MysqlEngine<commit_after><?php

/*
CREATE TABLE IF NOT EXISTS `jobs` (
  `id` mediumint(20) NOT NULL AUTO_INCREMENT,
  `queue` char(32) NOT NULL DEFAULT 'default',
  `data` mediumtext NOT NULL,
  `priority` int(1) NOT NULL DEFAULT '0',
  `expires_at` datetime DEFAULT NULL,
  `delay_until` datetime DEFAULT NULL,
  `locked` tinyint(1) NOT NULL DEFAULT '0',
  PRIMARY KEY (`id`),
  KEY `queue` (`queue`,`locked`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
*/

namespace josegonzalez\Queuesadilla\Engine;

use DateInterval;
use DateTime;
use PDO;
use PDOException;
use josegonzalez\Queuesadilla\Engine\Base;

class MysqlEngine extends Base
{
    protected $baseConfig = [
        'delay' => null,
        'database' => 'database_name',
        'expires_in' => null,
        'user' => null,
        'pass' => <PASSWORD>,
        'persistent' => true,
        'port' => 3306,
        'priority' => 0,
        'queue' => 'default',
        'host' => '127.0.0.1',
        'table' => 'jobs',
    ];


    /**
     * {@inheritDoc}
     */
    public function connect()
    {
        $config = $this->settings;
        if (empty($config['flags'])) {
            $config['flags'] = [];
        }

        $flags = [
            PDO::ATTR_PERSISTENT => $config['persistent'],
            PDO::MYSQL_ATTR_USE_BUFFERED_QUERY => true,
            PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION
        ] + $config['flags'];

        $dsn = "mysql:host={$config['host']};port={$config['port']};dbname={$config['database']}";

        try {
            $this->connection = new PDO(
                $dsn,
                $config['user'],
                $config['pass'],
                $flags
            );
        } catch (PDOException $e) {
            $this->logger()->error($e->getMessage());
            $this->connection = null;
        }

        return (bool)$this->connection;
    }

    /**
     * {@inheritDoc}
     */
    public function acknowledge($item)
    {
        if (!parent::acknowledge($item)) {
            return false;
        }

        $sql = sprintf('DELETE FROM `%s` WHERE id = ?', $this->config('table'));
        $sth = $this->connection()->prepare($sql);
        $sth->bindParam(1, $item['id'], PDO::PARAM_INT);
        $sth->execute();
        return $sth->rowCount() == 1;
    }

    /**
     * {@inheritDoc}
     */
    public function reject($item)
    {
        return $this->acknowledge($item);
    }

    /**
     * {@inheritDoc}
     */
    public function pop($options = [])
    {
        $queue = $this->setting($options, 'queue');
        $selectSql = implode(" ", [
            'SELECT `id`, `data` FROM `%s`',
            'WHERE `queue` = ? AND `locked` != 1',
            'AND (expires_at IS NULL OR expires_at > ?)',
            'AND (delay_until IS NULL OR delay_until < ?)',
            'ORDER BY priority ASC LIMIT 1 FOR UPDATE',
        ]);
        $selectSql = sprintf($selectSql, $this->config('table'));
        $updateSql = sprintf('UPDATE `%s` SET locked = 1 WHERE id = ?', $this->config('table'));

        $datetime = new DateTime;
        $dtFormatted = $datetime->format('Y-m-d H:i:s');

        try {
            $sth = $this->connection()->prepare($selectSql);
            $sth->bindParam(1, $queue, PDO::PARAM_STR);
            $sth->bindParam(2, $dtFormatted, PDO::PARAM_STR);
            $sth->bindParam(3, $dtFormatted, PDO::PARAM_STR);

            $this->connection()->beginTransaction();
            $sth->execute();
            $result = $sth->fetch(PDO::FETCH_ASSOC);

            if (!empty($result)) {
                $sth = $this->connection()->prepare($updateSql);
                $sth->bindParam(1, $result['id'], PDO::PARAM_INT);
                $sth->execute();
                $this->connection()->commit();
                if ($sth->rowCount() == 1) {
                    $data = json_decode($result['data'], true);
                    return [
                        'id' => $result['id'],
                        'class' => $data['class'],
                        'args' => $data['args'],
                        'queue' => $queue,
                        'options' => $data['options'],
                    ];
                }
            }
            $this->connection()->commit();
        } catch (PDOException $e) {
            $this->logger()->error($e->getMessage());
            $this->connection()->rollBack();
        }

        return null;
    }

    /**
     * {@inheritDoc}
     */
    public function push($item, $options = [])
    {
        if (!is_array($options)) {
            $options = ['queue' => $options];
        }

        $delay = $this->setting($options, 'delay');
        $expiresIn = $this->setting($options, 'expires_in');
        $queue = $this->setting($options, 'queue');
        $priority = $this->setting($options, 'priority');

        $delayUntil = null;
        if ($delay !== null) {
            $datetime = new DateTime;
            $delayUntil = $datetime->add(new DateInterval(sprintf('PT%sS', $delay)))->format('Y-m-d H:i:s');
        }

        $expiresAt = null;
        if ($expiresIn !== null) {
            $datetime = new DateTime;
{'fraction_non_alphanumeric': 0.17773647657243472, 'fraction_numerical': 0.005869118654015455, 'mean_word_length': 1.8638655462184874, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 135, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '974792', 'n_tokens_mistral': 3189, 'n_tokens_neox': 3064, 'n_words': 615}