#include <eosio/rodeos_plugin/cloner_plugin.hpp>

#include <eosio/rodeos_plugin/streams/stream.hpp>

#include <eosio/state_history/types.hpp>

#include <eosio/state_history/ship_client.hpp>

#include <b1/rodeos/rodeos.hpp>

#include <fc/log/logger.hpp>

#include <fc/log/logger_config.hpp>

#include <fc/io/json.hpp>

#include <fc/log/trace.hpp>

namespace b1 {

using namespace appbase;

using namespace eosio::ship_protocol;

namespace bpo = boost::program_options;

using rodeos::rodeos_db_partition;

using rodeos::rodeos_db_snapshot;

using rodeos::rodeos_filter;

struct cloner_session;

struct filter_ele {

std::string name;

std::string wasm;

uint32_t index = 0;

};

struct cloner_config {

bool exit_on_filter_wasm_error = false;

std::vector<filter_ele> filter_list = {};

bool profile = false;

bool undo_stack_enabled = false;

uint32_t force_write_stride = 0;

#ifdef EOSIO_EOS_VM_OC_RUNTIME_ENABLED

eosio::chain::eosvmoc::config eosvmoc_config;

#endif

};

struct cloner_plugin_impl : std::enable_shared_from_this<cloner_plugin_impl> {

std::shared_ptr<cloner_config> config = std::make_shared<cloner_config>();

std::shared_ptr<cloner_session> session;

std::shared_ptr<streamer_t> streamer;

cloner_plugin_impl() = default;

~cloner_plugin_impl();

void start();

};

namespace {

std::string to_string(const eosio::checksum256& cs) {

auto bytes = cs.extract_as_byte_array();

return fc::to_hex((const char*)bytes.data(), bytes.size());

}

} // namespace

struct cloner_session : std::enable_shared_from_this<cloner_session> {

cloner_plugin_impl* my = nullptr;

std::shared_ptr<cloner_config> config;

std::shared_ptr<chain_kv::database> db = app().find_plugin<rocksdb_plugin>()->get_db();

std::shared_ptr<rodeos_db_partition> partition =

std::make_shared<rodeos_db_partition>(db, std::vector<char>{}); // todo: prefix

std::optional<rodeos_db_snapshot> rodeos_snapshot;

bool reported_block = false;

struct filter_type {

std::unique_ptr<rodeos_filter> filter;

uint32_t index;

};

std::vector<filter_type> filters = {};

explicit cloner_session(cloner_plugin_impl* my) : my(my), config(my->config) {

// todo: remove

if (!config->filter_list.empty())

for (auto& filter: config->filter_list) {

#ifdef EOSIO_EOS_VM_OC_RUNTIME_ENABLED

bfs::path code_cache_dir = app().data_dir() / (filter.name + std::string{"_wasm"});

#endif

filters.emplace_back( filter_type { std::make_unique<rodeos_filter>(eosio::name{filter.name}, filter.wasm, config->profile

#ifdef EOSIO_EOS_VM_OC_RUNTIME_ENABLED

,

code_cache_dir, config->eosvmoc_config

#endif

),

filter.index });

}

ilog("number of filters: {n}", ("n", filters.size()));

}

void start() {

rodeos_snapshot.emplace(partition, true, config->undo_stack_enabled);

rodeos_snapshot->force_write_stride = config->force_write_stride;

ilog("cloner database status:");

ilog(" revisions: {f} - {r}",

("f", rodeos_snapshot->undo_stack->first_revision())("r", rodeos_snapshot->undo_stack->revision()));

ilog(" chain: {a}", ("a", eosio::convert_to_json(rodeos_snapshot->chain_id)));

ilog(" head: {a} {b}",

("a", rodeos_snapshot->head)("b", eosio::convert_to_json(rodeos_snapshot->head_id)));

ilog(" irreversible: {a} {b}",

("a", rodeos_snapshot->irreversible)("b", eosio::convert_to_json(rodeos_snapshot->irreversible_id)));

rodeos_snapshot->end_write(true);

}

std::vector<block_position> get_positions() {

std::vector<block_position> result;

if (rodeos_snapshot->head) {

rodeos::db_view_state view_state{ rodeos::state_account, *db, *rodeos_snapshot->write_session,

partition->contract_kv_prefix };

for (uint32_t i = rodeos_snapshot->irreversible; i <= rodeos_snapshot->head; ++i) {

auto info = rodeos::get_state_row<rodeos::block_info>(

view_state.kv_state.view, std::make_tuple(eosio::name{ "block.info" }, eosio::name{ "primary" }, i));

if (!info)

throw std::runtime_error("database is missing block.info for block " + std::to_string(i));

auto& info0 = std::get<rodeos::block_info_v0>(info->second);

result.push_back({ info0.num, info0.id });

}

}

return result;

}

static uint64_t to_trace_id(const eosio::checksum256& id) {

return fc::zipkin_span::to_id(fc::sha256{ reinterpret_cast<const char*>(id.extract_as_byte_array().data()), 32 });

}

template<typename GetBlocksResult>

bool process_received(const GetBlocksResult& result, std::vector<eosio::state_history::table_delta>&& deltas, eosio::input_stream bin) {

if (!result.this_block)

return true;

if (rodeos_snapshot->head && result.this_block->block_num > rodeos_snapshot->head + 1) {

std::string msg = "state-history plugin is missing block " + std::to_string(rodeos_snapshot->head + 1);

ilog(msg);

throw ship_client::retriable_failure(msg);

}

using namespace eosio::literals;

auto trace_id = to_trace_id(result.this_block->block_id);

auto token = fc::zipkin_span::token{ "rodeos"_n.value, trace_id };

auto blk_span = fc_create_span_from_token(token, "received");

fc_add_tag( blk_span, "block_id", to_string( result.this_block->block_id ) );

fc_add_tag( blk_span, "block_num", result.this_block->block_num );

rodeos_snapshot->start_block(result);

if (result.this_block->block_num <= rodeos_snapshot->head)

reported_block = false;

bool near = result.this_block->block_num + 4 >= result.last_irreversible.block_num;

bool write_now = !(result.this_block->block_num % 200) || near;

if (write_now || !reported_block) {

static uint64_t log_counter = 0;

if (log_counter++ % 1000 == 0) {

ilog("block {b} {i}", ( "b", result.this_block->block_num )

("i", result.this_block->block_num <= result.last_irreversible.block_num ? "irreversible" : ""));

} else {

dlog("block {b} {i}", ( "b", result.this_block->block_num )

("i", result.this_block->block_num <= result.last_irreversible.block_num ? "irreversible" : ""));

}

}

reported_block = true;

{

auto write_block_info_span = fc_create_span(blk_span, "write_block_info");

rodeos_snapshot->write_block_info(result);

}

{

auto write_deltas_span = fc_create_span(blk_span, "write_deltas");

rodeos_snapshot->write_deltas(result.this_block->block_num, std::move(deltas), [] { return app().is_quiting(); });

}

if (!filters.empty()) {

auto filter_span = fc_create_span(blk_span, "filter");

for (auto& filter: filters) {

if (my->streamer)

my->streamer->start_block(result.this_block->block_num, filter.index);

try {

filter.filter->process( *rodeos_snapshot, result, bin, [&]( const char* data, uint64_t data_size ) {

if( my->streamer ) {

my->streamer->stream_data( data, data_size, filter.index );

}

} );

} catch(...) {

handle_exception();

throw;

}

if (my->streamer)

my->streamer->stop_block(result.this_block->block_num, filter.index);

}

}

if( app().is_quiting() )

return false;

rodeos_snapshot->end_block(result, false, true);

{

auto end_block_span = fc_create_span(blk_span, "end_block");

rodeos_snapshot->end_block(result, false, true);

}

return true;

}

bool received( const get_status_result_v0& r, std::vector<eosio::state_history::table_delta>&& deltas, eosio::input_stream bin ) {

return false;

}

bool received( const get_blocks_result_v0& r, std::vector<eosio::state_history::table_delta>&& deltas, eosio::input_stream bin ) {

return process_received(r, std::move(deltas), bin);

}

bool received( const get_blocks_result_v1& r, std::vector<eosio::state_history::table_delta>&& deltas, eosio::input_stream bin ) {

return process_received(r, std::move(deltas), bin);

}

bool received( const get_blocks_result_v2& r, std::vector<eosio::state_history::table_delta>&& deltas, eosio::input_stream bin ) {

return process_received(r, std::move(deltas), bin);

}

// call on exception from rocksdb

void handle_exception() {

if( my ) {

rodeos_snapshot->end_write( true );

my->session.reset();

if( my->config->exit_on_filter_wasm_error ) {

appbase::app().quit();

}

}

}

~cloner_session() = default;

}; // cloner_session

static abstract_plugin& _cloner_plugin = app().register_plugin<cloner_plugin>();

cloner_plugin_impl::~cloner_plugin_impl() {

if (session)

session->my = nullptr;

}

void cloner_plugin_impl::start() {

session = std::make_shared<cloner_session>(this);

session->start();

}

cloner_plugin::cloner_plugin() : my(std::make_shared<cloner_plugin_impl>()) {}

cloner_plugin::~cloner_plugin() = default;

void cloner_plugin::set_program_options(options_description& cli, options_description& cfg) {

auto op = cfg.add_options();

op("clone-exit-on-filter-wasm-error", bpo::bool_switch()->default_value(false),

"Shutdown application if filter wasm throws an exception");

op("filter-name", bpo::value<std::string>(), "Filter name. Deprecated. Use filter-name-* instead");

op("filter-wasm", bpo::value<std::string>(), "Filter wasm. Deprecated. Use filter-wasm-* instead");

// Multiple filter contracts support

for (uint32_t i = 0; i < max_num_streamers; ++i) {

std::string i_str = std::to_string(i);

std::string name_str = std::string{"filter-name-"} + i_str;

std::string wasm_str = std::string{"filter-wasm-"} + i_str;

op(name_str.c_str(), bpo::value<std::string>(), "Filter name");

op(wasm_str.c_str(), bpo::value<std::string>(), "Filter wasm");

}

op("profile-filter", bpo::bool_switch(), "Enable filter profiling");

op("enable-undo-stack", bpo::value<bool>()->default_value(false), "Enable undo stack");

op("force-write-stride", bpo::value<uint32_t>()->default_value(10000),

"Maximum number of blocks to process before forcing rocksdb to flush. This option is primarily useful to control re-sync durations "

"under disaster recovery scenarios (when rodeos has unexpectedly exited, the option ensures blocks stored in rocksdb are at most "

"force-write-stride blocks behind the current head block being processed by rodeos. However, saving too frequently may affect performance. "

"It is likely that rocksdb itself will save rodeos data more frequently than this setting by flushing memtables to disk, based on various rocksdb "

"options. It is not recommended to set this to a small value in production use and should be instead used on a DR node. In contrast, when rodeos "

"exits normally, it saves the last block processed by rodeos into rocksdb and will continue processing "

"new blocks from that last processed block number when it next starts up.");

}

void cloner_plugin::plugin_initialize(const variables_map& options) {

try {

my->config->exit_on_filter_wasm_error = options["clone-exit-on-filter-wasm-error"].as<bool>();

// Old way, deprecated

if (options.count("filter-name") && options.count("filter-wasm")) {

my->config->filter_list.emplace_back(filter_ele{options["filter-name"].as<std::string>(), options["filter-wasm"].as<std::string>(), 0}); // index 0

} else if (options.count("filter-name") || options.count("filter-wasm")) {

throw std::runtime_error("filter-name and filter-wasm must be used together");

}

std::set<std::string> names {};

for (uint32_t i = 0; i < max_num_streamers; ++i) {

std::string i_str = std::to_string(i);

std::string name_str = std::string{"filter-name-"} + i_str;

std::string wasm_str = std::string{"filter-wasm-"} + i_str;

if ( options.count(name_str) && options.count(wasm_str) ) {

std::string name = options[name_str].as<std::string>();

std::string wasm = options[wasm_str].as<std::string>();

EOS_ASSERT(names.find(name) == names.end(), eosio::chain::plugin_exception, "Filter name " + name + " used multiple times");

EOS_ASSERT(my->config->filter_list.size() == 0 || i > 0, eosio::chain::plugin_exception, "legacy and multiple filter contracts cannot be mixed");

my->config->filter_list.emplace_back(filter_ele{name, wasm, i});

names.insert( name );

} else {

EOS_ASSERT( options.count( name_str ) == 0 && options.count( wasm_str ) == 0, eosio::chain::plugin_exception, name_str + " and " + wasm_str + " must be used together" );

}

}

my->config->profile = options["profile-filter"].as<bool>();

EOS_ASSERT(my->config->filter_list.size() <= max_num_streamers, eosio::chain::plugin_exception, "number of filter contracts: {num_names} greater than max_num_streamers: {max_num_streamers}", ("num_names", my->config->filter_list.size()) ("max_num_streamers", max_num_streamers));

ilog("number of filter contracts: {num_filters}", ("num_filters", my->config->filter_list.size()));

my->config->undo_stack_enabled = options["enable-undo-stack"].as<bool>();

#ifdef EOSIO_EOS_VM_OC_RUNTIME_ENABLED

// Added to options in chain_plugin

if (options.count("eos-vm-oc-cache-size-mb"))

my->config->eosvmoc_config.cache_size = options.at("eos-vm-oc-cache-size-mb").as<uint64_t>() * 1024u * 1024u;

if (options.count("eos-vm-oc-compile-threads"))

my->config->eosvmoc_config.threads = options.at("eos-vm-oc-compile-threads").as<uint64_t>();

if (options["eos-vm-oc-enable"].as<bool>())

my->config->eosvmoc_config.tierup = true;

my->config->eosvmoc_config.persistent = false;

#endif

my->config->force_write_stride = options["force-write-stride"].as<uint32_t>();

}

FC_LOG_AND_RETHROW()

}

void cloner_plugin::plugin_startup() {

handle_sighup();

my->start();

}

void cloner_plugin::plugin_shutdown() {

ilog("cloner_plugin stopped");

}

void cloner_plugin::handle_sighup() {

}

uint32_t cloner_plugin::get_snapshot_head() const {

if( my->session && my->session->rodeos_snapshot )

return my->session->rodeos_snapshot->head;

return 0;

}

void cloner_plugin::process(const std::vector<char>& data, std::vector<eosio::state_history::table_delta>&& deltas) {

if(!my->session) {

return;

}

eosio::input_stream bin{data.data(), data.data() + data.size()};

eosio::input_stream orig = bin;

eosio::ship_protocol::result res;

eosio::from_bin(res, bin);

std::visit([&](const auto& r) { return my->session->received(r, std::move(deltas), orig); }, res);

}

void cloner_plugin::handle_exception() {

if (!my) {

elog("try to handle exception for cloner session, but cloner instance does not exsit");

return;

}

if (!my->session) {

elog("try to handle exception for cloner session, but cloner session does not exsit");

return;

}

my->session->handle_exception();

}

void cloner_plugin::set_streamer(std::shared_ptr<streamer_t> streamer) {

my->streamer = std::move(streamer);

}

// Check every id in streamers' filter_ids is in my->config->filter_list

void cloner_plugin::validate_filter_ids(std::set<int>&& ids) {

for (auto& filter : my->config->filter_list) {

ids.erase(filter.index);

}

EOS_ASSERT(ids.empty(), eosio::chain::plugin_exception, "No filter contracts exist for streamers {id} ", ("id", ids));

}

} // namespace b1