CassandraMigrationBackend.hpp

#pragma once
 
#include "data/CassandraBackend.hpp"
#include "data/LedgerCacheInterface.hpp"
#include "data/cassandra/SettingsProvider.hpp"
#include "migration/cassandra/impl/CassandraMigrationSchema.hpp"
#include "migration/cassandra/impl/Spec.hpp"
#include "util/log/Logger.hpp"
 
#include <boost/asio/spawn.hpp>
 
#include <cstdint>
#include <string>
#include <utility>
 
namespace migration::cassandra {

class CassandraMigrationBackend : public data::cassandra::CassandraBackend {
    util::Logger log_{"Migration"};
    data::cassandra::SettingsProvider settingsProvider_;
    impl::CassandraMigrationSchema migrationSchema_;
 
public:
    explicit CassandraMigrationBackend(
        data::cassandra::SettingsProvider settingsProvider,
        data::LedgerCacheInterface& cache
    )
        : data::cassandra::CassandraBackend{auto{settingsProvider}, cache, false /* not readonly */}
        , settingsProvider_(std::move(settingsProvider))
        , migrationSchema_{settingsProvider_}
    {
    }

    template <impl::TableSpec TableDesc>
    void
    migrateInTokenRange(
        std::int64_t const start,
        std::int64_t const end,
        auto const& callback,
        boost::asio::yield_context yield
    )
    {
        LOG(log_.debug()) << "Travsering token range: " << start << " - " << end
                          << " ; table: " << TableDesc::kTableName;
        // for each table we only have one prepared statement
        static auto const kStatementPrepared = migrationSchema_.getPreparedFullScanStatement(
            handle_, TableDesc::kTableName, TableDesc::kPartitionKey
        );
 
        auto const statement = kStatementPrepared.bind(start, end);
 
        auto const res = this->executor_.read(yield, statement);
        if (not res) {
            LOG(log_.error()) << "Could not fetch data from table: " << TableDesc::kTableName
                              << " range: " << start << " - " << end << ";" << res.error();
            return;
        }
 
        auto const& results = res.value();
        if (not results.hasRows()) {
            LOG(log_.debug()) << "No rows returned  - table: " << TableDesc::kTableName
                              << " range: " << start << " - " << end;
            return;
        }
 
        for (auto const& row : std::apply(
                 [&](auto... args) { return data::cassandra::extract<decltype(args)...>(results); },
                 typename TableDesc::Row{}
             )) {
            callback(row);
        }
    }
};
}  // namespace migration::cassandra