#include <xrpl/nodestore/Database.h>

#include <xrpl/basics/Log.h>
#include <xrpl/basics/base_uint.h>
#include <xrpl/basics/contract.h>
#include <xrpl/beast/core/CurrentThreadName.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/config/BasicConfig.h>
#include <xrpl/config/Constants.h>
#include <xrpl/json/json_forwards.h>
#include <xrpl/json/json_value.h>
#include <xrpl/nodestore/Backend.h>
#include <xrpl/nodestore/NodeObject.h>
#include <xrpl/nodestore/Scheduler.h>
#include <xrpl/nodestore/Types.h>
#include <xrpl/protocol/SystemParameters.h>
#include <xrpl/protocol/jss.h>

#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstdint>
#include <exception>
#include <functional>
#include <memory>
#include <mutex>
#include <stdexcept>
#include <string>
#include <thread>
#include <utility>

namespace xrpl::NodeStore {

Database::Database(
    Scheduler& scheduler,
    int readThreads,
    Section const& config,
    beast::Journal journal)
    : j_(journal)
    , scheduler_(scheduler)
    , earliestLedgerSeq_(get<std::uint32_t>(config, Keys::kEarliestSeq, kXrpLedgerEarliestSeq))
    , requestBundle_(get<int>(config, Keys::kRqBundle, 4))
    , readThreads_(std::max(1, readThreads))
{
    XRPL_ASSERT(readThreads, "xrpl::NodeStore::Database::Database : nonzero threads input");

    if (earliestLedgerSeq_ < 1)
        Throw<std::runtime_error>("Invalid earliest_seq");

    if (requestBundle_ < 1 || requestBundle_ > 64)
        Throw<std::runtime_error>("Invalid rq_bundle");

    for (int i = readThreads_.load(); i != 0; --i)
    {
        std::thread t(
            [this](int i) {
                runningThreads_++;

                beast::setCurrentThreadName("db prefetch #" + std::to_string(i));

                decltype(read_) read;

                while (true)
                {
                    {
                        std::unique_lock<std::mutex> lock(readLock_);

                        if (isStopping())
                            break;

                        if (read_.empty())
                        {
                            runningThreads_--;
                            readCondVar_.wait(lock);
                            runningThreads_++;
                        }

                        if (isStopping())
                            break;

                        // extract multiple object at a time to minimize the
                        // overhead of acquiring the mutex.
                        for (int cnt = 0; !read_.empty() && cnt != requestBundle_; ++cnt)
                            read.insert(read_.extract(read_.begin()));
                    }

                    for (auto it = read.begin(); it != read.end(); ++it)
                    {
                        XRPL_ASSERT(
                            !it->second.empty(),
                            "xrpl::NodeStore::Database::Database : non-empty "
                            "data");

                        auto const& hash = it->first;
                        auto const& data = it->second;
                        auto const seqn = data[0].first;

                        auto obj = fetchNodeObject(hash, seqn, FetchType::Async);

                        // This could be further optimized: if there are
                        // multiple requests for sequence numbers mapping to
                        // multiple databases by sorting requests such that all
                        // indices mapping to the same database are grouped
                        // together and serviced by a single read.
                        for (auto const& req : data)
                        {
                            req.second(
                                (seqn == req.first) || isSameDB(req.first, seqn)
                                    ? obj
                                    : fetchNodeObject(hash, req.first, FetchType::Async));
                        }
                    }

                    read.clear();
                }

                --runningThreads_;
                --readThreads_;
            },
            i);
        t.detach();
    }
}

Database::~Database()
{
    // NOTE!
    // Any derived class should call the stop() method in its
    // destructor.  Otherwise, occasionally, the derived class may
    // crash during shutdown when its members are accessed by one of
    // these threads after the derived class is destroyed but before
    // this base class is destroyed.
    stop();
}

bool
Database::isStopping() const
{
    return readStopping_.load(std::memory_order_relaxed);
}

void
Database::stop()
{
    {
        std::scoped_lock const lock(readLock_);

        if (!readStopping_.exchange(true, std::memory_order_relaxed))
        {
            JLOG(j_.debug()) << "Clearing read queue because of stop request";
            read_.clear();
            readCondVar_.notify_all();
        }
    }

    JLOG(j_.debug()) << "Waiting for stop request to complete...";

    using namespace std::chrono;

    auto const start = steady_clock::now();

    while (readThreads_.load() != 0)
    {
        XRPL_ASSERT(
            steady_clock::now() - start < 30s,
            "xrpl::NodeStore::Database::stop : maximum stop duration");
        std::this_thread::yield();
    }

    JLOG(j_.debug())
        << "Stop request completed in "
        << duration_cast<std::chrono::milliseconds>(steady_clock::now() - start).count()
        << " milliseconds";
}

void
Database::asyncFetch(
    uint256 const& hash,
    std::uint32_t ledgerSeq,
    std::function<void(std::shared_ptr<NodeObject> const&)>&& cb)
{
    std::scoped_lock const lock(readLock_);

    if (!isStopping())
    {
        read_[hash].emplace_back(ledgerSeq, std::move(cb));
        readCondVar_.notify_one();
    }
}

void
Database::importInternal(Backend& dstBackend, Database& srcDB)
{
    Batch batch;
    batch.reserve(kBatchWritePreallocationSize);
    auto storeBatch = [&, fname = __func__]() {
        try
        {
            dstBackend.storeBatch(batch);
        }
        catch (std::exception const& e)
        {
            JLOG(j_.error()) << "Exception caught in function " << fname << ". Error: " << e.what();
            return;
        }

        std::uint64_t sz{0};
        for (auto const& nodeObject : batch)
            sz += nodeObject->getData().size();
        storeStats(batch.size(), sz);
        batch.clear();
    };

    srcDB.forEach([&](std::shared_ptr<NodeObject> nodeObject) {
        XRPL_ASSERT(nodeObject, "xrpl::NodeStore::Database::importInternal : non-null node");
        if (!nodeObject)  // This should never happen
            return;

        batch.emplace_back(std::move(nodeObject));
        if (batch.size() >= kBatchWritePreallocationSize)
            storeBatch();
    });

    if (!batch.empty())
        storeBatch();
}

// Perform a fetch and report the time it took
std::shared_ptr<NodeObject>
Database::fetchNodeObject(
    uint256 const& hash,
    std::uint32_t ledgerSeq,
    FetchType fetchType,
    bool duplicate)
{
    FetchReport fetchReport(fetchType);

    using namespace std::chrono;
    auto const begin{steady_clock::now()};

    auto nodeObject{fetchNodeObject(hash, ledgerSeq, fetchReport, duplicate)};
    auto dur = steady_clock::now() - begin;
    fetchDurationUs_ += duration_cast<microseconds>(dur).count();
    if (nodeObject)
    {
        ++fetchHitCount_;
        fetchSz_ += nodeObject->getData().size();
    }
    ++fetchTotalCount_;

    fetchReport.elapsed = duration_cast<milliseconds>(dur);
    scheduler_.onFetch(fetchReport);
    return nodeObject;
}

void
Database::getCountsJson(json::Value& obj)
{
    XRPL_ASSERT(obj.isObject(), "xrpl::NodeStore::Database::getCountsJson : valid input type");

    {
        std::unique_lock<std::mutex> const lock(readLock_);
        obj["read_queue"] = static_cast<json::UInt>(read_.size());
    }

    obj["read_threads_total"] = readThreads_.load();
    obj["read_threads_running"] = runningThreads_.load();
    obj["read_request_bundle"] = requestBundle_;

    obj[jss::node_writes] = std::to_string(storeCount_);
    obj[jss::node_reads_total] = std::to_string(fetchTotalCount_);
    obj[jss::node_reads_hit] = std::to_string(fetchHitCount_);
    obj[jss::node_written_bytes] = std::to_string(storeSz_);
    obj[jss::node_read_bytes] = std::to_string(fetchSz_);
    obj[jss::node_reads_duration_us] = std::to_string(fetchDurationUs_);
}

}  // namespace xrpl::NodeStore