draft

tidy up

Update document

update adapter

+ document

Add document

nits

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)
 
-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,8 @@ CliArgs::parse(int argc, char const* argv[])
         ("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")
+        ("migrate_rollback", po::value<std::string>(),"rollback the given migration")
     ;
     // clang-format on
     po::positional_options_description positional;
@@ -65,6 +68,19 @@ CliArgs::parse(int argc, char const* argv[])
     }
 
     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), MigratorApplication::Cmd::status()}};
+        return Action{Action::Migrate{std::move(configPath), MigratorApplication::Cmd::migration(opt)}};
+    }
+
+    if (parsed.count("migrate_rollback") != 0u) {
+        auto const opt = parsed["migrate_rollback"].as<std::string>();
+        return Action{Action::Migrate{std::move(configPath), MigratorApplication::Cmd::rollback(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;
+            MigratorApplication::Cmd 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

@@ -45,6 +45,7 @@
 #include <string>
 #include <thread>
 #include <type_traits>
+#include <unordered_set>
 #include <utility>
 #include <vector>
 
@@ -548,6 +549,15 @@ class BackendInterface {
         boost::asio::yield_context yield
     ) const;
 
+    /**
+     * @brief Database-specific implementation of fetching the migrated features.
+     *
+     * @param yield The coroutine context
+     * @return The name of migrated features on success; nullopt otherwise
+     */
+    virtual std::optional<std::unordered_set<std::string>>
+    fetchMigratedFeatures(boost::asio::yield_context yield) const = 0;
+
     /**
      * @brief Synchronously fetches the ledger range from DB.
      *

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,28 @@ class BasicCassandraBackend : public BackendInterface {
         return results;
     }
 
+    std::optional<std::unordered_set<std::string>>
+    fetchMigratedFeatures(boost::asio::yield_context yield) const override
+    {
+        auto const res = executor_.read(yield, schema_->selectMigratedFeatures);
+        if (not res) {
+            LOG(log_.error()) << "Could not fetch migrated features: " << res.error();
+            return {};
+        }
+
+        std::unordered_set<std::string> features;
+        auto const& results = res.value();
+        if (not results) {
+            LOG(log_.warn()) << "No migrated features in database";
+            return features;
+        }
+
+        for (auto [feature] : extract<std::string>(results))
+            features.insert(std::move(feature));
+
+        return features;
+    }
+
     void
     doWriteLedgerObject(std::string&& key, std::uint32_t const seq, std::string&& blob) 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.
+
+### migrated_features
+
+```
+CREATE TABLE clio.migrated_features (
+	status TEXT, 	  				# The migration status of the feature
+	feature_name TEXT,				# The name of the feature
+	PRIMARY KEY (status, feature_name)
+	) 
+```
+
+The `migrated_features` table stores the status of the migration of features in this database. If a feature's status is `migrated`, it means this database has finished data migration for this feature. The feature name is the name of the feature that has been migrated. Now the status can only be `migrated`. 

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 {}
+                  ( 
+                    status TEXT,
+                    feature_name TEXT,
+                    PRIMARY KEY (status, feature_name)
+                  ) 
+            )",
+            qualifiedTableName(settingsProvider_.get(), "migrated_features")
+        ));
+
         return statements;
     }();
 
@@ -768,6 +780,17 @@ class Schema {
                 qualifiedTableName(settingsProvider_.get(), "ledger_range")
             ));
         }();
+
+        PreparedStatement selectMigratedFeatures = [this]() {
+            return handle_.get().prepare(fmt::format(
+                R"(
+                SELECT feature_name
+                  FROM {}
+                WHERE status = 'migrated'
+                )",
+                qualifiedTableName(settingsProvider_.get(), "migrated_features")
+            ));
+        }();
     };
 
     /**

src/data/cassandra/Types.hpp

@@ -21,6 +21,8 @@
 
 #include <cstdint>
 #include <expected>
+#include <string>
+#include <utility>
 
 namespace data::cassandra {
 
@@ -55,6 +57,25 @@ 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;
+
+    explicit Text(std::string text) : text{std::move(text)}
+    {
+    }
+
+    explicit Text(char const* text) : text{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 @@ try {
             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 MigrationManagerFactory.cpp cassandra/ObjectsAdapter.cpp
+                         cassandra/TransactionsAdapter.cpp
+)
+
+target_link_libraries(clio_migration PRIVATE clio_util clio_etl)