<?php

/*
 * This file is part of the Predis package.
 *
 * (c) 2009-2020 Daniele Alessandri
 * (c) 2021-2025 Till Krüss
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Predis\Pipeline;

use Exception;
use InvalidArgumentException;
use Predis\ClientContextInterface;
use Predis\ClientException;
use Predis\ClientInterface;
use Predis\Command\CommandInterface;
use Predis\CommunicationException;
use Predis\Connection\AggregateConnectionInterface;
use Predis\Connection\Cluster\RedisCluster;
use Predis\Connection\ConnectionException;
use Predis\Connection\ConnectionInterface;
use Predis\Connection\Replication\ReplicationInterface;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
use Predis\Response\ResponseInterface;
use Predis\Response\ServerException;
use Predis\TimeoutException;
use SplQueue;
use Throwable;

/**
 * Implementation of a command pipeline in which write and read operations of
 * Redis commands are pipelined to alleviate the effects of network round-trips.
 *
 * {@inheritdoc}
 */
class Pipeline implements ClientContextInterface
{
    protected $client;
    private $pipeline;

    private $responses = [];
    private $running = false;

    /**
     * @param ClientInterface $client Client instance used by the context.
     */
    public function __construct(ClientInterface $client)
    {
        $this->client = $client;
        $this->pipeline = new SplQueue();
    }

    /**
     * Queues a command into the pipeline buffer.
     *
     * @param string $method    Command ID.
     * @param array  $arguments Arguments for the command.
     *
     * @return $this
     */
    public function __call($method, $arguments)
    {
        $command = $this->client->createCommand($method, $arguments);
        $this->recordCommand($command);

        return $this;
    }

    /**
     * Queues a command instance into the pipeline buffer.
     *
     * @param CommandInterface $command Command to be queued in the buffer.
     */
    protected function recordCommand(CommandInterface $command)
    {
        $this->pipeline->enqueue($command);
    }

    /**
     * Queues a command instance into the pipeline buffer.
     *
     * @param CommandInterface $command Command instance to be queued in the buffer.
     *
     * @return $this
     */
    public function executeCommand(CommandInterface $command)
    {
        $this->recordCommand($command);

        return $this;
    }

    /**
     * Throws an exception on -ERR responses returned by Redis.
     *
     * @param ConnectionInterface    $connection Redis connection that returned the error.
     * @param ErrorResponseInterface $response   Instance of the error response.
     *
     * @throws ServerException
     */
    protected function exception(ConnectionInterface $connection, ErrorResponseInterface $response)
    {
        $connection->disconnect();
        $message = $response->getMessage();

        throw new ServerException($message);
    }

    /**
     * Returns the underlying connection to be used by the pipeline.
     *
     * @return ConnectionInterface
     */
    protected function getConnection()
    {
        $connection = $this->getClient()->getConnection();

        if ($connection instanceof ReplicationInterface) {
            $connection->switchToMaster();
        }

        return $connection;
    }

    /**
     * Implements the logic to flush the queued commands and read the responses
     * from the current connection.
     *
     * @param ConnectionInterface $connection Current connection instance.
     * @param SplQueue            $commands   Queued commands.
     *
     * @return array
     * @throws Throwable
     */
    protected function executePipeline(ConnectionInterface $connection, SplQueue $commands)
    {
        $retry = $connection->getParameters()->retry;
        $backupQueue = $this->createDeepCloneQueue($commands);

        return $retry->callWithRetry(
            function () use ($connection, &$commands) {
                return $this->executePipelineInternal($connection, $commands);
            },
            function (Throwable $e) use (&$commands, $backupQueue, $connection) {
                if (!$e instanceof CommunicationException) {
                    throw $e;
                }

                if ($connection instanceof AggregateConnectionInterface) {
                    $this->onAggregateConnectionFailCallback($connection, $e);
                } else {
                    $connection = $e->getConnection();
                    $connection->disconnect();
                }

                // In case of error whole pipeline should be retried
                // So we need to write all original commands again
                $commands = $this->createDeepCloneQueue($backupQueue);
            }
        );
    }

    /**
     * @param  ConnectionInterface $connection
     * @param  SplQueue            $commands
     * @return array
     * @throws ServerException
     * @throws Throwable
     */
    protected function executePipelineInternal(
        ConnectionInterface $connection,
        SplQueue $commands
    ): array {
        $responses = [];
        $exceptions = $this->throwServerExceptions();
        $protocolVersion = (int) $connection->getParameters()->protocol;

        if ($connection instanceof AggregateConnectionInterface) {
            $this->writeToMultiNode($connection, $commands);
        } else {
            $this->writeToSingleNode($connection, $commands);
        }

        while (!$commands->isEmpty()) {
            $command = $commands->dequeue();

            if ($connection instanceof AggregateConnectionInterface) {
                $response = $connection->getConnectionByCommand($command)->readResponse($command);
            } else {
                $response = $connection->readResponse($command);
            }

            if (!$response instanceof ResponseInterface) {
                if ($protocolVersion === 2) {
                    $responses[] = $command->parseResponse($response);
                } else {
                    $responses[] = $command->parseResp3Response($response);
                }
            } elseif ($response instanceof ErrorResponseInterface && $exceptions) {
                $this->exception($connection, $response);
            } else {
                $responses[] = $response;
            }
        }

        return $responses;
    }

    /**
     * Creates a deep copy of commands queue for backup.
     *
     * @param  SplQueue $queue
     * @return SplQueue
     */
    private function createDeepCloneQueue(SplQueue $queue): SplQueue
    {
        $new = new SplQueue();

        foreach ($queue as $command) {
            $new->enqueue(clone $command);
        }

        return $new;
    }

    /**
     * Writes pipelined commands to single node connection.
     *
     * @param  ConnectionInterface $connection
     * @param  SplQueue            $commands
     * @return void
     * @throws Throwable
     */
    protected function writeToSingleNode(ConnectionInterface $connection, SplQueue $commands)
    {
        $buffer = '';

        foreach ($commands as $command) {
            $buffer .= $command->serializeCommand();
        }

        $connection->write($buffer);
    }

    /**
     * Writes pipelined commands to multi node connection.
     *
     * @param  AggregateConnectionInterface $connection
     * @param  SplQueue                     $commands
     * @return void
     * @throws Throwable
     */
    protected function writeToMultiNode(AggregateConnectionInterface $connection, SplQueue $commands)
    {
        $retry = $connection->getParameters()->retry;

        foreach ($commands as $command) {
            $nodeConnection = $connection->getConnectionByCommand($command);
            $nodeConnection->write($command->serializeCommand());
        }
    }

    /**
     * Flushes the buffer holding all of the commands queued so far.
     *
     * @param bool $send Specifies if the commands in the buffer should be sent to Redis.
     *
     * @return $this
     */
    public function flushPipeline($send = true)
    {
        if ($send && !$this->pipeline->isEmpty()) {
            $responses = $this->executePipeline($this->getConnection(), $this->pipeline);
            $this->responses = array_merge($this->responses, $responses);
        } else {
            $this->pipeline = new SplQueue();
        }

        return $this;
    }

    /**
     * Marks the running status of the pipeline.
     *
     * @param bool $bool Sets the running status of the pipeline.
     *
     * @throws ClientException
     */
    private function setRunning($bool)
    {
        if ($bool && $this->running) {
            throw new ClientException('The current pipeline context is already being executed.');
        }

        $this->running = $bool;
    }

    /**
     * Handles the actual execution of the whole pipeline.
     *
     * @param mixed $callable Optional callback for execution.
     *
     * @return array
     * @throws Exception
     * @throws InvalidArgumentException
     */
    public function execute($callable = null)
    {
        if ($callable && !is_callable($callable)) {
            throw new InvalidArgumentException('The argument must be a callable object.');
        }

        $exception = null;
        $this->setRunning(true);

        try {
            if ($callable) {
                call_user_func($callable, $this);
            }

            $this->flushPipeline();
        } catch (Exception $exception) {
            // NOOP
        }

        $this->setRunning(false);

        if ($exception) {
            throw $exception;
        }

        return $this->responses;
    }

    /**
     * Returns if the pipeline should throw exceptions on server errors.
     *
     * @return bool
     */
    protected function throwServerExceptions()
    {
        return (bool) $this->client->getOptions()->exceptions;
    }

    /**
     * Returns the underlying client instance used by the pipeline object.
     *
     * @return ClientInterface
     */
    public function getClient()
    {
        return $this->client;
    }

    /**
     * Handle aggregate connection exception.
     *
     * @param  AggregateConnectionInterface $connection
     * @param  CommunicationException       $e
     * @return void
     */
    private function onAggregateConnectionFailCallback(AggregateConnectionInterface $connection, Throwable $e)
    {
        if ($e instanceof ConnectionException) {
            $nodeConnection = $e->getConnection();

            if ($nodeConnection) {
                $nodeConnection->disconnect();
                $connection->remove($nodeConnection);
            }

            if ($connection instanceof RedisCluster) {
                if ($connection->useClusterSlots) {
                    $connection->askSlotMap();
                }
            }
        }

        if ($e instanceof TimeoutException) {
            $nodeConnection = $e->getConnection();

            if ($nodeConnection) {
                $nodeConnection->disconnect();
            }
        }
    }
}
