feat: Migration framework (#1768)

This PR implemented the migration framework, which contains the command
line interface to execute migration and helps to migrate data easily.
Please read README.md for more information about this framework.

src/data/BackendInterface.hpp

@@ -548,6 +548,16 @@ public:
         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 @@ public:
     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

@@ -72,13 +72,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 +837,26 @@ public:
         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 +984,14 @@ public:
         // 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 @@ public:
             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 @@ public:
             ));
         }();
 
+        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 @@ public:
                 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/SettingsProvider.cpp

@@ -24,7 +24,6 @@
 #include "util/Constants.hpp"
 #include "util/newconfig/ObjectView.hpp"
 
-
 #include <cerrno>
 #include <chrono>
 #include <cstddef>

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 @@ public:
             // 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_n(*this, idx, value.text.c_str(), value.text.size());
+            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)});