<?php

namespace Illuminate\Redis\Connections;

use Closure;
use ErrorException;
use Illuminate\Contracts\Redis\Connection as ConnectionContract;
use Illuminate\Support\Collection;
use Illuminate\Support\Str;
use RedisClusterException;
use RedisException;
use Throwable;

/**
 * @mixin \Redis
 */
class PhpRedisConnection extends Connection implements ConnectionContract
{
    use PacksPhpRedisValues;

    /**
     * The commands that may be safely retried after reconnecting.
     *
     * @var list<string>
     */
    protected const RETRYABLE_COMMANDS = [
        'bitcount',
        'bitpos',
        'dbsize',
        'dump',
        'exists',
        'geodist',
        'geohash',
        'geopos',
        'geosearch',
        'get',
        'getbit',
        'getrange',
        'hexists',
        'hget',
        'hgetall',
        'hkeys',
        'hlen',
        'hmget',
        'hmset',
        'hstrlen',
        'hvals',
        'keys',
        'lindex',
        'llen',
        'lpos',
        'lrange',
        'mget',
        'mset',
        'ping',
        'pttl',
        'randomkey',
        'scard',
        'sdiff',
        'sinter',
        'sismember',
        'smembers',
        'smismember',
        'srandmember',
        'strlen',
        'sunion',
        'time',
        'ttl',
        'type',
        'xinfo',
        'xlen',
        'xpending',
        'xrange',
        'xrevrange',
        'zcard',
        'zcount',
        'zlexcount',
        'zmscore',
        'zrange',
        'zrank',
        'zrevrank',
        'zscore',
    ];

    /**
     * The connection creation callback.
     *
     * @var callable
     */
    protected $connector;

    /**
     * The connection configuration array.
     *
     * @var array
     */
    protected $config;

    /**
     * Create a new PhpRedis connection.
     *
     * @param  \Redis  $client
     * @param  callable|null  $connector
     * @param  array  $config
     */
    public function __construct($client, ?callable $connector = null, array $config = [])
    {
        $this->client = $client;
        $this->config = $config;
        $this->connector = $connector;
    }

    /**
     * Returns the value of the given key.
     *
     * @param  string  $key
     * @return string|null
     */
    public function get($key)
    {
        $result = $this->command('get', [$key]);

        return $result !== false ? $result : null;
    }

    /**
     * Get the values of all the given keys.
     *
     * @param  array  $keys
     * @return array
     */
    public function mget(array $keys)
    {
        $result = $this->command('mget', [$keys]);

        if ($result === false) {
            return array_fill(0, count($keys), null);
        }

        return array_map(function ($value) {
            return $value !== false ? $value : null;
        }, $result);
    }

    /**
     * Set the string value in the argument as the value of the key.
     *
     * @param  string  $key
     * @param  mixed  $value
     * @param  string|null  $expireResolution
     * @param  int|null  $expireTTL
     * @param  string|null  $flag
     * @return bool
     */
    public function set($key, $value, $expireResolution = null, $expireTTL = null, $flag = null)
    {
        return $this->command('set', [
            $key,
            $value,
            $expireResolution ? [$flag, $expireResolution => $expireTTL] : null,
        ]);
    }

    /**
     * Set the given key if it doesn't exist.
     *
     * @param  string  $key
     * @param  string  $value
     * @return int
     */
    public function setnx($key, $value)
    {
        return (int) $this->command('setnx', [$key, $value]);
    }

    /**
     * Get the value of the given hash fields.
     *
     * @param  string  $key
     * @param  mixed  ...$dictionary
     * @return array
     */
    public function hmget($key, ...$dictionary)
    {
        if (count($dictionary) === 1) {
            $dictionary = $dictionary[0];
        }

        $result = $this->command('hmget', [$key, $dictionary]);

        if ($result === false) {
            return array_fill(0, count((array) $dictionary), null);
        }

        return array_values($result);
    }

    /**
     * Set the given hash fields to their respective values.
     *
     * @param  string  $key
     * @param  mixed  ...$dictionary
     * @return int
     */
    public function hmset($key, ...$dictionary)
    {
        if (count($dictionary) === 1) {
            $dictionary = $dictionary[0];
        } else {
            $input = new Collection($dictionary);

            $dictionary = $input->nth(2)->combine($input->nth(2, 1))->toArray();
        }

        return $this->command('hmset', [$key, $dictionary]);
    }

    /**
     * Set the given hash field if it doesn't exist.
     *
     * @param  string  $hash
     * @param  string  $key
     * @param  string  $value
     * @return int
     */
    public function hsetnx($hash, $key, $value)
    {
        return (int) $this->command('hsetnx', [$hash, $key, $value]);
    }

    /**
     * Removes the first count occurrences of the value element from the list.
     *
     * @param  string  $key
     * @param  int  $count
     * @param  mixed  $value
     * @return int|false
     */
    public function lrem($key, $count, $value)
    {
        return $this->command('lrem', [$key, $value, $count]);
    }

    /**
     * Removes and returns the first element of the list stored at key.
     *
     * @param  mixed  ...$arguments
     * @return array|null
     */
    public function blpop(...$arguments)
    {
        $result = $this->command('blpop', $arguments);

        return empty($result) ? null : $result;
    }

    /**
     * Removes and returns the last element of the list stored at key.
     *
     * @param  mixed  ...$arguments
     * @return array|null
     */
    public function brpop(...$arguments)
    {
        $result = $this->command('brpop', $arguments);

        return empty($result) ? null : $result;
    }

    /**
     * Removes and returns a random element from the set value at key.
     *
     * @param  string  $key
     * @param  int|null  $count
     * @return mixed|false
     */
    public function spop($key, $count = 1)
    {
        return $this->command('spop', func_get_args());
    }

    /**
     * Add one or more members to a sorted set or update its score if it already exists.
     *
     * @param  string  $key
     * @param  mixed  ...$dictionary
     * @return int
     */
    public function zadd($key, ...$dictionary)
    {
        if (is_array(end($dictionary))) {
            foreach (array_pop($dictionary) as $member => $score) {
                $dictionary[] = $score;
                $dictionary[] = $member;
            }
        }

        $options = [];

        foreach (array_slice($dictionary, 0, 3) as $i => $value) {
            if (in_array($value, ['nx', 'xx', 'ch', 'incr', 'gt', 'lt', 'NX', 'XX', 'CH', 'INCR', 'GT', 'LT'], true)) {
                $options[] = $value;

                unset($dictionary[$i]);
            }
        }

        return $this->command('zadd', array_merge([$key], [$options], array_values($dictionary)));
    }

    /**
     * Return elements with score between $min and $max.
     *
     * @param  string  $key
     * @param  mixed  $min
     * @param  mixed  $max
     * @param  array  $options
     * @return array
     */
    public function zrangebyscore($key, $min, $max, $options = [])
    {
        if (isset($options['limit']) && ! array_is_list($options['limit'])) {
            $options['limit'] = [
                $options['limit']['offset'],
                $options['limit']['count'],
            ];
        }

        return $this->command('zRangeByScore', [$key, $min, $max, $options]);
    }

    /**
     * Return elements with score between $min and $max.
     *
     * @param  string  $key
     * @param  mixed  $min
     * @param  mixed  $max
     * @param  array  $options
     * @return array
     */
    public function zrevrangebyscore($key, $min, $max, $options = [])
    {
        if (isset($options['limit']) && ! array_is_list($options['limit'])) {
            $options['limit'] = [
                $options['limit']['offset'],
                $options['limit']['count'],
            ];
        }

        return $this->command('zRevRangeByScore', [$key, $min, $max, $options]);
    }

    /**
     * Find the intersection between sets and store in a new set.
     *
     * @param  string  $output
     * @param  array  $keys
     * @param  array  $options
     * @return int
     */
    public function zinterstore($output, $keys, $options = [])
    {
        return $this->command('zinterstore', [$output, $keys,
            $options['weights'] ?? null,
            $options['aggregate'] ?? 'sum',
        ]);
    }

    /**
     * Find the union between sets and store in a new set.
     *
     * @param  string  $output
     * @param  array  $keys
     * @param  array  $options
     * @return int
     */
    public function zunionstore($output, $keys, $options = [])
    {
        return $this->command('zunionstore', [$output, $keys,
            $options['weights'] ?? null,
            $options['aggregate'] ?? 'sum',
        ]);
    }

    /**
     * Scans all keys based on options.
     *
     * @param  mixed  $cursor
     * @param  array  $options
     * @return mixed
     */
    public function scan($cursor, $options = [])
    {
        $result = $this->client->scan($cursor,
            $options['match'] ?? '*',
            $options['count'] ?? 10
        );

        if ($result === false) {
            $result = [];
        }

        return $cursor === 0 && empty($result) ? false : [$cursor, $result];
    }

    /**
     * Scans the given set for all values based on options.
     *
     * @param  string  $key
     * @param  mixed  $cursor
     * @param  array  $options
     * @return mixed
     */
    public function zscan($key, $cursor, $options = [])
    {
        $result = $this->client->zscan($key, $cursor,
            $options['match'] ?? '*',
            $options['count'] ?? 10
        );

        if ($result === false) {
            $result = [];
        }

        return $cursor === 0 && empty($result) ? false : [$cursor, $result];
    }

    /**
     * Scans the given hash for all values based on options.
     *
     * @param  string  $key
     * @param  mixed  $cursor
     * @param  array  $options
     * @return mixed
     */
    public function hscan($key, $cursor, $options = [])
    {
        $result = $this->client->hscan($key, $cursor,
            $options['match'] ?? '*',
            $options['count'] ?? 10
        );

        if ($result === false) {
            $result = [];
        }

        return $cursor === 0 && empty($result) ? false : [$cursor, $result];
    }

    /**
     * Scans the given set for all values based on options.
     *
     * @param  string  $key
     * @param  mixed  $cursor
     * @param  array  $options
     * @return mixed
     */
    public function sscan($key, $cursor, $options = [])
    {
        $result = $this->client->sscan($key, $cursor,
            $options['match'] ?? '*',
            $options['count'] ?? 10
        );

        if ($result === false) {
            $result = [];
        }

        return $cursor === 0 && empty($result) ? false : [$cursor, $result];
    }

    /**
     * Execute commands in a pipeline.
     *
     * @param  callable|null  $callback
     * @return \Redis|array
     */
    public function pipeline(?callable $callback = null)
    {
        $pipeline = $this->retryOnceOnLostConnection(fn () => $this->client()->pipeline());

        if (is_null($callback)) {
            return $pipeline;
        }

        try {
            return tap($pipeline, $callback)->exec();
        } catch (Throwable $e) {
            rescue(fn () => $this->client()->discard(), null, false);

            $this->rebuildClientOnLostConnection($e);

            throw $e;
        }
    }

    /**
     * Execute commands in a transaction.
     *
     * @param  callable|null  $callback
     * @return \Redis|array
     */
    public function transaction(?callable $callback = null)
    {
        $transaction = $this->retryOnceOnLostConnection(fn () => $this->client()->multi());

        if (is_null($callback)) {
            return $transaction;
        }

        try {
            return tap($transaction, $callback)->exec();
        } catch (Throwable $e) {
            rescue(fn () => $this->client()->discard(), null, false);

            $this->rebuildClientOnLostConnection($e);

            throw $e;
        }
    }

    /**
     * Evaluate a LUA script serverside, from the SHA1 hash of the script instead of the script itself.
     *
     * @param  string  $script
     * @param  int  $numkeys
     * @param  mixed  ...$arguments
     * @return mixed
     */
    public function evalsha($script, $numkeys, ...$arguments)
    {
        return $this->command('evalsha', [
            $this->script('load', $script), $arguments, $numkeys,
        ]);
    }

    /**
     * Evaluate a script and return its result.
     *
     * @param  string  $script
     * @param  int  $numberOfKeys
     * @param  mixed  ...$arguments
     * @return mixed
     */
    public function eval($script, $numberOfKeys, ...$arguments)
    {
        return $this->command('eval', [$script, $arguments, $numberOfKeys]);
    }

    /**
     * Subscribe to a set of given channels for messages.
     *
     * @param  array|string  $channels
     * @param  \Closure  $callback
     * @return void
     */
    public function subscribe($channels, Closure $callback)
    {
        $this->client->subscribe((array) $channels, function ($redis, $channel, $message) use ($callback) {
            $callback($message, $channel);
        });
    }

    /**
     * Subscribe to a set of given channels with wildcards.
     *
     * @param  array|string  $channels
     * @param  \Closure  $callback
     * @return void
     */
    public function psubscribe($channels, Closure $callback)
    {
        $this->client->psubscribe((array) $channels, function ($redis, $pattern, $channel, $message) use ($callback) {
            $callback($message, $channel);
        });
    }

    /**
     * Subscribe to a set of given channels for messages.
     *
     * @param  array|string  $channels
     * @param  \Closure  $callback
     * @param  string  $method
     * @return void
     */
    public function createSubscription($channels, Closure $callback, $method = 'subscribe')
    {
        //
    }

    /**
     * Flush the selected Redis database.
     *
     * @return mixed
     */
    public function flushdb()
    {
        $arguments = func_get_args();

        if (strtoupper((string) ($arguments[0] ?? null)) === 'ASYNC') {
            return $this->command('flushdb', [true]);
        }

        return $this->command('flushdb');
    }

    /**
     * Execute a raw command.
     *
     * @param  array  $parameters
     * @return mixed
     */
    public function executeRaw(array $parameters)
    {
        return $this->command('rawCommand', $parameters);
    }

    /**
     * Run a command against the Redis database.
     *
     * @param  string  $method
     * @param  array  $parameters
     * @return mixed
     *
     * @throws \RedisClusterException
     * @throws \RedisException
     */
    public function command($method, array $parameters = [])
    {
        $retries = max(
            $this->isRetryable($method, $parameters) ? 1 : 0,
            (int) ($this->config['command_retries'] ?? 0),
        );

        while (true) {
            try {
                return parent::command($method, $parameters);
            } catch (RedisClusterException|RedisException|ErrorException $e) {
                if (! $this->causedByLostConnection($e)) {
                    throw $e;
                }

                $this->rebuildClient();

                if ($retries-- === 0) {
                    throw $e;
                }
            }
        }
    }

    /**
     * Determine whether the command may be safely retried.
     *
     * @param  string  $method
     * @param  array  $parameters
     * @return bool
     */
    protected function isRetryable($method, array $parameters)
    {
        $method = strtolower($method);

        if ($method === 'set') {
            return ! isset($parameters[2]);
        }

        return in_array($method, static::RETRYABLE_COMMANDS, true);
    }

    /**
     * Run the given callback, retrying it once on a rebuilt client if the connection was lost.
     *
     * For operations that have not yet sent a command that could have taken effect, such as opening a pipeline or a transaction.
     *
     * @param  \Closure  $callback
     * @return mixed
     */
    protected function retryOnceOnLostConnection(Closure $callback)
    {
        $retries = 1;

        while (true) {
            try {
                return $callback();
            } catch (RedisClusterException|RedisException|ErrorException $e) {
                if (! $this->causedByLostConnection($e)) {
                    throw $e;
                }

                $this->rebuildClient();

                if ($retries-- === 0) {
                    throw $e;
                }
            }
        }
    }

    /**
     * Rebuild the client if the given exception was caused by a lost connection.
     *
     * @param  \Throwable  $e
     * @return void
     */
    protected function rebuildClientOnLostConnection(Throwable $e)
    {
        if ($this->causedByLostConnection($e)) {
            $this->rebuildClient();
        }
    }

    /**
     * Determine if the given exception was caused by a lost connection to the Redis server.
     *
     * @param  \Throwable  $e
     * @return bool
     */
    protected function causedByLostConnection(Throwable $e)
    {
        if (! $e instanceof RedisClusterException && ! $e instanceof RedisException && ! $e instanceof ErrorException) {
            return false;
        }

        return Str::contains($e->getMessage(), [
            'went away',
            'socket',
            'Error while reading',
            'read error on connection',
            'READONLY',
            'Connection lost',
            'Error processing response from Redis node',
            'Connection reset by peer',
        ]);
    }

    /**
     * Replace the underlying client with a freshly connected one.
     *
     * @return void
     */
    protected function rebuildClient()
    {
        $this->client = $this->connector ? call_user_func($this->connector) : $this->client;
    }

    /**
     * Disconnects from the Redis instance.
     *
     * @return void
     */
    public function disconnect()
    {
        $this->client->close();
    }

    /**
     * Pass other method calls down to the underlying client.
     *
     * @param  string  $method
     * @param  array  $parameters
     * @return mixed
     */
    public function __call($method, $parameters)
    {
        return parent::__call(strtolower($method), $parameters);
    }
}