add basic account_tx handler

CMakeLists.txt

@@ -62,7 +62,8 @@ target_sources(reporting PRIVATE
     reporting/ReportingETL.cpp
     handlers/AccountInfo.cpp
     handlers/Tx.cpp
-    handlers/RPCHelpers.cpp)
+    handlers/RPCHelpers.cpp
+    handlers/AccountTx.cpp)
 
 
 message(${Boost_LIBRARIES})

handlers/AccountTx.cpp

@@ -0,0 +1,163 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2012-2014 Ripple Labs Inc.
+
+    Permission to use, copy, modify, and/or distribute this software for any
+    purpose  with  or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include <handlers/RPCHelpers.h>
+#include <reporting/Pg.h>
+#include <reporting/ReportingBackend.h>
+
+std::vector<std::pair<
+    std::shared_ptr<ripple::STTx const>,
+    std::shared_ptr<ripple::STObject const>>>
+doAccountTxStoredProcedure(
+    ripple::AccountID const& account,
+    std::shared_ptr<PgPool>& pgPool,
+    CassandraFlatMapBackend const& backend)
+{
+    pg_params dbParams;
+
+    char const*& command = dbParams.first;
+    std::vector<std::optional<std::string>>& values = dbParams.second;
+    command =
+        "SELECT account_tx($1::bytea, $2::bool, "
+        "$3::bigint, $4::bigint, $5::bigint, $6::bytea, "
+        "$7::bigint, $8::bool, $9::bigint, $10::bigint)";
+    values.resize(10);
+    values[0] = "\\x" + ripple::strHex(account);
+    values[1] = "true";
+    static std::uint32_t const page_length(200);
+    values[2] = std::to_string(page_length);
+
+    auto res = PgQuery(pgPool)(dbParams);
+    if (!res)
+    {
+        BOOST_LOG_TRIVIAL(error)
+            << __func__ << " : Postgres response is null - account = "
+            << ripple::strHex(account);
+        assert(false);
+        return {};
+    }
+    else if (res.status() != PGRES_TUPLES_OK)
+    {
+        assert(false);
+        return {};
+    }
+
+    if (res.isNull() || res.ntuples() == 0)
+    {
+        BOOST_LOG_TRIVIAL(error)
+            << __func__ << " : No data returned from Postgres : account = "
+            << ripple::strHex(account);
+
+        assert(false);
+        return {};
+    }
+
+    char const* resultStr = res.c_str();
+
+    boost::json::object result = boost::json::parse(resultStr).as_object();
+    if (result.contains("transactions"))
+    {
+        std::vector<ripple::uint256> nodestoreHashes;
+        for (auto& t : result.at("transactions").as_array())
+        {
+            boost::json::object obj = t.as_object();
+            if (obj.contains("ledger_seq") && obj.contains("nodestore_hash"))
+            {
+                std::string nodestoreHashHex =
+                    obj.at("nodestore_hash").as_string().c_str();
+                nodestoreHashHex.erase(0, 2);
+                ripple::uint256 nodestoreHash;
+                if (!nodestoreHash.parseHex(nodestoreHashHex))
+                    assert(false);
+
+                if (nodestoreHash.isNonZero())
+                {
+                    nodestoreHashes.push_back(nodestoreHash);
+                }
+                else
+                {
+                    assert(false);
+                }
+            }
+            else
+            {
+                assert(false);
+            }
+        }
+
+        std::vector<std::pair<
+            std::shared_ptr<ripple::STTx const>,
+            std::shared_ptr<ripple::STObject const>>>
+            results;
+        auto dbResults = backend.fetchBatch(nodestoreHashes);
+        for (auto const& res : dbResults)
+        {
+            if (res.first.size() && res.second.size())
+                results.push_back(deserializeTxPlusMeta(res));
+        }
+        return results;
+    }
+    return {};
+}
+
+// {
+//   account: account,
+//   ledger_index_min: ledger_index  // optional, defaults to earliest
+//   ledger_index_max: ledger_index, // optional, defaults to latest
+//   binary: boolean,                // optional, defaults to false
+//   forward: boolean,               // optional, defaults to false
+//   limit: integer,                 // optional
+//   marker: object {ledger: ledger_index, seq: txn_sequence} // optional,
+//   resume previous query
+// }
+boost::json::object
+doAccountTx(
+    boost::json::object const& request,
+    CassandraFlatMapBackend const& backend,
+    std::shared_ptr<PgPool>& pgPool)
+{
+    boost::json::object response;
+
+    if (!request.contains("account"))
+    {
+        response["error"] = "Please specify an account";
+        return response;
+    }
+
+    auto const account = ripple::parseBase58<ripple::AccountID>(
+        request.at("account").as_string().c_str());
+    if (!account)
+    {
+        response["error"] = "account malformed";
+        return response;
+    }
+
+    boost::json::array txns;
+    auto res = doAccountTxStoredProcedure(*account, pgPool, backend);
+    for (auto const& [sttx, meta] : res)
+    {
+        boost::json::object obj;
+        obj["transaction"] = getJson(*sttx);
+        obj["metadata"] = getJson(*meta);
+        txns.push_back(obj);
+    }
+    response["transactions"] = txns;
+    return response;
+}
+

reporting/ReportingBackend.h

@@ -20,6 +20,7 @@
 #ifndef RIPPLE_APP_REPORTING_REPORTINGBACKEND_H_INCLUDED
 #define RIPPLE_APP_REPORTING_REPORTINGBACKEND_H_INCLUDED
 
+#include <ripple/basics/base_uint.h>
 #include <boost/asio.hpp>
 #include <boost/filesystem.hpp>
 #include <boost/json.hpp>
@@ -905,8 +906,8 @@ public:
 
     struct ReadCallbackData
     {
-        CassandraFlatMapBackend& backend;
+        CassandraFlatMapBackend const& backend;
-        const void* const hash;
+        ripple::uint256 const& hash;
         BlobPair& result;
         std::condition_variable& cv;
 
@@ -914,8 +915,8 @@ public:
         size_t batchSize;
 
         ReadCallbackData(
-            CassandraFlatMapBackend& backend,
+            CassandraFlatMapBackend const& backend,
-            const void* const hash,
+            ripple::uint256 const& hash,
             BlobPair& result,
             std::condition_variable& cv,
             std::atomic_uint32_t& numFinished,
@@ -933,7 +934,7 @@ public:
     };
 
     std::vector<BlobPair>
-    fetchBatch(std::vector<void const*> const& hashes)
+    fetchBatch(std::vector<ripple::uint256> const& hashes) const
     {
         std::size_t const numHashes = hashes.size();
         BOOST_LOG_TRIVIAL(trace)
@@ -947,7 +948,7 @@ public:
         for (std::size_t i = 0; i < hashes.size(); ++i)
         {
             cbs.push_back(std::make_shared<ReadCallbackData>(
-                *this, &hashes[i], results[i], cv, numFinished, numHashes));
+                *this, hashes[i], results[i], cv, numFinished, numHashes));
             read(*cbs[i]);
         }
         assert(results.size() == cbs.size());
@@ -963,12 +964,15 @@ public:
     }
 
     void
-    read(ReadCallbackData& data)
+    read(ReadCallbackData& data) const
     {
         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*>(data.hash), 32);
+            statement,
+            0,
+            static_cast<cass_byte_t const*>(data.hash.data()),
+            32);
         if (rc != CASS_OK)
         {
             size_t batchSize = data.batchSize;

test.py

@@ -22,6 +22,16 @@ async def account_info(ip, port):
     except websockets.exceptions.ConnectionClosedError as e:
         print(e)
 
+async def account_tx(ip, port):
+    address = 'ws://' + str(ip) + ':' + str(port)
+    try:
+        async with websockets.connect(address) as ws:
+            await ws.send(json.dumps({"command":"account_tx","account":"rDzTZxa7NwD9vmNf5dvTbW4FQDNSRsfPv6"}))
+            res = json.loads(await ws.recv())
+            print(res)
+    except websockets.exceptions.ConnectionClosedError as e:
+        print(e)
+
 async def tx(ip, port, tx_hash):
     address = 'ws://' + str(ip) + ':' + str(port)
     try:
@@ -36,7 +46,7 @@ async def tx(ip, port, tx_hash):
 
 
 parser = argparse.ArgumentParser(description='test script for xrpl-reporting')
-parser.add_argument('action', choices=["account_info", "tx"])
+parser.add_argument('action', choices=["account_info", "tx", "account_tx"])
 parser.add_argument('--ip', default='127.0.0.1')
 parser.add_argument('--port', default='8080')
 parser.add_argument('--hash')
@@ -53,6 +63,9 @@ def run(args):
     if args.action == "tx":
         asyncio.get_event_loop().run_until_complete(
                 tx(args.ip, args.port, args.hash))
+    if args.action == "account_tx":
+        asyncio.get_event_loop().run_until_complete(
+                account_tx(args.ip, args.port))
     else:
         print("incorrect arguments")
 

websocket_server_async.cpp

@@ -51,6 +51,11 @@ doTx(
     boost::json::object const& request,
     CassandraFlatMapBackend const& backend,
     std::shared_ptr<PgPool>& pgPool);
+boost::json::object
+doAccountTx(
+    boost::json::object const& request,
+    CassandraFlatMapBackend const& backend,
+    std::shared_ptr<PgPool>& pgPool);
 
 boost::json::object
 buildResponse(
@@ -66,6 +71,7 @@ buildResponse(
             return doTx(request, backend, pgPool);
             break;
         case account_tx:
+            return doAccountTx(request, backend, pgPool);
             break;
         case ledger:
             break;