/*
   Copyright (c) 2018-2019 Nokia.

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

#include "private/redis/asynchiredisclustercommanddispatcher.hpp"
#include <algorithm>
#include <cstring>
#include <cerrno>
#include <sstream>
#include "private/abort.hpp"
#include "private/createlogger.hpp"
#include "private/error.hpp"
#include "private/logger.hpp"
#include "private/redis/asyncredisreply.hpp"
#include "private/redis/reply.hpp"
#include "private/redis/hiredisclustersystem.hpp"
#include "private/engine.hpp"
#include "private/redis/hiredisclusterepolladapter.hpp"
#include "private/redis/contents.hpp"
#include "private/redis/redisgeneral.hpp"

using namespace shareddatalayer;
using namespace shareddatalayer::redis;

namespace
{
    void connectCb(const redisClusterAsyncContext*, const redisAsyncContext* ac, int status)
    {
        if (!status)
        {
            std::ostringstream msg;
            msg << "redis cluster instance connected, fd: " << ac->c.fd;
            logDebugOnce(msg.str());
        }
    }

    void disconnectCb(const redisClusterAsyncContext* acc, const redisAsyncContext* ac, int status)
    {
        if (status)
        {
            std::ostringstream msg;
            msg << "redis cluster instance disconnected, fd: " << ac->c.fd
                << ", status: " << ac->err;
            logDebugOnce(msg.str());
        }
        auto instance(static_cast<AsyncHiredisClusterCommandDispatcher*>(acc->data));
        instance->handleDisconnect(ac);
    }

    void cb(redisClusterAsyncContext* acc, void* rr, void* pd)
    {
        auto instance(static_cast<AsyncHiredisClusterCommandDispatcher*>(acc->data));
        auto reply(static_cast<redisReply*>(rr));
        auto cb(static_cast<AsyncHiredisClusterCommandDispatcher::CommandCb*>(pd));
        if (instance->isClientCallbacksEnabled())
            instance->handleReply(*cb, getRedisError(acc->err, acc->errstr, reply), reply);
    }
}

AsyncHiredisClusterCommandDispatcher::AsyncHiredisClusterCommandDispatcher(Engine& engine,
                                                                           const boost::optional<std::string>& ns,
                                                                           const DatabaseConfiguration::Addresses& addresses,
                                                                           std::shared_ptr<ContentsBuilder> contentsBuilder,
                                                                           bool usePermanentCommandCallbacks,
                                                                           std::shared_ptr<Logger> logger):
    AsyncHiredisClusterCommandDispatcher(engine,
                                         ns,
                                         addresses,
                                         contentsBuilder,
                                         usePermanentCommandCallbacks,
                                         HiredisClusterSystem::getInstance(),
                                         std::make_shared<HiredisClusterEpollAdapter>(engine),
                                         logger)
{
}

AsyncHiredisClusterCommandDispatcher::AsyncHiredisClusterCommandDispatcher(Engine& engine,
                                                                           const boost::optional<std::string>& ns,
                                                                           const DatabaseConfiguration::Addresses& addresses,
                                                                           std::shared_ptr<ContentsBuilder> contentsBuilder,
                                                                           bool usePermanentCommandCallbacks,
                                                                           HiredisClusterSystem& hiredisClusterSystem,
                                                                           std::shared_ptr<HiredisClusterEpollAdapter> adapter,
                                                                           std::shared_ptr<Logger> logger):
    engine(engine),
    initialNamespace(ns),
    addresses(addresses),
    contentsBuilder(contentsBuilder),
    usePermanentCommandCallbacks(usePermanentCommandCallbacks),
    hiredisClusterSystem(hiredisClusterSystem),
    adapter(adapter),
    acc(nullptr),
    serviceState(ServiceState::DISCONNECTED),
    clientCallbacksEnabled(true),
    connectionRetryTimer(engine),
    connectionRetryTimerDuration(std::chrono::seconds(1)),
    logger(logger)
{
    connect();
}

AsyncHiredisClusterCommandDispatcher::~AsyncHiredisClusterCommandDispatcher()
{
    disconnectHiredisCluster();
}

void AsyncHiredisClusterCommandDispatcher::connect()
{
    // Disconnect and free possibly (in re-connecting scenarios) already existing Redis cluster context.
    disconnectHiredisCluster();
    acc = hiredisClusterSystem.redisClusterAsyncConnect(formatToClusterSyntax(addresses).c_str(),
                                                        HIRCLUSTER_FLAG_ROUTE_USE_SLOTS);
    if (acc == nullptr)
    {
        logger->error() << "SDL: connecting to redis cluster failed, null context returned";
        armConnectionRetryTimer();
        return;
    }
    if (acc->err)
    {
        logger->error() << "SDL: connecting to redis cluster failed, error: " << acc->err;
        armConnectionRetryTimer();
        return;
    }
    acc->data = this;
    adapter->setup(acc);
    hiredisClusterSystem.redisClusterAsyncSetConnectCallback(acc, connectCb);
    hiredisClusterSystem.redisClusterAsyncSetDisconnectCallback(acc, disconnectCb);
    verifyConnection();
}

void AsyncHiredisClusterCommandDispatcher::verifyConnection()
{
    /* redisClusterAsyncConnect only queries available cluster nodes but it does
     * not connect to any cluster node (as it does not know to which node it should connect to).
     * As we use same cluster node for one SDL namespace (namespace is a hash tag), it is already
     * determined to which cluster node this instance will connect to. We do initial operation
     * to get connection to right redis node established already now. This also verifies that
     * connection really works. When Redis has max amount of users, it will still accept new
     * connections but is will close them immediately. Therefore, we need to verify that just
     * established connection really works.
     */
    /* Connection setup/verification is now done by doing redis command list query. Because we anyway
     * need to verify that Redis has required commands, we can now combine these two operations
     * (command list query and connection setup/verification). If either one of the functionalities
     * is not needed in the future and it is removed, remember to still leave the other one.
     */
    /* Non namespace-specific command list query can be used for connection setup purposes,
     * because SDL uses redisClusterAsyncCommandArgvWithKey which adds namespace to all
     * commands dispacthed.
     */

    /* If initial namespace was not given during dispatcher creation (multi namespace API),
     * verification is sent to hardcoded namespace. This works for verification purposes
     * because in our environment cluster is configured to operate only if all nodes
     * are working (so we can do verification to any node). However, this is not optimal
     * because we do not necessarily connect to such cluster node which will later be
     * used by client. Also our cluster configuration can change. This needs to be
     * optimized later (perhaps to connect to all nodes). */
    std::string nsForVerification;
    if (initialNamespace)
        nsForVerification = *initialNamespace;
    else
        nsForVerification = "namespace";

    dispatchAsync(std::bind(&AsyncHiredisClusterCommandDispatcher::verifyConnectionReply,
                            this,
                            std::placeholders::_1,
                            std::placeholders::_2),
                  nsForVerification,
                  contentsBuilder->build("COMMAND"),
                  false);
}

void AsyncHiredisClusterCommandDispatcher::verifyConnectionReply(const std::error_code& error, const redis::Reply& reply)
{
    if(error)
    {
        logger->error() << "AsyncHiredisClusterCommandDispatcher: connection verification failed: "
                        << error.message();
        armConnectionRetryTimer();
    }
    else
    {
        if (checkRedisModuleCommands(parseCommandListReply(reply)))
            setConnected();
        else
            SHAREDDATALAYER_ABORT("Required Redis module extension commands not available.");
    }
}

void AsyncHiredisClusterCommandDispatcher::waitConnectedAsync(const ConnectAck& connectAck)
{
    this->connectAck = connectAck;
    if (serviceState == ServiceState::CONNECTED)
        engine.postCallback(connectAck);
}

void AsyncHiredisClusterCommandDispatcher::registerDisconnectCb(const DisconnectCb& disconnectCb)
{
    disconnectCallback = disconnectCb;
}

void AsyncHiredisClusterCommandDispatcher::dispatchAsync(const CommandCb& commandCb,
                                                         const AsyncConnection::Namespace& ns,
                                                         const Contents& contents)
{
    dispatchAsync(commandCb, ns, contents, true);
}

void AsyncHiredisClusterCommandDispatcher::dispatchAsync(const CommandCb& commandCb,
                                                         const AsyncConnection::Namespace& ns,
                                                         const Contents& contents,
                                                         bool checkConnectionState)
{
    if (checkConnectionState && serviceState != ServiceState::CONNECTED)
    {
        engine.postCallback(std::bind(&AsyncHiredisClusterCommandDispatcher::callCommandCbWithError,
                                       this,
                                       commandCb,
                                       std::error_code(AsyncRedisCommandDispatcherErrorCode::NOT_CONNECTED)));
        return;
    }
    cbs.push_back(commandCb);
    std::vector<const char*> chars;
    std::transform(contents.stack.begin(), contents.stack.end(),
                   std::back_inserter(chars), [](const std::string& str){ return str.c_str(); });
    if (hiredisClusterSystem.redisClusterAsyncCommandArgvWithKey(acc, cb, &cbs.back(), ns.c_str(), static_cast<int>(ns.size()),
                                                                 static_cast<int>(contents.stack.size()), &chars[0],
                                                                 &contents.sizes[0]) != REDIS_OK)
    {
        removeCb(cbs.back());
        engine.postCallback(std::bind(&AsyncHiredisClusterCommandDispatcher::callCommandCbWithError,
                                       this,
                                       commandCb,
                                       getRedisError(acc->err, acc->errstr, nullptr)));
    }
}

void AsyncHiredisClusterCommandDispatcher::disableCommandCallbacks()
{
    clientCallbacksEnabled = false;
}

void AsyncHiredisClusterCommandDispatcher::callCommandCbWithError(const CommandCb& commandCb, const std::error_code& error)
{
    commandCb(error, AsyncRedisReply());
}

void AsyncHiredisClusterCommandDispatcher::setConnected()
{
    serviceState = ServiceState::CONNECTED;

    if (connectAck)
    {
        connectAck();
        connectAck = ConnectAck();
    }
}

void AsyncHiredisClusterCommandDispatcher::armConnectionRetryTimer()
{
    connectionRetryTimer.arm(connectionRetryTimerDuration,
                             [this] () { connect(); });

}

void AsyncHiredisClusterCommandDispatcher::handleReply(const CommandCb& commandCb,
                                                       const std::error_code& error,
                                                       const redisReply* rr)
{
    if (!isValidCb(commandCb))
        SHAREDDATALAYER_ABORT("Invalid callback function.");
    if (error)
        commandCb(error, AsyncRedisReply());
    else
        commandCb(error, AsyncRedisReply(*rr));
    if (!usePermanentCommandCallbacks)
        removeCb(commandCb);
}

bool AsyncHiredisClusterCommandDispatcher::isClientCallbacksEnabled() const
{
    return clientCallbacksEnabled;
}

bool AsyncHiredisClusterCommandDispatcher::isValidCb(const CommandCb& commandCb)
{
    for (auto i(cbs.begin()); i != cbs.end(); ++i)
        if (&*i == &commandCb)
            return true;
    return false;
}

void AsyncHiredisClusterCommandDispatcher::removeCb(const CommandCb& commandCb)
{
    for (auto i(cbs.begin()); i != cbs.end(); ++i)
        if (&*i == &commandCb)
        {
            cbs.erase(i);
            break;
        }
}

void AsyncHiredisClusterCommandDispatcher::handleDisconnect(const redisAsyncContext* ac)
{
    adapter->detach(ac);

    if (disconnectCallback)
        disconnectCallback();
}

void AsyncHiredisClusterCommandDispatcher::disconnectHiredisCluster()
{
    /* hiredis sometimes crashes if redisClusterAsyncFree is called without being connected (even
     * if acc is a valid pointer).
     */
    if (serviceState == ServiceState::CONNECTED)
        hiredisClusterSystem.redisClusterAsyncFree(acc);

    serviceState = ServiceState::DISCONNECTED;
}