dht_proxy_server.h

/*
 *  Copyright (C) 2014-2023 Savoir-faire Linux Inc.
 *  Author: Sébastien Blin <sebastien.blin@savoirfairelinux.com>
 *          Adrien Béraud <adrien.beraud@savoirfairelinux.com>
 *          Vsevolod Ivanov <vsevolod.ivanov@savoirfairelinux.com>
 *
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 3 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program. If not, see <https://www.gnu.org/licenses/>.
 */
 
#pragma once
 
#include "callbacks.h"
#include "def.h"
#include "infohash.h"
#include "proxy.h"
#include "scheduler.h"
#include "sockaddr.h"
#include "value.h"
#include "http.h"
 
#include <restinio/all.hpp>
#include <restinio/tls.hpp>
#include <json/json.h>
 
#include <memory>
#include <mutex>
 
namespace dht {
enum class PushType {
    None = 0,
    Android,
    iOS,
    UnifiedPush
};
}
MSGPACK_ADD_ENUM(dht::PushType)
 
namespace Json {
class Value;
}
 
namespace dht {
 
namespace http {
class Request;
struct ListenerSession;
}
 
class DhtRunner;
 
using RestRouter = restinio::router::express_router_t<>;
using RequestStatus = restinio::request_handling_status_t;
 
struct OPENDHT_PUBLIC ProxyServerConfig {
    std::string address {};
    in_port_t port {8000};
    std::string pushServer {};
    std::string persistStatePath {};
    dht::crypto::Identity identity {};
    std::string bundleId {};
};

class OPENDHT_PUBLIC DhtProxyServer
{
public:
    DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
        const ProxyServerConfig& config = {},
        const std::shared_ptr<log::Logger>& logger = {});
 
    virtual ~DhtProxyServer();
 
    DhtProxyServer(const DhtProxyServer& other) = delete;
    DhtProxyServer(DhtProxyServer&& other) = delete;
    DhtProxyServer& operator=(const DhtProxyServer& other) = delete;
    DhtProxyServer& operator=(DhtProxyServer&& other) = delete;
 
    asio::io_context& io_context() const;
 
    using clock = std::chrono::steady_clock;
    using time_point = clock::time_point;
 
    struct PushStats {
        uint64_t highPriorityCount {0};
        uint64_t normalPriorityCount {0};
 
        void increment(bool highPriority) {
            if (highPriority)
                highPriorityCount++;
            else
                normalPriorityCount++;
        }
 
        Json::Value toJson() const {
            Json::Value val;
            val["highPriorityCount"] = static_cast<Json::UInt64>(highPriorityCount);
            val["normalPriorityCount"] = static_cast<Json::UInt64>(normalPriorityCount);
            return val;
        }
 
        std::string toString() const {
            return fmt::format("{} high priority, {} normal priority", highPriorityCount, normalPriorityCount);
        }
    };
 
    struct ServerStats {
        size_t listenCount {0};
        size_t putCount {0};
        size_t totalPermanentPuts {0};
        size_t pushListenersCount {0};

        time_point serverStartTime;
        time_point lastUpdated;
        PushStats androidPush;
        PushStats iosPush;
        PushStats unifiedPush;

        double requestRate {0};
        std::shared_ptr<NodeInfo> nodeInfo {};
 
        std::string toString() const {
            auto ret = fmt::format("Listens: {}, Puts: {}, PushListeners: {}\n"
                        "Push requests in the last {}: [Android: {}], [iOS: {}], [Unified: {}]\n"
                        "Requests: {} per second.",
                        listenCount, putCount, pushListenersCount,
                        print_duration(lastUpdated - serverStartTime),
                        androidPush.toString(), iosPush.toString(), unifiedPush.toString(),
                        requestRate);
            if (nodeInfo) {
                auto& ipv4 = nodeInfo->ipv4;
                if (ipv4.table_depth > 1)
                    ret += fmt::format("IPv4 Network estimation: {}\n", ipv4.getNetworkSizeEstimation());
                auto& ipv6 = nodeInfo->ipv6;
                if (ipv6.table_depth > 1)
                    ret += fmt::format("IPv6 Network estimation: {}\n", ipv6.getNetworkSizeEstimation());
            }
            return ret;
        }

        Json::Value toJson() const {
            Json::Value result;
            result["listenCount"] = static_cast<Json::UInt64>(listenCount);
            result["putCount"] = static_cast<Json::UInt64>(putCount);
            result["totalPermanentPuts"] = static_cast<Json::UInt64>(totalPermanentPuts);
            result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
            result["serverStartTime"] = static_cast<Json::LargestInt>(to_time_t(serverStartTime));
            result["lastUpdated"] = static_cast<Json::LargestInt>(to_time_t(lastUpdated));
            result["androidPush"] = androidPush.toJson();
            result["iosPush"] = iosPush.toJson();
            result["unifiedPush"] = unifiedPush.toJson();
            result["requestRate"] = requestRate;
            if (nodeInfo)
                result["nodeInfo"] = nodeInfo->toJson();
            return result;
        }
    };
 
    std::shared_ptr<ServerStats> stats() const { return stats_; }
 
    std::shared_ptr<ServerStats> updateStats(std::shared_ptr<NodeInfo> info) const;
 
    std::shared_ptr<DhtRunner> getNode() const { return dht_; }
 
private:
    class ConnectionListener;
    struct RestRouterTraitsTls;
    struct RestRouterTraits;
 
    template <typename HttpResponse>
    static HttpResponse initHttpResponse(HttpResponse response);
    static restinio::request_handling_status_t serverError(restinio::request_t& request);
 
    template< typename ServerSettings >
    void addServerSettings(ServerSettings& serverSettings,
                           const unsigned int max_pipelined_requests = 16);
 
    std::unique_ptr<RestRouter> createRestRouter();
 
    void onConnectionClosed(restinio::connection_id_t);

    RequestStatus getNodeInfo(restinio::request_handle_t request,
                               restinio::router::route_params_t params) const;

    RequestStatus getStats(restinio::request_handle_t request,
                           restinio::router::route_params_t params);

    RequestStatus get(restinio::request_handle_t request,
                       restinio::router::route_params_t params);

    RequestStatus listen(restinio::request_handle_t request,
                         restinio::router::route_params_t params);

    RequestStatus put(restinio::request_handle_t request,
                      restinio::router::route_params_t params);
 
    void handleCancelPermamentPut(const asio::error_code &ec, const InfoHash& key, Value::Id vid);
 
#ifdef OPENDHT_PROXY_SERVER_IDENTITY
    RequestStatus putSigned(restinio::request_handle_t request,
                            restinio::router::route_params_t params) const;

    RequestStatus putEncrypted(restinio::request_handle_t request,
                               restinio::router::route_params_t params);
 
#endif // OPENDHT_PROXY_SERVER_IDENTITY

    RequestStatus getFiltered(restinio::request_handle_t request,
                              restinio::router::route_params_t params);

    RequestStatus options(restinio::request_handle_t request,
                           restinio::router::route_params_t params);
 
    struct PushSessionContext {
        std::mutex lock;
        std::string sessionId;
        PushSessionContext(const std::string& id) : sessionId(id) {}
    };
 
#ifdef OPENDHT_PUSH_NOTIFICATIONS
    PushType getTypeFromString(const std::string& type);
    std::string getDefaultTopic(PushType type);
 
    RequestStatus pingPush(restinio::request_handle_t request,
                         restinio::router::route_params_t /*params*/);
    RequestStatus subscribe(restinio::request_handle_t request,
                            restinio::router::route_params_t params);

    RequestStatus unsubscribe(restinio::request_handle_t request,
                              restinio::router::route_params_t params);

    void sendPushNotification(const std::string& key, Json::Value&& json, PushType type, bool highPriority, const std::string& topic);

    void handleNotifyPushListenExpire(const asio::error_code &ec, const std::string pushToken,
                                      std::function<Json::Value()> json, PushType type, const std::string& topic);

    void handleCancelPushListen(const asio::error_code &ec, const std::string pushToken,
                                const InfoHash key, const std::string clientId);

    bool handlePushListen(const InfoHash& infoHash, const std::string& pushToken,
                          PushType type, const std::string& clientId,
                          const std::shared_ptr<DhtProxyServer::PushSessionContext>& sessionCtx, const std::string& topic,
                          const std::vector<std::shared_ptr<Value>>& values, bool expired);
 
#endif //OPENDHT_PUSH_NOTIFICATIONS
 
    void handlePrintStats(const asio::error_code &ec);
    void updateStats();
 
    template <typename Os>
    void saveState(Os& stream);
 
    template <typename Is>
    void loadState(Is& is, size_t size);
 
    std::shared_ptr<asio::io_context> ioContext_;
    std::shared_ptr<DhtRunner> dht_;
    Json::StreamWriterBuilder jsonBuilder_;
    Json::CharReaderBuilder jsonReaderBuilder_;
    std::mt19937_64 rd {crypto::getSeededRandomEngine<std::mt19937_64>()};
 
    std::string persistPath_;
 
    // http server
    std::thread serverThread_;
    std::unique_ptr<restinio::http_server_t<RestRouterTraitsTls>> httpsServer_;
    std::unique_ptr<restinio::http_server_t<RestRouterTraits>> httpServer_;
 
    // http client
    std::pair<std::string, std::string> pushHostPort_;
 
    mutable std::mutex requestLock_;
    std::map<unsigned int /*id*/, std::shared_ptr<http::Request>> requests_;
 
    std::shared_ptr<log::Logger> logger_;
 
    std::shared_ptr<ServerStats> stats_;
    std::shared_ptr<NodeInfo> nodeInfo_ {};
    std::unique_ptr<asio::steady_timer> printStatsTimer_;
    const time_point serverStartTime_;
    mutable std::mutex pushStatsMutex_;
    PushStats androidPush_;
    PushStats iosPush_;
    PushStats unifiedPush_;
 
    // Thread-safe access to listeners map.
    std::mutex lockListener_;
    // Shared with connection listener.
    std::map<restinio::connection_id_t, http::ListenerSession> listeners_;
    // Connection Listener observing conn state changes.
    std::shared_ptr<ConnectionListener> connListener_;
    struct PermanentPut {
        time_point expiration;
        std::string pushToken;
        std::string clientId;
        std::shared_ptr<PushSessionContext> sessionCtx;
        std::unique_ptr<asio::steady_timer> expireTimer;
        std::unique_ptr<asio::steady_timer> expireNotifyTimer;
        Sp<Value> value;
        PushType type;
        std::string topic;
 
        template <typename Packer>
        void msgpack_pack(Packer& p) const
        {
            p.pack_map(2 + (sessionCtx ? 1 : 0) + (clientId.empty() ? 0 : 1) + (type == PushType::None ? 0 : 2) + (topic.empty() ? 0 : 1));
            p.pack("value"); p.pack(value);
            p.pack("exp"); p.pack(to_time_t(expiration));
            if (not clientId.empty()) {
                p.pack("cid"); p.pack(clientId);
            }
            if (sessionCtx) {
                std::lock_guard<std::mutex> l(sessionCtx->lock);
                p.pack("sid"); p.pack(sessionCtx->sessionId);
            }
            if (type != PushType::None) {
                p.pack("t"); p.pack(type);
                p.pack("token"); p.pack(pushToken);
            }
            if (not topic.empty()) {
                p.pack("top"); p.pack(topic);
            }
        }
 
        void msgpack_unpack(const msgpack::object& o);
    };
    struct SearchPuts {
        std::map<dht::Value::Id, PermanentPut> puts;
        MSGPACK_DEFINE_ARRAY(puts)
    };
    std::mutex lockSearchPuts_;
    std::map<InfoHash, SearchPuts> puts_;
 
    mutable std::atomic<size_t> requestNum_ {0};
    mutable std::atomic<time_point> lastStatsReset_ {time_point::min()};
 
    std::string pushServer_;
    std::string bundleId_;
 
#ifdef OPENDHT_PUSH_NOTIFICATIONS
    struct Listener {
        time_point expiration;
        std::string clientId;
        std::shared_ptr<PushSessionContext> sessionCtx;
        std::future<size_t> internalToken;
        std::unique_ptr<asio::steady_timer> expireTimer;
        std::unique_ptr<asio::steady_timer> expireNotifyTimer;
        PushType type;
        std::string topic;
 
        template <typename Packer>
        void msgpack_pack(Packer& p) const
        {
            p.pack_map(3 + (sessionCtx ? 1 : 0) + (topic.empty() ? 0 : 1));
            p.pack("cid"); p.pack(clientId);
            p.pack("exp"); p.pack(to_time_t(expiration));
            if (sessionCtx) {
                std::lock_guard<std::mutex> l(sessionCtx->lock);
                p.pack("sid"); p.pack(sessionCtx->sessionId);
            }
            p.pack("t"); p.pack(type);
            if (!topic.empty()) {
                p.pack("top"); p.pack(topic);
            }
        }
 
        void msgpack_unpack(const msgpack::object& o);
    };
    struct PushListener {
        std::map<InfoHash, std::vector<Listener>> listeners;
        MSGPACK_DEFINE_ARRAY(listeners)
    };
    std::map<std::string, PushListener> pushListeners_;
#endif //OPENDHT_PUSH_NOTIFICATIONS
};
 
}