// Build a ledger from consensus transactions
std::shared_ptr<Ledger>
buildLedger(
    std::shared_ptr<Ledger const> const& parent,
    NetClock::time_point closeTime,
    const bool closeTimeCorrect,
    NetClock::duration closeResolution,
    SHAMap const& txs,
    Application& app,
    CanonicalTXSet& retriableTxs,
    beast::Journal j)
{
    JLOG(j.debug()) << "Report: TxSt = " << txs.getHash().as_uint256()
                    << ", close " << closeTime.time_since_epoch().count()
                    << (closeTimeCorrect ? "" : " (incorrect)");

    return buildLedgerImpl(
        parent,
        closeTime,
        closeTimeCorrect,
        closeResolution,
        app,
        j,
        [&](OpenView& accum, std::shared_ptr<Ledger> const& buildLCL) {
            retriableTxs = applyTransactions(app, txs, accum, buildLCL, j);
        });
}

void Stoppable::stopRecursive (beast::Journal j)
{
    // Block on each child from the bottom of the tree up.
    //
    for (Children::const_iterator iter (m_children.cbegin ());
        iter != m_children.cend(); ++iter)
        iter->stoppable->stopRecursive (j);

    // if we get here then all children have stopped
    //
    m_childrenStopped = true;
    onChildrenStopped ();

    // Now block on this Stoppable.
    //
    bool const timedOut (! m_stoppedEvent.wait (1 * 1000)); // milliseconds
    if (timedOut)
    {
        if (auto stream = j.error())
            stream << "Waiting for '" << m_name << "' to stop";
        m_stoppedEvent.wait ();
    }

    // once we get here, we know the stoppable has stopped.
    m_stopped = true;
}

bool RootStoppable::stopAsync (beast::Journal j)
{
    bool alreadyCalled;
    {
        // Even though m_calledStop is atomic, we change its value under a
        // lock.  This removes a small timing window that occurs if the
        // waiting thread is handling a spurious wakeup while m_calledStop
        // changes state.
        std::unique_lock<std::mutex> lock (m_);
        alreadyCalled = m_calledStop.exchange (true);
    }
    if (alreadyCalled)
    {
        if (auto stream = j.warn())
            stream << "Stoppable::stop called again";
        return false;
    }

    // Wait until all in-flight JobQueue Jobs are completed.
    using namespace std::chrono_literals;
    jobCounter_.join (m_name.c_str(), 1s, j);

    c_.notify_all();
    stopAsyncRecursive(j);
    return true;
}

ApplyResult
applyTransaction (Application& app, OpenView& view,
    STTx const& txn,
        bool retryAssured, ApplyFlags flags,
            beast::Journal j)
{
    // Returns false if the transaction has need not be retried.
    if (retryAssured)
        flags = flags | tapRETRY;

    JLOG (j.debug()) << "TXN "
        << txn.getTransactionID ()
        //<< (engine.view().open() ? " open" : " closed")
        // because of the optional in engine
        << (retryAssured ? "/retry" : "/final");

    try
    {
        auto const result = apply(app,
            view, txn, flags, j);
        if (result.second)
        {
            JLOG (j.debug())
                << "Transaction applied: " << transHuman (result.first);
            return ApplyResult::Success;
        }

        if (isTefFailure (result.first) || isTemMalformed (result.first) ||
            isTelLocal (result.first))
        {
            // failure
            JLOG (j.debug())
                << "Transaction failure: " << transHuman (result.first);
            return ApplyResult::Fail;
        }

        JLOG (j.debug())
            << "Transaction retry: " << transHuman (result.first);
        return ApplyResult::Retry;
    }
    catch (std::exception const&)
    {
        JLOG (j.warn()) << "Throws";
        return ApplyResult::Fail;
    }
}

//------------------------------------------------------------------------------
bool Ledger::walkLedger (beast::Journal j) const
{
    std::vector <SHAMapMissingNode> missingNodes1;
    std::vector <SHAMapMissingNode> missingNodes2;

    if (stateMap_->getHash().isZero() &&
        ! info_.accountHash.isZero() &&
        ! stateMap_->fetchRoot (SHAMapHash{info_.accountHash}, nullptr))
    {
        missingNodes1.emplace_back (SHAMapType::STATE, SHAMapHash{info_.accountHash});
    }
    else
    {
        stateMap_->walkMap (missingNodes1, 32);
    }

    if (!missingNodes1.empty ())
    {
        if (auto stream = j.info())
        {
            stream << missingNodes1.size () << " missing account node(s)";
            stream << "First: " << missingNodes1[0];
        }
    }

    if (txMap_->getHash().isZero() &&
        info_.txHash.isNonZero() &&
        ! txMap_->fetchRoot (SHAMapHash{info_.txHash}, nullptr))
    {
        missingNodes2.emplace_back (SHAMapType::TRANSACTION, SHAMapHash{info_.txHash});
    }
    else
    {
        txMap_->walkMap (missingNodes2, 32);
    }

    if (!missingNodes2.empty ())
    {
        if (auto stream = j.info())
        {
            stream << missingNodes2.size () << " missing transaction node(s)";
            stream << "First: " << missingNodes2[0];
        }
    }
    return missingNodes1.empty () && missingNodes2.empty ();
}

Ledger::Ledger (
        LedgerInfo const& info,
        bool& loaded,
        Config const& config,
        Family& family,
        beast::Journal j)
    : mImmutable (true)
    , txMap_ (std::make_shared <SHAMap> (SHAMapType::TRANSACTION,
        info.txHash, family,
        SHAMap::version{getSHAMapV2(info) ? 2 : 1}))
    , stateMap_ (std::make_shared <SHAMap> (SHAMapType::STATE,
        info.accountHash, family,
        SHAMap::version{getSHAMapV2(info) ? 2 : 1}))
    , rules_ (config.features)
    , info_ (info)
{
    loaded = true;

    if (info_.txHash.isNonZero () &&
        !txMap_->fetchRoot (SHAMapHash{info_.txHash}, nullptr))
    {
        loaded = false;
        JLOG (j.warn()) << "Don't have TX root for ledger";
    }

    if (info_.accountHash.isNonZero () &&
        !stateMap_->fetchRoot (SHAMapHash{info_.accountHash}, nullptr))
    {
        loaded = false;
        JLOG (j.warn()) << "Don't have AS root for ledger";
    }

    txMap_->setImmutable ();
    stateMap_->setImmutable ();

    if (! setup(config))
        loaded = false;

    if (! loaded)
    {
        info_.hash = calculateLedgerHash(info_);
        family.missing_node (info_.hash);
    }
}

std::pair<std::vector<SignerEntries::SignerEntry>, TER>
SignerEntries::deserialize (
    STObject const& obj, beast::Journal journal, std::string const& annotation)
{
    std::pair<std::vector<SignerEntry>, TER> s;

    if (!obj.isFieldPresent (sfSignerEntries))
    {
        JLOG(journal.trace()) <<
            "Malformed " << annotation << ": Need signer entry array.";
        s.second = temMALFORMED;
        return s;
    }

    auto& accountVec = s.first;
    accountVec.reserve (STTx::maxMultiSigners);

    STArray const& sEntries (obj.getFieldArray (sfSignerEntries));
    for (STObject const& sEntry : sEntries)
    {
        // Validate the SignerEntry.
        if (sEntry.getFName () != sfSignerEntry)
        {
            JLOG(journal.trace()) <<
                "Malformed " << annotation << ": Expected SignerEntry.";
            s.second = temMALFORMED;
            return s;
        }

        // Extract SignerEntry fields.
        AccountID const account = sEntry.getAccountID (sfAccount);
        std::uint16_t const weight = sEntry.getFieldU16 (sfSignerWeight);
        accountVec.emplace_back (account, weight);
    }

    s.second = tesSUCCESS;
    return s;
}

std::shared_ptr<Ledger>
buildLedgerImpl(
    std::shared_ptr<Ledger const> const& parent,
    NetClock::time_point closeTime,
    const bool closeTimeCorrect,
    NetClock::duration closeResolution,
    Application& app,
    beast::Journal j,
    ApplyTxs&& applyTxs)
{
    auto buildLCL = std::make_shared<Ledger>(*parent, closeTime);

    if (buildLCL->rules().enabled(featureSHAMapV2) &&
        !buildLCL->stateMap().is_v2())
    {
        buildLCL->make_v2();
    }

    // Set up to write SHAMap changes to our database,
    //   perform updates, extract changes

    {
        OpenView accum(&*buildLCL);
        assert(!accum.open());
        applyTxs(accum, buildLCL);
        accum.apply(*buildLCL);
    }

    buildLCL->updateSkipList();

    {
        // Write the final version of all modified SHAMap
        // nodes to the node store to preserve the new LCL

        int const asf = buildLCL->stateMap().flushDirty(
            hotACCOUNT_NODE, buildLCL->info().seq);
        int const tmf = buildLCL->txMap().flushDirty(
            hotTRANSACTION_NODE, buildLCL->info().seq);
        JLOG(j.debug()) << "Flushed " << asf << " accounts and " << tmf
                        << " transaction nodes";
    }
    buildLCL->unshare();

    // Accept ledger
    buildLCL->setAccepted(
        closeTime, closeResolution, closeTimeCorrect, app.config());

    return buildLCL;
}

bool
adjustDescriptorLimit(int needed, beast::Journal j)
{
#ifdef RLIMIT_NOFILE
    // Get the current limit, then adjust it to what we need.
    struct rlimit rl;

    int available = 0;

    if (getrlimit(RLIMIT_NOFILE, &rl) == 0)
    {
        // If the limit is infinite, then we are good.
        if (rl.rlim_cur == RLIM_INFINITY)
            available = needed;
        else
            available = rl.rlim_cur;

        if (available < needed)
        {
            // Ignore the rlim_max, as the process may
            // be configured to override it anyways. We
            // ask for the number descriptors we need.
            rl.rlim_cur = needed;

            if (setrlimit(RLIMIT_NOFILE, &rl) == 0)
                available = rl.rlim_cur;
        }
    }

    if (needed > available)
    {
        j.fatal() <<
            "Insufficient number of file descriptors: " <<
            needed << " are needed, but only " <<
            available << " are available.";

        std::cerr <<
            "Insufficient number of file descriptors: " <<
            needed << " are needed, but only " <<
            available << " are available.\n";

        return false;
    }
#endif

    return true;
}

PreflightResult
preflight(Application& app, Rules const& rules,
    STTx const& tx, ApplyFlags flags,
        beast::Journal j)
{
    PreflightContext const pfctx(app, tx,
        rules, flags, j);
    try
    {
        return{ pfctx, invoke_preflight(pfctx) };
    }
    catch (std::exception const& e)
    {
        JLOG(j.fatal()) <<
            "apply: " << e.what();
        return{ pfctx, tefEXCEPTION };
    }
}

void RootStoppable::stop (beast::Journal j)
{
    // Must have a prior call to start()
    assert (m_started);

    {
        std::lock_guard<std::mutex> lock(m_);
        if (m_calledStop)
        {
            if (auto stream = j.warn())
                stream << "Stoppable::stop called again";
            return;
        }
        m_calledStop = true;
        c_.notify_all();
    }
    stopAsync (j);
    stopRecursive (j);
}

void Stoppable::stopAsyncRecursive (beast::Journal j)
{
    using namespace std::chrono;
    auto const start = high_resolution_clock::now();
    onStop ();
    auto const ms = duration_cast<milliseconds>(
        high_resolution_clock::now() - start).count();

#ifdef NDEBUG
    if (ms >= 10)
        if (auto stream = j.fatal())
            stream << m_name << "::onStop took " << ms << "ms";
#else
    (void)ms;
#endif

    for (Children::const_iterator iter (m_children.cbegin ());
        iter != m_children.cend(); ++iter)
        iter->stoppable->stopAsyncRecursive(j);
}

bool Ledger::assertSane (beast::Journal ledgerJ) const
{
    if (info_.hash.isNonZero () &&
            info_.accountHash.isNonZero () &&
            stateMap_ &&
            txMap_ &&
            (info_.accountHash == stateMap_->getHash ().as_uint256()) &&
            (info_.txHash == txMap_->getHash ().as_uint256()))
    {
        return true;
    }

    Json::Value j = getJson (*this);

    j [jss::accountTreeHash] = to_string (info_.accountHash);
    j [jss::transTreeHash] = to_string (info_.txHash);

    JLOG (ledgerJ.fatal()) << "ledger is not sane" << j;

    assert (false);

    return false;
}

void SHAMapNodeID::dump (beast::Journal journal) const
{
    JLOG(journal.debug()) <<
        getString ();
}

void HTTPReply (
    int nStatus, std::string const& content, Json::Output const& output, beast::Journal j)
{
    JLOG (j.trace())
        << "HTTP Reply " << nStatus << " " << content;

    if (nStatus == 401)
    {
        output ("HTTP/1.0 401 Authorization Required\r\n");
        output (getHTTPHeaderTimestamp ());

        // CHECKME this returns a different version than the replies below. Is
        //         this by design or an accident or should it be using
        //         BuildInfo::getFullVersionString () as well?
        output ("Server: " + systemName () + "-json-rpc/v1");
        output ("\r\n");

        // Be careful in modifying this! If you change the contents you MUST
        // update the Content-Length header as well to indicate the correct
        // size of the data.
        output ("WWW-Authenticate: Basic realm=\"jsonrpc\"\r\n"
                    "Content-Type: text/html\r\n"
                    "Content-Length: 296\r\n"
                    "\r\n"
                    "<!DOCTYPE HTML PUBLIC \"-//W3C//DTD HTML 4.01 "
                    "Transitional//EN\"\r\n"
                    "\"http://www.w3.org/TR/1999/REC-html401-19991224/loose.dtd"
                    "\">\r\n"
                    "<HTML>\r\n"
                    "<HEAD>\r\n"
                    "<TITLE>Error</TITLE>\r\n"
                    "<META HTTP-EQUIV='Content-Type' "
                    "CONTENT='text/html; charset=ISO-8859-1'>\r\n"
                    "</HEAD>\r\n"
                    "<BODY><H1>401 Unauthorized.</H1></BODY>\r\n");

        return;
    }

    switch (nStatus)
    {
    case 200: output ("HTTP/1.1 200 OK\r\n"); break;
    case 400: output ("HTTP/1.1 400 Bad Request\r\n"); break;
    case 403: output ("HTTP/1.1 403 Forbidden\r\n"); break;
    case 404: output ("HTTP/1.1 404 Not Found\r\n"); break;
    case 500: output ("HTTP/1.1 500 Internal Server Error\r\n"); break;
    case 503: output ("HTTP/1.1 503 Server is overloaded\r\n"); break;
    }

    output (getHTTPHeaderTimestamp ());

    output ("Connection: Keep-Alive\r\n"
            "Content-Length: ");

    // VFALCO TODO Determine if/when this header should be added
    //if (context.app.config().RPC_ALLOW_REMOTE)
    //    output ("Access-Control-Allow-Origin: *\r\n");

    output (std::to_string(content.size () + 2));
    output ("\r\n"
            "Content-Type: application/json; charset=UTF-8\r\n");

    output ("Server: " + systemName () + "-json-rpc/");
    output (BuildInfo::getFullVersionString ());
    output ("\r\n"
            "\r\n");
    output (content);
    output ("\r\n");
}

    void stop () override
    {
        stop_async ();

        JLOG(m_journal.debug()) << "Waiting to stop";
        std::unique_lock<std::mutex> lk{m_mut};
        m_cv.wait(lk, [this]{return m_asyncHandlersCompleted;});
        lk.unlock();
        JLOG(m_journal.debug()) << "Stopped";
    }

    /** Process a single ledger
        @param ledgerIndex The index of the ledger to process.
        @param ledgerHash  The known correct hash of the ledger.
        @param doNodes Ensure all ledger nodes are in the node db.
        @param doTxns Reprocess (account) transactions to SQL databases.
        @return `true` if the ledger was cleaned.
    */
    bool doLedger(
        LedgerIndex const& ledgerIndex,
        LedgerHash const& ledgerHash,
        bool doNodes,
        bool doTxns)
    {
        auto nodeLedger = app_.getInboundLedgers().acquire (
            ledgerHash, ledgerIndex, InboundLedger::fcGENERIC);
        if (!nodeLedger)
        {
            JLOG (j_.debug()) << "Ledger " << ledgerIndex << " not available";
            app_.getLedgerMaster().clearLedger (ledgerIndex);
            app_.getInboundLedgers().acquire(
                ledgerHash, ledgerIndex, InboundLedger::fcGENERIC);
            return false;
        }

        auto dbLedger = loadByIndex(ledgerIndex, app_);
        if (! dbLedger ||
            (dbLedger->info().hash != ledgerHash) ||
            (dbLedger->info().parentHash != nodeLedger->info().parentHash))
        {
            // Ideally we'd also check for more than one ledger with that index
            JLOG (j_.debug()) <<
                "Ledger " << ledgerIndex << " mismatches SQL DB";
            doTxns = true;
        }

        if(! app_.getLedgerMaster().fixIndex(ledgerIndex, ledgerHash))
        {
            JLOG (j_.debug()) << "ledger " << ledgerIndex
                            << " had wrong entry in history";
            doTxns = true;
        }

        if (doNodes && !nodeLedger->walkLedger(app_.journal ("Ledger")))
        {
            JLOG (j_.debug()) << "Ledger " << ledgerIndex << " is missing nodes";
            app_.getLedgerMaster().clearLedger (ledgerIndex);
            app_.getInboundLedgers().acquire(
                ledgerHash, ledgerIndex, InboundLedger::fcGENERIC);
            return false;
        }

        if (doTxns && !pendSaveValidated(app_, nodeLedger, true, false))
        {
            JLOG (j_.debug()) << "Failed to save ledger " << ledgerIndex;
            return false;
        }

        return true;
    }

    /** Returns the hash of the specified ledger.
        @param ledgerIndex The index of the desired ledger.
        @param referenceLedger [out] An optional known good subsequent ledger.
        @return The hash of the ledger. This will be all-bits-zero if not found.
    */
    LedgerHash getHash(
        LedgerIndex const& ledgerIndex,
        std::shared_ptr<ReadView const>& referenceLedger)
    {
        LedgerHash ledgerHash;

        if (!referenceLedger || (referenceLedger->info().seq < ledgerIndex))
        {
            referenceLedger = app_.getLedgerMaster().getValidatedLedger();
            if (!referenceLedger)
            {
                JLOG (j_.warn()) << "No validated ledger";
                return ledgerHash; // Nothing we can do. No validated ledger.
            }
        }

        if (referenceLedger->info().seq >= ledgerIndex)
        {
            // See if the hash for the ledger we need is in the reference ledger
            ledgerHash = getLedgerHash(referenceLedger, ledgerIndex);
            if (ledgerHash.isZero())
            {
                // No. Try to get another ledger that might have the hash we
                // need: compute the index and hash of a ledger that will have
                // the hash we need.
                LedgerIndex refIndex = getCandidateLedger (ledgerIndex);
                LedgerHash refHash = getLedgerHash (referenceLedger, refIndex);

                bool const nonzero (refHash.isNonZero ());
                assert (nonzero);
                if (nonzero)
                {
                    // We found the hash and sequence of a better reference
                    // ledger.
                    referenceLedger =
                        app_.getInboundLedgers().acquire(
                            refHash, refIndex, InboundLedger::fcGENERIC);
                    if (referenceLedger)
                        ledgerHash = getLedgerHash(
                            referenceLedger, ledgerIndex);
                }
            }
        }
        else
            JLOG (j_.warn()) << "Validated ledger is prior to target ledger";

        return ledgerHash;
    }

    std::shared_ptr<Ledger const>
    acquire(uint256 const& hash, std::uint32_t seq,
        InboundLedger::Reason reason) override
    {
        assert(hash.isNonZero());
        assert(reason != InboundLedger::Reason::SHARD ||
            (seq != 0 && app_.getShardStore()));
        if (isStopping())
            return {};

        bool isNew = true;
        std::shared_ptr<InboundLedger> inbound;
        {
            ScopedLockType sl(mLock);
            auto it = mLedgers.find(hash);
            if (it != mLedgers.end())
            {
                isNew = false;
                inbound = it->second;
            }
            else
            {
                inbound = std::make_shared <InboundLedger>(
                    app_, hash, seq, reason, std::ref(m_clock));
                mLedgers.emplace(hash, inbound);
                inbound->init(sl);
                ++mCounter;
            }
        }

        if (inbound->isFailed())
            return {};

        if (! isNew)
            inbound->update(seq);

        if (! inbound->isComplete())
            return {};

        if (reason == InboundLedger::Reason::HISTORY)
        {
            if (inbound->getLedger()->stateMap().family().isShardBacked())
                app_.getNodeStore().copyLedger(inbound->getLedger());
        }
        else if (reason == InboundLedger::Reason::SHARD)
        {
            auto shardStore = app_.getShardStore();
            if (!shardStore)
            {
                JLOG(j_.error()) <<
                    "Acquiring shard with no shard store available";
                return {};
            }
            if (inbound->getLedger()->stateMap().family().isShardBacked())
                shardStore->setStored(inbound->getLedger());
            else
                shardStore->copyLedger(inbound->getLedger());
        }
        return inbound->getLedger();
    }

    void runImpl ()
    {
        beast::Thread::setCurrentThreadName ("LedgerCleaner");
        JLOG (j_.debug()) << "Started";

        init();

        while (true)
        {
            {
                std::unique_lock<std::mutex> lock (mutex_);
                wakeup_.wait(lock, [this]()
                    {
                        return (
                            shouldExit_ ||
                            state_ == State::startCleaning);
                    });
                if (shouldExit_)
                    break;

                state_ = State::cleaning;
            }
            doLedgerCleaner();
        }

        stopped();
    }