clio/src/etl/impl/SubscriptionSource.hpp

#pragma once

#include "etl/NetworkValidatedLedgersInterface.hpp"
#include "etl/Source.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "util/Mutex.hpp"
#include "util/Retry.hpp"
#include "util/StopHelper.hpp"
#include "util/log/Logger.hpp"
#include "util/prometheus/Gauge.hpp"
#include "util/requests/Types.hpp"
#include "util/requests/WsConnection.hpp"

#include <boost/asio/io_context.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/asio/strand.hpp>
#include <boost/beast/http/field.hpp>
#include <fmt/format.h>

#include <atomic>
#include <chrono>
#include <cstdint>
#include <functional>
#include <memory>
#include <optional>
#include <string>
#include <utility>
#include <vector>

namespace etl::impl {

/**
 * @brief This class is used to subscribe to a source of ledger data and forward it to the
 * subscription manager.
 * @note This class is safe to delete only if io_context is stopped.
 */
class SubscriptionSource {
public:
    using OnConnectHook = SourceBase::OnConnectHook;
    using OnDisconnectHook = SourceBase::OnDisconnectHook;
    using OnLedgerClosedHook = SourceBase::OnLedgerClosedHook;

private:
    util::Logger log_;
    util::requests::WsConnectionBuilder wsConnectionBuilder_;
    util::requests::WsConnectionPtr wsConnection_;

    struct ValidatedLedgersData {
        std::vector<std::pair<uint32_t, uint32_t>> validatedLedgers;
        std::string validatedLedgersRaw{"N/A"};
    };
    util::Mutex<ValidatedLedgersData> validatedLedgersData_;

    std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers_;
    std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions_;

    boost::asio::strand<boost::asio::io_context::executor_type> strand_;

    std::chrono::steady_clock::duration wsTimeout_;

    util::Retry retry_;

    OnConnectHook onConnect_;
    OnDisconnectHook onDisconnect_;
    OnLedgerClosedHook onLedgerClosed_;

    std::atomic_bool isConnected_{false};
    std::atomic_bool stop_{false};
    std::atomic_bool isForwarding_{false};

    util::Mutex<std::chrono::steady_clock::time_point> lastMessageTime_;

    std::reference_wrapper<util::prometheus::GaugeInt> lastMessageTimeSecondsSinceEpoch_;

    util::StopHelper stopHelper_;

    static constexpr std::chrono::seconds kWsTimeout{30};
    static constexpr std::chrono::seconds kMaxRetryDelay{30};
    static constexpr std::chrono::seconds kRetryDelay{1};

public:
    /**
     * @brief Construct a new Subscription Source object
     *
     * @tparam NetworkValidatedLedgersType The type of the network validated ledgers object
     * @param ioContext The io_context to use
     * @param ip The ip address of the source
     * @param wsPort The port of the source
     * @param validatedLedgers The network validated ledgers object
     * @param subscriptions The subscription manager object
     * @param onConnect The onConnect hook. Called when the connection is established
     * @param onDisconnect The onDisconnect hook. Called when the connection is lost
     * @param onLedgerClosed The onLedgerClosed hook. Called when the ledger is closed if the source
     * is forwarding
     * @param wsTimeout A timeout for websocket operations. Defaults to 30 seconds
     * @param retryDelay The retry delay. Defaults to 1 second
     */
    SubscriptionSource(
        boost::asio::io_context& ioContext,
        std::string const& ip,
        std::string const& wsPort,
        std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers,
        std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
        OnConnectHook onConnect,
        OnDisconnectHook onDisconnect,
        OnLedgerClosedHook onLedgerClosed,
        std::chrono::steady_clock::duration const wsTimeout = SubscriptionSource::kWsTimeout,
        std::chrono::steady_clock::duration const retryDelay = SubscriptionSource::kRetryDelay
    );

    /**
     * @brief Run the source
     */
    void
    run();

    /**
     * @brief Check if the source has a ledger
     *
     * @param sequence The sequence of the ledger
     * @return true if the source has the ledger, false otherwise
     */
    bool
    hasLedger(uint32_t sequence) const;

    /**
     * @brief Check if the source is connected
     *
     * @return true if the source is connected, false otherwise
     */
    bool
    isConnected() const;

    /**
     * @brief Get whether the source is forwarding
     *
     * @return true if the source is forwarding, false otherwise
     */
    bool
    isForwarding() const;

    /**
     * @brief Set source forwarding
     *
     * @note If forwarding is true the source will forward messages to the subscription manager.
     * Forwarding is being reset on disconnect.
     * @param isForwarding The new forwarding state
     */
    void
    setForwarding(bool isForwarding);

    /**
     * @brief Get the last message time (even if the last message had an error)
     *
     * @return The last message time
     */
    std::chrono::steady_clock::time_point
    lastMessageTime() const;

    /**
     * @brief Get the last received raw string of the validated ledgers
     *
     * @return The validated ledgers string
     */
    std::string const&
    validatedRange() const;

    /**
     * @brief Stop the source. The source will complete already scheduled operations but will not
     * schedule new ones
     */
    void
    stop(boost::asio::yield_context yield);

private:
    void
    subscribe();

    std::optional<util::requests::RequestError>
    handleMessage(std::string const& message);

    void
    handleError(util::requests::RequestError const& error, boost::asio::yield_context yield);

    void
    logError(util::requests::RequestError const& error) const;

    void
    setLastMessageTime();

    void
    setValidatedRange(std::string range);

    static std::string const&
    getSubscribeCommandJson();
};

}  // namespace etl::impl