dht_proxy_client.cpp

/*
 *  Copyright (C) 2016 Savoir-faire Linux Inc.
 *  Author: Sébastien Blin <sebastien.blin@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/>.
 */

#if OPENDHT_PROXY_CLIENT

#include "dht_proxy_client.h"

#include <chrono>
#include <json/json.h>
#include <restbed>
#include <vector>

#include "dhtrunner.h"

#include <iostream>

constexpr const char* const HTTP_PROTO {"http://"};

namespace dht {

DhtProxyClient::DhtProxyClient(const std::string& serverHost)
: serverHost_(serverHost), lockCurrentProxyInfos_(new std::mutex()),
  scheduler(DHT_LOG), currentProxyInfos_(new Json::Value())
{
    if (!serverHost_.empty())
        startProxy(serverHost_);
}

void
DhtProxyClient::confirmProxy()
{
    if (serverHost_.empty()) return;
    // Retrieve the connectivity each hours if connected, else every 5 seconds.
    auto disconnected_old_status =  statusIpv4_ == NodeStatus::Disconnected && statusIpv6_ == NodeStatus::Disconnected;
    getConnectivityStatus();
    auto disconnected_new_status = statusIpv4_ == NodeStatus::Disconnected && statusIpv6_ == NodeStatus::Disconnected;
    auto time = disconnected_new_status ? std::chrono::seconds(5) : std::chrono::hours(1);
    if (disconnected_old_status && !disconnected_new_status) {
        restartListeners();
    }
    auto confirm_proxy_time = scheduler.time() + time;
    scheduler.edit(nextProxyConfirmation, confirm_proxy_time);
}

void
DhtProxyClient::startProxy(const std::string& serverHost)
{
    serverHost_ = serverHost;
    if (serverHost_.empty()) return;
    auto confirm_proxy_time = scheduler.time() + std::chrono::seconds(5);
    nextProxyConfirmation = scheduler.add(confirm_proxy_time, std::bind(&DhtProxyClient::confirmProxy, this));
    auto confirm_connectivity = scheduler.time() + std::chrono::seconds(5);
    nextConnectivityConfirmation = scheduler.add(confirm_connectivity, std::bind(&DhtProxyClient::confirmConnectivity, this));

    getConnectivityStatus();
}

void
DhtProxyClient::confirmConnectivity()
{
    // The scheduler must get if the proxy is disconnected
    auto confirm_connectivity = scheduler.time() + std::chrono::seconds(3);
    scheduler.edit(nextConnectivityConfirmation, confirm_connectivity);
}

DhtProxyClient::~DhtProxyClient()
{
    cancelAllOperations();
    cancelAllListeners();
}

void
DhtProxyClient::cancelAllOperations()
{
    std::lock_guard<std::mutex> lock(lockOperations_);
    auto operation = operations_.begin();
    while (operation != operations_.end()) {
        if (operation->thread.joinable()) {
            // Close connection to stop operation?
            restbed::Http::close(operation->req);
            operation->thread.join();
            operation = operations_.erase(operation);
        } else {
            ++operation;
        }
    }
}

void
DhtProxyClient::cancelAllListeners()
{
    std::lock_guard<std::mutex> lock(lockListener_);
    for (auto& listener: listeners_) {
        if (listener.thread.joinable()) {
            // Close connection to stop listener?
            if (listener.req)
                restbed::Http::close(listener.req);
            listener.thread.join();
        }
    }
}

void
DhtProxyClient::shutdown(ShutdownCallback cb)
{
    cancelAllOperations();
    cancelAllListeners();
    if (cb)
        cb();
}

NodeStatus
DhtProxyClient::getStatus(sa_family_t af) const
{
    switch (af)
    {
    case AF_INET:
        return statusIpv4_;
    case AF_INET6:
        return statusIpv6_;
    default:
        return NodeStatus::Disconnected;
    }
}

bool
DhtProxyClient::isRunning(sa_family_t af) const
{
    switch (af)
    {
    case AF_INET:
        return statusIpv4_ == NodeStatus::Connected;
    case AF_INET6:
        return statusIpv6_ == NodeStatus::Connected;
    default:
        return false;
    }
}

time_point
DhtProxyClient::periodic(const uint8_t*, size_t, const SockAddr&)
{
    // Exec all currently stored callbacks
    scheduler.syncTime();
    if (!callbacks_.empty()) {
        std::lock_guard<std::mutex> lock(lockCallbacks);
        for (auto& callback : callbacks_)
            callback();
        callbacks_.clear();
    }
    // Remove finished operations
    {
        std::lock_guard<std::mutex> lock(lockOperations_);
        auto operation = operations_.begin();
        while (operation != operations_.end()) {
            if (*(operation->finished)) {
                if (operation->thread.joinable()) {
                    // Close connection to stop operation?
                    restbed::Http::close(operation->req);
                    operation->thread.join();
                }
                operation = operations_.erase(operation);
            } else {
                ++operation;
            }
        }
    }
    return scheduler.run();
}

void
DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb,
                    Value::Filter&& filter, Where&& where)
{
    restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + key.toString());
    auto req = std::make_shared<restbed::Request>(uri);
    Query query {{}, where};
    auto filterChain = filter.chain(query.where.getFilter());

    auto finished = std::make_shared<bool>(false);
    Operation o;
    o.req = req;
    o.finished = finished;
    o.thread = std::thread([=](){
        // Try to contact the proxy and set the status to connected when done.
        // will change the connectivity status
        auto ok = std::make_shared<bool>(true);
        restbed::Http::async(req,
            [=](const std::shared_ptr<restbed::Request>& req,
                const std::shared_ptr<restbed::Response>& reply) {
            auto code = reply->get_status_code();

            if (code == 200) {
                try {
                    while (restbed::Http::is_open(req) and not *finished) {
                        restbed::Http::fetch("\n", reply);
                        if (*finished)
                            break;
                        std::string body;
                        reply->get_body(body);
                        reply->set_body(""); // Reset the body for the next fetch

                        Json::Value json;
                        Json::Reader reader;
                        if (reader.parse(body, json)) {
                            auto value = std::make_shared<Value>(json);
                            if ((not filterChain or filterChain(*value)) && cb) {
                                std::lock_guard<std::mutex> lock(lockCallbacks);
                                callbacks_.emplace_back([cb, value, finished]() {
                                    if (not *finished and not cb({value}))
                                        *finished = true;
                                });
                            }
                        } else {
                            *ok = false;
                        }
                    }
                } catch (std::runtime_error& e) { }
            } else {
                *ok = false;
            }
        }).wait();
        if (donecb) {
            std::lock_guard<std::mutex> lock(lockCallbacks);
            callbacks_.emplace_back([=](){
                donecb(*ok, {});
            });
        }
        if (!ok) {
            // Connection failed, update connectivity
            getConnectivityStatus();
        }
        *finished = true;
    });
    {
        std::lock_guard<std::mutex> lock(lockOperations_);
        operations_.emplace_back(std::move(o));
    }
}

void
DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_point, bool permanent)
{
    restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + key.toString());
    auto req = std::make_shared<restbed::Request>(uri);
    req->set_method("POST");
    Json::FastWriter writer;
    auto json = val->toJson();
    if (permanent)
        json["permanent"] = true;
    auto body = writer.write(json);
    req->set_body(body);
    req->set_header("Content-Length", std::to_string(body.size()));

    auto finished = std::make_shared<bool>(false);
    Operation o;
    o.req = req;
    o.finished = finished;
    o.thread = std::thread([=](){
        auto ok = std::make_shared<bool>(true);
        restbed::Http::async(req,
            [this, ok](const std::shared_ptr<restbed::Request>& /*req*/,
                        const std::shared_ptr<restbed::Response>& reply) {
            auto code = reply->get_status_code();

            if (code == 200) {
                restbed::Http::fetch("\n", reply);
                std::string body;
                reply->get_body(body);
                reply->set_body(""); // Reset the body for the next fetch

                Json::Value json;
                Json::Reader reader;
                try {
                    if (!reader.parse(body, json))
                        *ok = false;
                } catch (...) {
                    *ok = false;
                }
            } else {
                *ok = false;
            }
        }).wait();
        if (cb) {
            std::lock_guard<std::mutex> lock(lockCallbacks);
            callbacks_.emplace_back([=](){
                cb(*ok, {});
            });
        }
        if (!ok) {
            // Connection failed, update connectivity
            getConnectivityStatus();
        }
        *finished = true;
    });
    {
        std::lock_guard<std::mutex> lock(lockOperations_);
        operations_.emplace_back(std::move(o));
    }
}

NodeStats
DhtProxyClient::getNodesStats(sa_family_t af) const
{
    lockCurrentProxyInfos_->lock();
    auto proxyInfos = *currentProxyInfos_;
    lockCurrentProxyInfos_->unlock();
    NodeStats stats {};
    auto identifier = af == AF_INET6 ? "ipv6" : "ipv4";
    try {
        stats = NodeStats(proxyInfos[identifier]);
    } catch (...) { }
    return stats;
}

Json::Value
DhtProxyClient::getProxyInfos() const
{
    restbed::Uri uri(HTTP_PROTO + serverHost_ + "/");
    auto req = std::make_shared<restbed::Request>(uri);

    // Try to contact the proxy and set the status to connected when done.
    // will change the connectivity status
    restbed::Http::async(req,
        [this](const std::shared_ptr<restbed::Request>&,
                       const std::shared_ptr<restbed::Response>& reply) {
        auto code = reply->get_status_code();

        if (code == 200) {
            restbed::Http::fetch("\n", reply);
            std::string body;
            reply->get_body(body);

            Json::Reader reader;
            lockCurrentProxyInfos_->lock();
            try {
                reader.parse(body, *currentProxyInfos_);
            } catch (...) {
                *currentProxyInfos_ = Json::Value();
            }
            lockCurrentProxyInfos_->unlock();
        } else {
            lockCurrentProxyInfos_->lock();
            *currentProxyInfos_ = Json::Value();
            lockCurrentProxyInfos_->unlock();
        }
    }).wait();
    lockCurrentProxyInfos_->lock();
    auto result = *currentProxyInfos_;
    lockCurrentProxyInfos_->unlock();
    return result;
}

std::vector<SockAddr>
DhtProxyClient::getPublicAddress(sa_family_t family)
{
    lockCurrentProxyInfos_->lock();
    auto proxyInfos = *currentProxyInfos_;
    lockCurrentProxyInfos_->unlock();
    // json["public_ip"] contains [ipv6:ipv4]:port or ipv4:port
    auto public_ip = proxyInfos["public_ip"].asString();
    if (!proxyInfos.isMember("public_ip") || (public_ip.length() < 2)) return {};
    std::string ipv4Address = "", ipv6Address = "", port = "";
    if (public_ip[0] == '[') {
        // ipv6 complient
        auto endIp = public_ip.find(']');
        if (public_ip.length() > endIp + 2) {
            port = public_ip.substr(endIp + 2);
            auto ips = public_ip.substr(1, endIp - 1);
            auto ipv4And6Separator = ips.find_last_of(':');
            ipv4Address = ips.substr(ipv4And6Separator + 1);
            ipv6Address = ips.substr(0, ipv4And6Separator - 1);
        }
    } else {
        auto endIp = public_ip.find_last_of(':');
        port = public_ip.substr(endIp + 1);
        ipv4Address = public_ip.substr(0, endIp - 1);
    }
    switch (family)
    {
    case AF_INET:
        return SockAddr::resolve(ipv4Address, port);
    case AF_INET6:
        return SockAddr::resolve(ipv6Address, port);
    default:
        return {};
    }
}

size_t
DhtProxyClient::listen(const InfoHash& key, GetCallback cb, Value::Filter&& filter, Where&& where)
{
    restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + key.toString());
    auto req = std::make_shared<restbed::Request>(uri);
    req->set_method("LISTEN");

    Query query {{}, where};
    auto filterChain = filter.chain(query.where.getFilter());

    Listener l;
    ++listener_token_;
    l.key = key.toString();
    l.token = listener_token_;
    l.req = req;
    l.cb = cb;
    l.filterChain = std::move(filterChain);
    l.thread = std::thread([=]()
        {
            auto settings = std::make_shared<restbed::Settings>();
            std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
            settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds.

            struct State {
                bool ok {true};
                bool cancel {false};
            };
            auto state = std::make_shared<State>();
            restbed::Http::async(req,
                [this, filterChain, cb, state](const std::shared_ptr<restbed::Request>& req,
                                            const std::shared_ptr<restbed::Response>& reply) {
                auto code = reply->get_status_code();

                if (code == 200) {
                    try {
                        while (restbed::Http::is_open(req) and not state->cancel) {
                            restbed::Http::fetch("\n", reply);
                            if (state->cancel)
                                break;
                            std::string body;
                            reply->get_body(body);
                            reply->set_body(""); // Reset the body for the next fetch

                            Json::Value json;
                            Json::Reader reader;
                            if (reader.parse(body, json)) {
                                auto value = std::make_shared<Value>(json);
                                if ((not filterChain or filterChain(*value)) && cb)  {
                                    std::lock_guard<std::mutex> lock(lockCallbacks);
                                    callbacks_.emplace_back([cb, value, state]() {
                                        if (not state->cancel and not cb({value})) {
                                            state->cancel = true;
                                        }
                                    });
                                }
                            } else {
                                state->ok = false;
                            }
                        }
                    } catch (std::runtime_error&) {
                        state->ok = false;
                    }
                } else {
                    state->ok = false;
                }
            }, settings).get();
            if (not state->ok) {
                getConnectivityStatus();
            }
        }
    );
    {
        std::lock_guard<std::mutex> lock(lockListener_);
        listeners_.emplace_back(std::move(l));
    }
    return listener_token_;
}

bool
DhtProxyClient::cancelListen(const InfoHash&, size_t token)
{
    std::lock_guard<std::mutex> lock(lockListener_);
    for (auto it = listeners_.begin(); it != listeners_.end(); ++it) {
        auto& listener = *it;
        if (listener.token == token) {
            if (listener.thread.joinable()) {
                // Close connection to stop listener?
                if (listener.req)
                    restbed::Http::close(listener.req);
                listener.thread.join();
                listeners_.erase(it);
                return true;
            }
        }
    }
    return false;
}

void
DhtProxyClient::getConnectivityStatus()
{
    auto proxyInfos = getProxyInfos();
    // NOTE: json["ipvX"] contains NodeStats::toJson()
    try {
        auto goodIpv4 = static_cast<long>(proxyInfos["ipv4"]["good"].asLargestUInt());
        auto dubiousIpv4 = static_cast<long>(proxyInfos["ipv4"]["dubious"].asLargestUInt());
        statusIpv4_ = (goodIpv4 + dubiousIpv4 > 0) ?  NodeStatus::Connected : NodeStatus::Disconnected;

        auto goodIpv6 = static_cast<long>(proxyInfos["ipv6"]["good"].asLargestUInt());
        auto dubiousIpv6 = static_cast<long>(proxyInfos["ipv6"]["dubious"].asLargestUInt());
        statusIpv6_ = (goodIpv6 + dubiousIpv6 > 0) ?  NodeStatus::Connected : NodeStatus::Disconnected;

        myid = InfoHash(proxyInfos["node_id"].asString());
        if (statusIpv4_ == NodeStatus::Disconnected && statusIpv6_ == NodeStatus::Disconnected) {
            const auto& now = scheduler.time();
            scheduler.edit(nextProxyConfirmation, now);
        }
    } catch (...) {
        statusIpv4_ = NodeStatus::Disconnected;
        statusIpv6_ = NodeStatus::Disconnected;
        const auto& now = scheduler.time();
        scheduler.edit(nextProxyConfirmation, now);
    }
}

void
DhtProxyClient::restartListeners()
{
    std::lock_guard<std::mutex> lock(lockListener_);
    for (auto& listener: listeners_) {
        if (listener.thread.joinable())
            listener.thread.join();
        // Redo listen
        auto filterChain = listener.filterChain;
        auto cb = listener.cb;
        restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + listener.key);
        auto req = std::make_shared<restbed::Request>(uri);
        req->set_method("LISTEN");
        listener.req = req;
        listener.thread = std::thread([this, filterChain, cb, req]()
            {
                auto settings = std::make_shared<restbed::Settings>();
                std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
                settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds.

                restbed::Http::async(req,
                    [this, filterChain, cb](const std::shared_ptr<restbed::Request>& req,
                                     const std::shared_ptr<restbed::Response>& reply) {
                    auto code = reply->get_status_code();

                    if (code == 200) {
                        try {
                            while (restbed::Http::is_open(req)) {
                                restbed::Http::fetch("\n", reply);
                                std::string body;
                                reply->get_body(body);
                                reply->set_body(""); // Reset the body for the next fetch

                                Json::Value json;
                                Json::Reader reader;
                                if (reader.parse(body, json)) {
                                    auto value = std::make_shared<Value>(json);
                                    if ((not filterChain or filterChain(*value)) && cb) {
                                        auto okCb = std::make_shared<std::promise<bool>>();
                                        auto futureCb = okCb->get_future();
                                        {
                                            std::lock_guard<std::mutex> lock(lockCallbacks);
                                            callbacks_.emplace_back([cb, value, okCb](){
                                                okCb->set_value(cb({value}));
                                            });
                                        }
                                        futureCb.wait();
                                        if (!futureCb.get()) {
                                            return;
                                        }
                                    }
                                }
                            }
                        } catch (std::runtime_error&) {
                            // NOTE: Http::close() can occurs here. Ignore this.
                        }

                    } else {
                        this->statusIpv4_ = NodeStatus::Disconnected;
                        this->statusIpv6_ = NodeStatus::Disconnected;
                    }
                }, settings).get();
                getConnectivityStatus();
            }
        );
    }
}


} // namespace dht

#endif // OPENDHT_PROXY_CLIENT