feat: Migration framework

src/CMakeLists.txt

@@ -5,5 +5,6 @@ add_subdirectory(etlng)
 add_subdirectory(feed)
 add_subdirectory(rpc)
 add_subdirectory(web)
+add_subdirectory(migration)
 add_subdirectory(app)
 add_subdirectory(main)

src/app/CMakeLists.txt

@@ -1,4 +1,4 @@
 add_library(clio_app)
 target_sources(clio_app PRIVATE CliArgs.cpp ClioApplication.cpp WebHandlers.cpp)
 
-target_link_libraries(clio_app PUBLIC clio_etl clio_etlng clio_feed clio_web clio_rpc)
+target_link_libraries(clio_app PUBLIC clio_etl clio_etlng clio_feed clio_web clio_rpc clio_migration)

src/app/CliArgs.cpp

@@ -19,6 +19,7 @@
 
 #include "app/CliArgs.hpp"
 
+#include "migration/MigrationApplication.hpp"
 #include "util/build/Build.hpp"
 
 #include <boost/program_options/options_description.hpp>
@@ -45,6 +46,7 @@
         ("version,v", "print version and exit")
         ("conf,c", po::value<std::string>()->default_value(defaultConfigPath), "configuration file")
         ("ng-web-server,w", "Use ng-web-server")
+        ("migrate", po::value<std::string>(),"start migration helper")
     ;
     // clang-format on
     po::positional_options_description positional;
@@ -65,6 +67,14 @@
     }
 
     auto configPath = parsed["conf"].as<std::string>();
+
+    if (parsed.count("migrate") != 0u) {
+        auto const opt = parsed["migrate"].as<std::string>();
+        if (opt == "status")
+            return Action{Action::Migrate{std::move(configPath), MigrateSubCmd::status()}};
+        return Action{Action::Migrate{std::move(configPath), MigrateSubCmd::migration(opt)}};
+    }
+
     return Action{Action::Run{.configPath = std::move(configPath), .useNgWebServer = parsed.count("ng-web-server") != 0}
     };
 }

src/app/CliArgs.hpp

@@ -19,6 +19,7 @@
 
 #pragma once
 
+#include "migration/MigrationApplication.hpp"
 #include "util/OverloadSet.hpp"
 
 #include <string>
@@ -52,13 +53,20 @@ class CliArgs {
             int exitCode;  ///< Exit code.
         };
 
+        /** @brief Migration action. */
+        struct Migrate {
+            std::string configPath;
+            MigrateSubCmd subCmd;
+        };
+
         /**
          * @brief Construct an action from a Run.
          *
          * @param action Run action.
          */
         template <typename ActionType>
-            requires std::is_same_v<ActionType, Run> or std::is_same_v<ActionType, Exit>
+            requires std::is_same_v<ActionType, Run> or std::is_same_v<ActionType, Exit> or
+            std::is_same_v<ActionType, Migrate>
         explicit Action(ActionType&& action) : action_(std::forward<ActionType>(action))
         {
         }
@@ -78,7 +86,7 @@ class CliArgs {
         }
 
     private:
-        std::variant<Run, Exit> action_;
+        std::variant<Run, Exit, Migrate> action_;
     };
 
     /**

src/data/BackendInterface.hpp

@@ -548,6 +548,16 @@ class BackendInterface {
         boost::asio::yield_context yield
     ) const;
 
+    /**
+     * @brief Fetches the status of migrator by name.
+     *
+     * @param migratorName The name of the migrator
+     * @param yield The coroutine context
+     * @return The status of the migrator if found; nullopt otherwise
+     */
+    virtual std::optional<std::string>
+    fetchMigratorStatus(std::string const& migratorName, boost::asio::yield_context yield) const = 0;
+
     /**
      * @brief Synchronously fetches the ledger range from DB.
      *
@@ -673,6 +683,15 @@ class BackendInterface {
     bool
     finishWrites(std::uint32_t ledgerSequence);
 
+    /**
+     *@brief Mark the migration status of a migrator as Migrated in the database
+     *
+     *@param migratorName The name of the migrator
+     *@param status The status to set
+     */
+    virtual void
+    writeMigratorStatus(std::string const& migratorName, std::string const& status) = 0;
+
     /**
      * @return true if database is overwhelmed; false otherwise
      */

src/data/CassandraBackend.hpp

@@ -53,6 +53,7 @@
 #include <stdexcept>
 #include <string>
 #include <tuple>
+#include <unordered_set>
 #include <utility>
 #include <vector>
 
@@ -72,13 +73,15 @@ class BasicCassandraBackend : public BackendInterface {
 
     SettingsProviderType settingsProvider_;
     Schema<SettingsProviderType> schema_;
+
+    std::atomic_uint32_t ledgerSequence_ = 0u;
+
+protected:
     Handle handle_;
 
     // have to be mutable because BackendInterface constness :(
     mutable ExecutionStrategyType executor_;
 
-    std::atomic_uint32_t ledgerSequence_ = 0u;
-
 public:
     /**
      * @brief Create a new cassandra/scylla backend instance.
@@ -835,6 +838,26 @@ class BasicCassandraBackend : public BackendInterface {
         return results;
     }
 
+    std::optional<std::string>
+    fetchMigratorStatus(std::string const& migratorName, boost::asio::yield_context yield) const override
+    {
+        auto const res = executor_.read(yield, schema_->selectMigratorStatus, Text(migratorName));
+        if (not res) {
+            LOG(log_.error()) << "Could not fetch migrator status: " << res.error();
+            return {};
+        }
+
+        auto const& results = res.value();
+        if (not results) {
+            return {};
+        }
+
+        for (auto [statusString] : extract<std::string>(results))
+            return statusString;
+
+        return {};
+    }
+
     void
     doWriteLedgerObject(std::string&& key, std::uint32_t const seq, std::string&& blob) override
     {
@@ -962,6 +985,14 @@ class BasicCassandraBackend : public BackendInterface {
         // probably was used in PG to start a transaction or smth.
     }
 
+    void
+    writeMigratorStatus(std::string const& migratorName, std::string const& status) override
+    {
+        executor_.writeSync(
+            schema_->insertMigratorStatus, data::cassandra::Text{migratorName}, data::cassandra::Text(status)
+        );
+    }
+
     bool
     isTooBusy() const override
     {

src/data/README.md

@@ -262,3 +262,15 @@ CREATE TABLE clio.nf_token_transactions (
 ```
 
 The `nf_token_transactions` table serves as the NFT counterpart to `account_tx`, inspired by the same motivations and fulfilling a similar role within this context. It drives the `nft_history` API.
+
+### migrator_status
+
+```
+CREATE TABLE clio.migrator_status (
+    migrator_name TEXT,     # The name of the migrator
+    status TEXT,            # The status of the migrator
+    PRIMARY KEY (migrator_name)
+) 
+```
+
+The `migrator_status` table stores the status of the migratior in this database. If a migrator's status is `migrated`, it means this database has finished data migration for this migrator.

src/data/cassandra/Schema.hpp

@@ -270,6 +270,18 @@ class Schema {
             qualifiedTableName(settingsProvider_.get(), "mp_token_holders")
         ));
 
+        statements.emplace_back(fmt::format(
+            R"(
+           CREATE TABLE IF NOT EXISTS {}
+                  ( 
+                   migrator_name TEXT,
+                          status TEXT,
+                         PRIMARY KEY (migrator_name)
+                  ) 
+            )",
+            qualifiedTableName(settingsProvider_.get(), "migrator_status")
+        ));
+
         return statements;
     }();
 
@@ -466,6 +478,17 @@ class Schema {
             ));
         }();
 
+        PreparedStatement insertMigratorStatus = [this]() {
+            return handle_.get().prepare(fmt::format(
+                R"(
+                INSERT INTO {}
+                       (migrator_name, status)
+                VALUES (?, ?)
+                )",
+                qualifiedTableName(settingsProvider_.get(), "migrator_status")
+            ));
+        }();
+
         //
         // Select queries
         //
@@ -768,6 +791,17 @@ class Schema {
                 qualifiedTableName(settingsProvider_.get(), "ledger_range")
             ));
         }();
+
+        PreparedStatement selectMigratorStatus = [this]() {
+            return handle_.get().prepare(fmt::format(
+                R"(
+                SELECT status
+                  FROM {}
+                 WHERE migrator_name = ?
+                )",
+                qualifiedTableName(settingsProvider_.get(), "migrator_status")
+            ));
+        }();
     };
 
     /**

src/data/cassandra/Types.hpp

@@ -21,6 +21,8 @@
 
 #include <cstdint>
 #include <expected>
+#include <string>
+#include <utility>
 
 namespace data::cassandra {
 
@@ -55,6 +57,26 @@ struct Limit {
     int32_t limit;
 };
 
+/**
+ * @brief A strong type wrapper for string
+ *
+ * This is unfortunately needed right now to support TEXT properly
+ * because clio uses string to represent BLOB
+ * If we want to bind TEXT with string, we need to use this type
+ */
+struct Text {
+    std::string text;
+
+    /**
+     * @brief Construct a new Text object from string type
+     *
+     * @param text The text to wrap
+     */
+    explicit Text(std::string text) : text{std::move(text)}
+    {
+    }
+};
+
 class Handle;
 class CassandraError;
 

src/data/cassandra/impl/Statement.hpp

@@ -119,6 +119,9 @@ class Statement : public ManagedObject<CassStatement> {
             // reinterpret_cast is needed here :'(
             auto const rc = bindBytes(reinterpret_cast<unsigned char const*>(value.data()), value.size());
             throwErrorIfNeeded(rc, "Bind string (as bytes)");
+        } else if constexpr (std::is_convertible_v<DecayedType, Text>) {
+            auto const rc = cass_statement_bind_string(*this, idx, value.text.c_str());
+            throwErrorIfNeeded(rc, "Bind string (as TEXT)");
         } else if constexpr (std::is_same_v<DecayedType, UintTupleType> ||
                              std::is_same_v<DecayedType, UintByteTupleType>) {
             auto const rc = cass_statement_bind_tuple(*this, idx, Tuple{std::forward<Type>(value)});

src/main/Main.cpp

@@ -19,6 +19,7 @@
 
 #include "app/CliArgs.hpp"
 #include "app/ClioApplication.hpp"
+#include "migration/MigrationApplication.hpp"
 #include "rpc/common/impl/HandlerProvider.hpp"
 #include "util/TerminationHandler.hpp"
 #include "util/config/Config.hpp"
@@ -46,6 +47,16 @@
             app::ClioApplication clio{config};
 
             return clio.run(run.useNgWebServer);
+        },
+        [](app::CliArgs::Action::Migrate const& migrate) {
+            auto const config = util::ConfigReader::open(migrate.configPath);
+            if (!config) {
+                std::cerr << "Couldnt parse migration config '" << migrate.configPath << "'." << std::endl;
+                return EXIT_FAILURE;
+            }
+            util::LogService::init(config);
+            app::MigratorApplication migrator{config, migrate.subCmd};
+            return migrator.run();
         }
     );
 } catch (std::exception const& e) {

src/migration/CMakeLists.txt

@@ -0,0 +1,8 @@
+add_library(clio_migration)
+
+target_sources(
+  clio_migration PRIVATE MigrationApplication.cpp impl/MigrationManagerFactory.cpp MigratorStatus.cpp
+                         cassandra/impl/ObjectsAdapter.cpp cassandra/impl/TransactionsAdapter.cpp
+)
+
+target_link_libraries(clio_migration PRIVATE clio_util clio_etl)