tx handler

reporting/Pg.cpp

@@ -1534,3 +1534,37 @@ getLedger(
 
     return info;
 }
+
+std::optional<LedgerRange>
+getLedgerRange(std::shared_ptr<PgPool>& pgPool)
+{
+    auto range = PgQuery(pgPool)("SELECT complete_ledgers()");
+    if (!range)
+        return {};
+
+    std::string res{range.c_str()};
+    try
+    {
+        size_t minVal = 0;
+        size_t maxVal = 0;
+        if (res == "empty" || res == "error" || res.empty())
+            return {};
+        else if (size_t delim = res.find('-'); delim != std::string::npos)
+        {
+            minVal = std::stol(res.substr(0, delim));
+            maxVal = std::stol(res.substr(delim + 1));
+        }
+        else
+        {
+            minVal = maxVal = std::stol(res);
+        }
+        return LedgerRange{minVal, maxVal};
+    }
+    catch (std::exception&)
+    {
+        BOOST_LOG_TRIVIAL(error)
+            << __func__ << " : "
+            << "Error parsing result of getCompleteLedgers()";
+    }
+    return {};
+}

reporting/Pg.h

@@ -22,6 +22,7 @@
 
 #include <ripple/basics/chrono.h>
 #include <ripple/ledger/ReadView.h>
+#include <boost/icl/closed_interval.hpp>
 #include <boost/json.hpp>
 #include <boost/lexical_cast.hpp>
 #include <boost/log/trivial.hpp>
@@ -523,4 +524,8 @@ getLedger(
     std::variant<std::monostate, ripple::uint256, uint32_t> const& whichLedger,
     std::shared_ptr<PgPool>& pgPool);
 
+using LedgerRange = boost::icl::closed_interval<uint32_t>;
+std::optional<LedgerRange>
+getLedgerRange(std::shared_ptr<PgPool>& pgPool);
+
 #endif  // RIPPLE_CORE_PG_H_INCLUDED

reporting/ReportingBackend.h

@@ -700,6 +700,83 @@ public:
             << " microseconds";
         return result;
     }
+    std::optional<
+        std::pair<std::vector<unsigned char>, std::vector<unsigned char>>>
+    fetchTransaction(void const* hash) const
+    {
+        BOOST_LOG_TRIVIAL(trace) << "Fetching from cassandra";
+        auto start = std::chrono::system_clock::now();
+        CassStatement* statement = cass_prepared_bind(selectTransaction_);
+        cass_statement_set_consistency(statement, CASS_CONSISTENCY_QUORUM);
+        CassError rc = cass_statement_bind_bytes(
+            statement, 0, static_cast<cass_byte_t const*>(hash), 32);
+        if (rc != CASS_OK)
+        {
+            cass_statement_free(statement);
+            BOOST_LOG_TRIVIAL(error) << "Binding Cassandra fetch query: " << rc
+                                     << ", " << cass_error_desc(rc);
+            return {};
+        }
+        CassFuture* fut;
+        do
+        {
+            fut = cass_session_execute(session_.get(), statement);
+            rc = cass_future_error_code(fut);
+            if (rc != CASS_OK)
+            {
+                std::stringstream ss;
+                ss << "Cassandra fetch error";
+                ss << ", retrying";
+                ss << ": " << cass_error_desc(rc);
+                BOOST_LOG_TRIVIAL(warning) << ss.str();
+            }
+        } while (rc != CASS_OK);
+
+        CassResult const* res = cass_future_get_result(fut);
+        cass_statement_free(statement);
+        cass_future_free(fut);
+
+        CassRow const* row = cass_result_first_row(res);
+        if (!row)
+        {
+            BOOST_LOG_TRIVIAL(error) << "Cassandra fetch error: no rows";
+            cass_result_free(res);
+            return {};
+        }
+        cass_byte_t const* txBuf;
+        std::size_t txBufSize;
+        rc = cass_value_get_bytes(
+            cass_row_get_column(row, 0), &txBuf, &txBufSize);
+        if (rc != CASS_OK)
+        {
+            cass_result_free(res);
+            BOOST_LOG_TRIVIAL(error) << "Cassandra fetch result error: " << rc
+                                     << ", " << cass_error_desc(rc);
+            return {};
+        }
+        std::vector<unsigned char> txResult{txBuf, txBuf + txBufSize};
+        cass_byte_t const* metaBuf;
+        std::size_t metaBufSize;
+        rc = cass_value_get_bytes(
+            cass_row_get_column(row, 0), &metaBuf, &metaBufSize);
+        if (rc != CASS_OK)
+        {
+            cass_result_free(res);
+            BOOST_LOG_TRIVIAL(error) << "Cassandra fetch result error: " << rc
+                                     << ", " << cass_error_desc(rc);
+            return {};
+        }
+        std::vector<unsigned char> metaResult{metaBuf, metaBuf + metaBufSize};
+        cass_result_free(res);
+        auto end = std::chrono::system_clock::now();
+        BOOST_LOG_TRIVIAL(debug)
+            << "Fetched from cassandra in "
+            << std::chrono::duration_cast<std::chrono::microseconds>(
+                   end - start)
+                   .count()
+            << " microseconds";
+        return {{txResult, metaResult}};
+    }
     /*
         std::pair<std::vector<std::pair<uint256, std::shared_ptr<Blob>>>,
     Status> doUpperBound(uint256 marker, std::uint32_t seq, std::uint32_t limit)

reporting/ReportingETL.h

@@ -326,6 +326,12 @@ public:
         return flatMapBackend_;
     }
 
+    std::shared_ptr<PgPool>&
+    getPgPool()
+    {
+        return pgPool_;
+    }
+
 private:
     void
     doWork();