KeyspaceBackend.hpp

#pragma once
 
#include "data/LedgerHeaderCache.hpp"
#include "data/Types.hpp"
#include "data/cassandra/CassandraBackendFamily.hpp"
#include "data/cassandra/Concepts.hpp"
#include "data/cassandra/KeyspaceSchema.hpp"
#include "data/cassandra/SettingsProvider.hpp"
#include "data/cassandra/Types.hpp"
#include "data/cassandra/impl/ExecutionStrategy.hpp"
#include "util/Assert.hpp"
#include "util/log/Logger.hpp"
 
#include <boost/asio/spawn.hpp>
#include <boost/json/object.hpp>
#include <boost/uuid/string_generator.hpp>
#include <boost/uuid/uuid.hpp>
#include <cassandra.h>
#include <fmt/format.h>
#include <xrpl/basics/Blob.h>
#include <xrpl/basics/base_uint.h>
#include <xrpl/basics/strHex.h>
#include <xrpl/protocol/AccountID.h>
#include <xrpl/protocol/Indexes.h>
#include <xrpl/protocol/LedgerHeader.h>
#include <xrpl/protocol/nft.h>
 
#include <cstddef>
#include <cstdint>
#include <iterator>
#include <optional>
#include <stdexcept>
#include <utility>
#include <vector>
 
namespace data::cassandra {

template <
    SomeSettingsProvider SettingsProviderType,
    SomeExecutionStrategy ExecutionStrategyType,
    typename FetchLedgerCacheType = FetchLedgerCache>
class BasicKeyspaceBackend : public CassandraBackendFamily<
                                 SettingsProviderType,
                                 ExecutionStrategyType,
                                 KeyspaceSchema<SettingsProviderType>,
                                 FetchLedgerCacheType> {
    using DefaultCassandraFamily = CassandraBackendFamily<
        SettingsProviderType,
        ExecutionStrategyType,
        KeyspaceSchema<SettingsProviderType>,
        FetchLedgerCacheType>;
 
    using DefaultCassandraFamily::executor_;
    using DefaultCassandraFamily::ledgerSequence_;
    using DefaultCassandraFamily::log_;
    using DefaultCassandraFamily::range_;
    using DefaultCassandraFamily::schema_;
 
public:
    using DefaultCassandraFamily::DefaultCassandraFamily;

    BasicKeyspaceBackend(BasicKeyspaceBackend&&) = delete;
 
    bool
    doFinishWrites() override
    {
        this->waitForWritesToFinish();
 
        // !range_.has_value() means the table 'ledger_range' is not populated;
        // This would be the first write to the table.
        // In this case, insert both min_sequence/max_sequence range into the table.
        if (not range_.has_value()) {
            executor_.writeSync(
                schema_->insertLedgerRange, /* isLatestLedger =*/false, ledgerSequence_
            );
            executor_.writeSync(
                schema_->insertLedgerRange, /* isLatestLedger =*/true, ledgerSequence_
            );
        }
 
        if (not this->executeSyncUpdate(
                schema_->updateLedgerRange.bind(ledgerSequence_, true, ledgerSequence_ - 1)
            )) {
            log_.warn() << "Update failed for ledger " << ledgerSequence_;
            return false;
        }
 
        log_.info() << "Committed ledger " << ledgerSequence_;
        return true;
    }
 
    NFTsAndCursor
    fetchNFTsByIssuer(
        ripple::AccountID const& issuer,
        std::optional<std::uint32_t> const& taxon,
        std::uint32_t const ledgerSequence,
        std::uint32_t const limit,
        std::optional<ripple::uint256> const& cursorIn,
        boost::asio::yield_context yield
    ) const override
    {
        std::vector<ripple::uint256> nftIDs;
        if (taxon.has_value()) {
            // Keyspace and ScyllaDB uses the same logic for taxon-filtered queries
            nftIDs = fetchNFTIDsByTaxon(issuer, *taxon, limit, cursorIn, yield);
        } else {
            // Amazon Keyspaces Workflow for non-taxon queries
            auto const startTaxon =
                cursorIn.has_value() ? ripple::nft::toUInt32(ripple::nft::getTaxon(*cursorIn)) : 0;
            auto const startTokenID = cursorIn.value_or(ripple::uint256(0));
 
            Statement const firstQuery = schema_->selectNFTIDsByIssuerTaxon.bind(issuer);
            firstQuery.bindAt(1, startTaxon);
            firstQuery.bindAt(2, startTokenID);
            firstQuery.bindAt(3, Limit{limit});
 
            auto const firstRes = executor_.read(yield, firstQuery);
            if (firstRes.has_value()) {
                for (auto const [nftID] : extract<ripple::uint256>(*firstRes))
                    nftIDs.push_back(nftID);
            }
 
            if (nftIDs.size() < limit) {
                auto const remainingLimit = limit - nftIDs.size();
                Statement const secondQuery = schema_->selectNFTsAfterTaxonKeyspaces.bind(issuer);
                secondQuery.bindAt(1, startTaxon);
                secondQuery.bindAt(2, Limit{remainingLimit});
 
                auto const secondRes = executor_.read(yield, secondQuery);
                if (secondRes.has_value()) {
                    for (auto const [nftID] : extract<ripple::uint256>(*secondRes))
                        nftIDs.push_back(nftID);
                }
            }
        }
        return populateNFTsAndCreateCursor(nftIDs, ledgerSequence, limit, yield);
    }

    std::vector<ripple::uint256>
    fetchAccountRoots(
        [[maybe_unused]] std::uint32_t number,
        [[maybe_unused]] std::uint32_t pageSize,
        [[maybe_unused]] std::uint32_t seq,
        [[maybe_unused]] boost::asio::yield_context yield
    ) const override
    {
        ASSERT(false, "Fetching account roots is not supported by the Keyspaces backend.");
        std::unreachable();
    }
 
private:
    std::vector<ripple::uint256>
    fetchNFTIDsByTaxon(
        ripple::AccountID const& issuer,
        std::uint32_t const taxon,
        std::uint32_t const limit,
        std::optional<ripple::uint256> const& cursorIn,
        boost::asio::yield_context yield
    ) const
    {
        std::vector<ripple::uint256> nftIDs;
        Statement const statement = schema_->selectNFTIDsByIssuerTaxon.bind(issuer);
        statement.bindAt(1, taxon);
        statement.bindAt(2, cursorIn.value_or(ripple::uint256(0)));
        statement.bindAt(3, Limit{limit});
 
        auto const res = executor_.read(yield, statement);
        if (res.has_value() && res->hasRows()) {
            for (auto const [nftID] : extract<ripple::uint256>(*res))
                nftIDs.push_back(nftID);
        }
        return nftIDs;
    }
 
    std::vector<ripple::uint256>
    fetchNFTIDsWithoutTaxon(
        ripple::AccountID const& issuer,
        std::uint32_t const limit,
        std::optional<ripple::uint256> const& cursorIn,
        boost::asio::yield_context yield
    ) const
    {
        std::vector<ripple::uint256> nftIDs;
 
        auto const startTaxon =
            cursorIn.has_value() ? ripple::nft::toUInt32(ripple::nft::getTaxon(*cursorIn)) : 0;
        auto const startTokenID = cursorIn.value_or(ripple::uint256(0));
 
        Statement firstQuery = schema_->selectNFTIDsByIssuerTaxon.bind(issuer);
        firstQuery.bindAt(1, startTaxon);
        firstQuery.bindAt(2, startTokenID);
        firstQuery.bindAt(3, Limit{limit});
 
        auto const firstRes = executor_.read(yield, firstQuery);
        if (firstRes.has_value()) {
            for (auto const [nftID] : extract<ripple::uint256>(*firstRes))
                nftIDs.push_back(nftID);
        }
 
        if (nftIDs.size() < limit) {
            auto const remainingLimit = limit - nftIDs.size();
            Statement secondQuery = schema_->selectNFTsAfterTaxonKeyspaces.bind(issuer);
            secondQuery.bindAt(1, startTaxon);
            secondQuery.bindAt(2, Limit{remainingLimit});
 
            auto const secondRes = executor_.read(yield, secondQuery);
            if (secondRes.has_value()) {
                for (auto const [nftID] : extract<ripple::uint256>(*secondRes))
                    nftIDs.push_back(nftID);
            }
        }
        return nftIDs;
    }

    NFTsAndCursor
    populateNFTsAndCreateCursor(
        std::vector<ripple::uint256> const& nftIDs,
        std::uint32_t const ledgerSequence,
        std::uint32_t const limit,
        boost::asio::yield_context yield
    ) const
    {
        if (nftIDs.empty()) {
            LOG(log_.debug()) << "No rows returned";
            return {};
        }
 
        NFTsAndCursor ret;
        if (nftIDs.size() == limit)
            ret.cursor = nftIDs.back();
 
        // Prepare and execute queries to fetch NFT info and URIs in parallel.
        std::vector<Statement> selectNFTStatements;
        selectNFTStatements.reserve(nftIDs.size());
        std::transform(
            std::cbegin(nftIDs),
            std::cend(nftIDs),
            std::back_inserter(selectNFTStatements),
            [&](auto const& nftID) { return schema_->selectNFT.bind(nftID, ledgerSequence); }
        );
 
        std::vector<Statement> selectNFTURIStatements;
        selectNFTURIStatements.reserve(nftIDs.size());
        std::transform(
            std::cbegin(nftIDs),
            std::cend(nftIDs),
            std::back_inserter(selectNFTURIStatements),
            [&](auto const& nftID) { return schema_->selectNFTURI.bind(nftID, ledgerSequence); }
        );
 
        auto const nftInfos = executor_.readEach(yield, selectNFTStatements);
        auto const nftUris = executor_.readEach(yield, selectNFTURIStatements);
 
        // Combine the results into final NFT objects.
        for (auto i = 0u; i < nftIDs.size(); ++i) {
            if (auto const maybeRow = nftInfos[i].template get<uint32_t, ripple::AccountID, bool>();
                maybeRow.has_value()) {
                auto [seq, owner, isBurned] = *maybeRow;
                NFT nft(nftIDs[i], seq, owner, isBurned);
                if (auto const maybeUri = nftUris[i].template get<ripple::Blob>();
                    maybeUri.has_value())
                    nft.uri = *maybeUri;
                ret.nfts.push_back(nft);
            }
        }
        return ret;
    }
};
 
using KeyspaceBackend = BasicKeyspaceBackend<SettingsProvider, impl::DefaultExecutionStrategy<>>;
 
}  // namespace data::cassandra