lokinet/llarp/rpc/lokid_rpc_client.cpp

#include <rpc/lokid_rpc_client.hpp>

#include <util/logging/logger.h>
#include <util/logging/logger.hpp>

#include <router/abstractrouter.hpp>

#include <nlohmann/json.hpp>

#include <util/time.hpp>
#include <util/thread/logic.hpp>

namespace llarp
{
  namespace rpc
  {
    static lokimq::LogLevel
    toLokiMQLogLevel(llarp::LogLevel level)
    {
      switch (level)
      {
        case eLogError:
          return lokimq::LogLevel::error;
        case eLogWarn:
          return lokimq::LogLevel::warn;
        case eLogInfo:
          return lokimq::LogLevel::info;
        case eLogDebug:
          return lokimq::LogLevel::debug;
        case eLogNone:
        default:
          return lokimq::LogLevel::trace;
      }
    }

    LokidRpcClient::LokidRpcClient(LMQ_ptr lmq, AbstractRouter* r)
        : m_lokiMQ(std::move(lmq)), m_Router(r)
    {
      // m_lokiMQ->log_level(toLokiMQLogLevel(LogLevel::Instance().curLevel));
    }

    void
    LokidRpcClient::ConnectAsync(std::string_view url)
    {
      LogInfo("connecting to lokid via LMQ at ", url);
      m_lokiMQ->connect_remote(
          std::move(url),
          [self = shared_from_this()](lokimq::ConnectionID c) {
            self->m_Connection = std::move(c);
            self->Connected();
          },
          [](lokimq::ConnectionID, std::string_view f) {
            llarp::LogWarn("Failed to connect to lokid: ", f);
          });
    }

    void
    LokidRpcClient::Command(std::string_view cmd)
    {
      LogDebug("lokid command: ", cmd);
      m_lokiMQ->send(*m_Connection, std::move(cmd));
    }

    void
    LokidRpcClient::Connected()
    {
      constexpr auto PingInterval = 1min;
      constexpr auto NodeListUpdateInterval = 90s;

      LogInfo("we connected to lokid [", *m_Connection, "]");
      m_lokiMQ->add_timer(
          [self = shared_from_this()]() {
            LogInfo("Telling lokid we are live");
            self->Command("rpc.lokinet_ping");
          },
          PingInterval);

      m_lokiMQ->add_timer(
          [self = shared_from_this()]() {
            nlohmann::json request;
            request["pubkey_ed25519"] = true;
            request["active_only"] = true;
            if (not self->m_CurrentBlockHash.empty())
              request["poll_block_hash"] = self->m_CurrentBlockHash;
            self->Request(
                "rpc.get_service_nodes",
                [self](bool success, std::vector<std::string> data) {
                  if (not success)
                  {
                    LogWarn("failed to update service node list");
                    return;
                  }
                  if (data.size() < 2)
                  {
                    LogWarn("lokid gave empty reply for service node list");
                    return;
                  }
                  try
                  {
                    self->HandleGotServiceNodeList(std::move(data[1]));
                  }
                  catch (std::exception& ex)
                  {
                    LogError("failed to process service node list: ", ex.what());
                  }
                },
                request.dump());
          },
          NodeListUpdateInterval);
    }

    void
    LokidRpcClient::HandleGotServiceNodeList(std::string data)
    {
      auto j = nlohmann::json::parse(std::move(data));
      {
        const auto itr = j.find("block_hash");
        if (itr != j.end())
        {
          m_CurrentBlockHash = itr->get<std::string>();
        }
      }
      {
        const auto itr = j.find("unchanged");
        if (itr != j.end())
        {
          if (itr->get<bool>())
          {
            LogDebug("service node list unchanged");
            return;
          }
        }
      }

      std::vector<RouterID> nodeList;
      {
        const auto itr = j.find("service_node_states");
        if (itr != j.end() and itr->is_array())
        {
          for (auto j_itr = itr->begin(); j_itr != itr->end(); j_itr++)
          {
            const auto ed_itr = j_itr->find("pubkey_ed25519");
            if (ed_itr == j_itr->end() or not ed_itr->is_string())
              continue;
            RouterID rid;
            if (rid.FromHex(ed_itr->get<std::string>()))
              nodeList.emplace_back(std::move(rid));
          }
        }
      }

      if (nodeList.empty())
      {
        LogWarn("got empty service node list from lokid");
        return;
      }

      // inform router about the new list
      LogicCall(m_Router->logic(), [r = m_Router, nodeList]() { r->SetRouterWhitelist(nodeList); });
    }

  }  // namespace rpc
}  // namespace llarp