// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
// SPDX-License-Identifier: MIT

#include "dht_proxy_client.h"
#include "dhtrunner.h"
#include "op_cache.h"
#include "utils.h"

#include <llhttp.h>
#include <deque>

namespace dht {

struct DhtProxyClient::InfoState
{
    std::atomic_uint ipv4 {0}, ipv6 {0};
    std::atomic_bool cancel {false};
};

struct DhtProxyClient::OperationState
{
    std::atomic_bool ok {true};
    std::atomic_bool stop {false};
};

struct DhtProxyClient::Listener
{
    Listener(OpValueCache&& c)
        : cache(std::move(c))
    {}

    OpValueCache cache;
    CacheValueCallback cb;
    Sp<OperationState> opstate;
    std::shared_ptr<http::Request> request;
#ifdef OPENDHT_PUSH_NOTIFICATIONS
    std::unique_ptr<asio::steady_timer> refreshSubscriberTimer;
#endif
};

struct PermanentPut
{
    PermanentPut(const Sp<Value>& v, std::unique_ptr<asio::steady_timer>&& j, const Sp<std::atomic_bool>& o)
        : value(v)
        , refreshPutTimer(std::move(j))
        , ok(o)
    {}

    Sp<Value> value;
    std::unique_ptr<asio::steady_timer> refreshPutTimer;
    Sp<std::atomic_bool> ok;
};

struct DhtProxyClient::ProxySearch
{
    SearchCache ops {};
    std::unique_ptr<asio::steady_timer> opExpirationTimer;
    std::map<size_t, Listener> listeners {};
    std::map<Value::Id, PermanentPut> puts {};
    std::set<Sp<Value>> pendingPuts {};
};

struct LineSplit
{
    void append(const char* d, size_t l) { buf_.insert(buf_.end(), d, d + l); }
    bool getLine(char c)
    {
        auto it = buf_.begin();
        while (it != buf_.end()) {
            if (*(it++) == c) {
                line_.clear();
                line_.insert(line_.end(), buf_.begin(), it);
                buf_.erase(buf_.begin(), it);
                return true;
            }
        }
        return false;
    }
    const std::string& line() const { return line_; }

private:
    std::deque<char> buf_ {};
    std::string line_ {};
};

std::string
getRandomSessionId(size_t length = 8)
{
    static constexpr const char chars[]
        = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789!#$%&()*+,./:;<=>?@[]^_`{|}~";
    std::string str(length, 0);
    std::random_device rdev;
    std::uniform_int_distribution<> dist(0, (sizeof(chars) / sizeof(char)) - 2);
    std::generate_n(str.begin(), length, [&] { return chars[dist(rdev)]; });
    return str;
}

DhtProxyClient::DhtProxyClient() {}

DhtProxyClient::DhtProxyClient(std::shared_ptr<dht::crypto::Certificate> serverCA,
                               dht::crypto::Identity clientIdentity,
                               std::function<void()> signal,
                               const std::string& serverHost,
                               const std::string& userAgent,
                               const std::string& pushClientId,
                               const std::string& pushToken,
                               const std::string& pushTopic,
                               const std::string& pushPlatform,
                               std::shared_ptr<dht::Logger> logger)
    : DhtInterface(logger)
    , proxyUrl_(serverHost)
    , clientIdentity_(clientIdentity)
    , serverCertificate_(serverCA)
    , userAgent_(userAgent)
    , pushClientId_(pushClientId)
    , pushSessionId_(getRandomSessionId())
    , deviceKey_(pushToken)
    , notificationTopic_(pushTopic)
    , platform_(pushPlatform)
    , loopSignal_(signal)
    , jsonReader_(Json::CharReaderBuilder {}.newCharReader())
{
    localAddrv4_.setFamily(AF_INET);
    localAddrv6_.setFamily(AF_INET6);

    jsonBuilder_["commentStyle"] = "None";
    jsonBuilder_["indentation"] = "";
    if (logger_) {
        if (serverCertificate_)
            logger_->debug("[proxy:client] using ca certificate for ssl:\n{}",
                           serverCertificate_->toString(false /*chain*/));
        if (clientIdentity_.first and clientIdentity_.second)
            logger_->debug("[proxy:client] using client certificate for ssl:\n{}",
                           clientIdentity_.second->toString(false /*chain*/));
    }
    // run http client
    httpClientThread_ = std::thread([this]() {
        try {
            if (logger_)
                logger_->debug("[proxy:client] starting io_context");
            // Ensures the httpContext_ won't run out of work
            auto work = asio::make_work_guard(httpContext_);
            httpContext_.run();
            if (logger_)
                logger_->debug("[proxy:client] http client io_context stopped");
        } catch (const std::exception& ex) {
            if (logger_)
                logger_->error("[proxy:client] run error: {}", ex.what());
        }
    });
    if (!proxyUrl_.empty())
        startProxy();
}

void
DhtProxyClient::startProxy()
{
    if (proxyUrl_.empty())
        return;

    if (logger_)
        logger_->debug("[proxy:client] start proxy with {}", proxyUrl_);

    nextProxyConfirmationTimer_ = std::make_unique<asio::steady_timer>(httpContext_, std::chrono::steady_clock::now());
    nextProxyConfirmationTimer_->async_wait(std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1));

    listenerRestartTimer_ = std::make_unique<asio::steady_timer>(httpContext_);

    loopSignal_();
}

void
DhtProxyClient::handleProxyConfirm(const asio::error_code& ec)
{
    if (ec == asio::error::operation_aborted)
        return;
    else if (ec) {
        if (logger_)
            logger_->error("[proxy:client] confirm error: {}", ec.message());
        return;
    }
    if (proxyUrl_.empty())
        return;
    getConnectivityStatus();
}

DhtProxyClient::~DhtProxyClient()
{
    stop();
}

void
DhtProxyClient::stop()
{
    if (not isDestroying_.exchange(true)) {
        std::shared_ptr<http::Resolver> resolver;
        {
            std::lock_guard l(resolverLock_);
            resolver = std::move(resolver_);
        }
        if (resolver)
            resolver->cancel();
        cancelAllListeners();
        if (infoState_)
            infoState_->cancel = true;
        {
            std::lock_guard lock(requestLock_);
            for (auto& request : requests_)
                request.second->cancel();
        }
        if (not httpContext_.stopped())
            httpContext_.stop();
        if (httpClientThread_.joinable())
            httpClientThread_.join();
        requests_.clear();
    }
}

std::vector<Sp<Value>>
DhtProxyClient::getLocal(const InfoHash& k, const Value::Filter& filter) const
{
    std::lock_guard lock(searchLock_);
    auto s = searches_.find(k);
    if (s == searches_.end())
        return {};
    return s->second.ops.get(filter);
}

Sp<Value>
DhtProxyClient::getLocalById(const InfoHash& k, Value::Id id) const
{
    std::lock_guard lock(searchLock_);
    auto s = searches_.find(k);
    if (s == searches_.end())
        return {};
    return s->second.ops.get(id);
}

void
DhtProxyClient::cancelAllListeners()
{
    std::lock_guard lock(searchLock_);
    if (logger_)
        logger_->debug("[proxy:client] [listeners] [{} searches] cancel all", searches_.size());
    for (auto& s : searches_) {
        s.second.ops.cancelAll([&](size_t token) {
            auto l = s.second.listeners.find(token);
            if (l == s.second.listeners.end())
                return;
            l->second.opstate->stop.store(true);
            l->second.request->cancel();
            // implicit request.reset()
            s.second.listeners.erase(token);
        });
    }
}

void
DhtProxyClient::shutdown(ShutdownCallback cb, bool)
{
    stop();
    if (cb)
        cb();
}

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

bool
DhtProxyClient::isRunning(sa_family_t af) const
{
    std::lock_guard l(lockCurrentProxyInfos_);
    switch (af) {
    case AF_INET:
        return statusIpv4_ != NodeStatus::Disconnected;
    case AF_INET6:
        return statusIpv6_ != NodeStatus::Disconnected;
    default:
        return false;
    }
}

time_point
DhtProxyClient::periodic(const uint8_t*, size_t, SockAddr, const time_point& /*now*/)
{
    // Exec all currently stored callbacks
    decltype(callbacks_) callbacks;
    {
        std::lock_guard lock(lockCallbacks_);
        callbacks = std::move(callbacks_);
    }
    for (auto& callback : callbacks)
        callback();
    callbacks.clear();
    return time_point::max();
}

void
DhtProxyClient::setHeaderFields(http::Request& request)
{
    request.set_header_field(restinio::http_field_t::accept, "*/*");
    request.set_header_field(restinio::http_field_t::content_type, "application/json");
}

void
DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Value::Filter&& f, Where&& w)
{
    if (logger_)
        logger_->debug("[proxy:client] [get] [search {}]", key.to_view());

    if (isDestroying_) {
        if (donecb)
            donecb(false, {});
        return;
    }
    try {
        auto request = buildRequest("/key/" + key.toString());
        auto reqid = request->id();
        // request->set_connection_type(restinio::http_connection_header_t::keep_alive);
        request->set_method(restinio::http_method_get());
        setHeaderFields(*request);

        auto opstate = std::make_shared<OperationState>();

        request->add_on_body_callback([this,
                                       key,
                                       opstate,
                                       filter = Value::Filter::chain(std::move(f), w.getFilter()),
                                       rxBuf = std::make_shared<LineSplit>(),
                                       cb](const char* at, size_t length) {
            try {
                rxBuf->append(at, length);
                // one value per body line
                std::vector<Sp<Value>> values;
                while (rxBuf->getLine('\n') and !opstate->stop) {
                    std::string err;
                    Json::Value json;
                    const auto& line = rxBuf->line();
                    if (!jsonReader_->parse(line.data(), line.data() + line.size(), &json, &err)) {
                        opstate->ok.store(false);
                        return;
                    }
                    auto value = std::make_shared<Value>(json);
                    if ((not filter or filter(*value)) and cb)
                        values.emplace_back(std::move(value));
                }
                if (not values.empty() and cb) {
                    {
                        std::lock_guard lock(lockCallbacks_);
                        callbacks_.emplace_back([opstate, cb, values = std::move(values)]() {
                            if (not opstate->stop.load() and not cb(values)) {
                                opstate->stop.store(true);
                            }
                        });
                    }
                    loopSignal_();
                }
            } catch (const std::exception& e) {
                if (logger_)
                    logger_->error("[proxy:client] [get {}] body parsing error: {}", key.to_view(), e.what());
                opstate->ok.store(false);
            }
        });
        request->add_on_done_callback([this, reqid, opstate, donecb, key](const http::Response& response) {
            if (response.status_code != 200) {
                if (logger_)
                    logger_->error("[proxy:client] [get {}] failed with code={}", key.to_view(), response.status_code);
                opstate->ok.store(false);
                if (not response.aborted and response.status_code == 0)
                    opFailed();
            }
            if (donecb) {
                {
                    std::lock_guard lock(lockCallbacks_);
                    callbacks_.emplace_back([donecb, opstate]() {
                        donecb(opstate->ok, {});
                        opstate->stop.store(true);
                    });
                }
                loopSignal_();
            }
            if (not isDestroying_) {
                std::lock_guard l(requestLock_);
                requests_.erase(reqid);
            }
        });
        {
            std::lock_guard l(requestLock_);
            requests_[reqid] = request;
        }
        request->send();
    } catch (const std::exception& e) {
        if (logger_)
            logger_->error("[proxy:client] [get {}] error: {}", key.to_view(), e.what());
    }
}

void
DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_point created, bool permanent)
{
    if (not val or isDestroying_) {
        if (cb)
            cb(false, {});
        return;
    }
    if (logger_)
        logger_->debug("[proxy:client] [put] [search {}]", key.to_view());

    std::shared_ptr<std::atomic_bool> ok;
    if (permanent) {
        std::lock_guard lock(searchLock_);
        ok = std::make_shared<std::atomic_bool>(true);
        auto& search = searches_[key];
        if (val->id) {
            auto id = val->id;
            auto refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_,
                                                                        proxy::OP_TIMEOUT - proxy::OP_MARGIN);
            refreshPutTimer->async_wait(
                std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id));
            search.puts.erase(id);
            search.puts.emplace(std::piecewise_construct,
                                std::forward_as_tuple(id),
                                std::forward_as_tuple(val, std::move(refreshPutTimer), ok));
        } else {
            search.pendingPuts.emplace(val);
        }
    }
    doPut(
        key,
        val,
        [this, cb, ok](bool result) {
            if (ok)
                *ok = result;
            if (cb) {
                std::lock_guard lock(lockCallbacks_);
                callbacks_.emplace_back([cb, result]() { cb(result, {}); });
            }
            loopSignal_();
        },
        created,
        permanent);
}

void
DhtProxyClient::handleRefreshPut(const asio::error_code& ec, InfoHash key, Value::Id id)
{
    if (ec == asio::error::operation_aborted)
        return;
    else if (ec) {
        if (logger_)
            logger_->error("[proxy:client] [put] [refresh {}] {}", key.to_view(), ec.message());
        return;
    }
    if (logger_)
        logger_->debug("[proxy:client] [put] [refresh {}]", key.to_view());
    std::lock_guard lock(searchLock_);
    auto search = searches_.find(key);
    if (search != searches_.end()) {
        auto p = search->second.puts.find(id);
        if (p != search->second.puts.end()) {
            doPut(key, p->second.value, [ok = p->second.ok](bool result) { *ok = result; }, time_point::max(), true);
            p->second.refreshPutTimer->expires_after(proxy::OP_TIMEOUT - proxy::OP_MARGIN);
            p->second.refreshPutTimer->async_wait(
                std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id));
        }
    }
}

std::shared_ptr<http::Request>
DhtProxyClient::buildRequest(const std::string& target)
{
    std::unique_lock l(resolverLock_);
    auto resolver = resolver_;
    l.unlock();
    if (not resolver)
        resolver = std::make_shared<http::Resolver>(httpContext_, proxyUrl_, logger_);
    auto request = target.empty() ? std::make_shared<http::Request>(httpContext_, resolver)
                                  : std::make_shared<http::Request>(httpContext_, resolver, target);
    if (serverCertificate_)
        request->set_certificate_authority(serverCertificate_);
    if (clientIdentity_.first and clientIdentity_.second)
        request->set_identity(clientIdentity_);
    request->set_header_field(restinio::http_field_t::user_agent, userAgent_);
    return request;
}

void
DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallbackSimple cb, time_point /*created*/, bool permanent)
{
    if (logger_)
        logger_->debug("[proxy:client] [put] [search {}] executing for {}", key, val->toString());

    try {
        auto request = buildRequest("/key/" + key.toString());
        auto reqid = request->id();
        request->set_method(restinio::http_method_post());
        setHeaderFields(*request);

        auto json = val->toJson();
        if (permanent) {
            if (deviceKey_.empty()) {
                json["permanent"] = true;
            } else {
#ifdef OPENDHT_PUSH_NOTIFICATIONS
                Json::Value refresh;
                getPushRequest(refresh);
                json["permanent"] = refresh;
#else
                json["permanent"] = true;
#endif
            }
        }
        request->set_body(Json::writeString(jsonBuilder_, json));
        request->add_on_done_callback([this, reqid, cb, val, key, permanent](const http::Response& response) {
            bool ok = response.status_code == 200;
            if (ok) {
                if (val->id == Value::INVALID_ID) {
                    std::string err;
                    Json::Value parsedValue;
                    if (jsonReader_->parse(response.body.data(),
                                           response.body.data() + response.body.size(),
                                           &parsedValue,
                                           &err)) {
                        auto id = dht::Value(parsedValue).id;
                        val->id = id;
                        if (permanent) {
                            std::lock_guard lock(searchLock_);
                            auto& search = searches_[key];
                            auto it = search.pendingPuts.find(val);
                            if (it != search.pendingPuts.end()) {
                                auto sok = std::make_shared<std::atomic_bool>(ok);
                                auto refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_,
                                                                                            proxy::OP_TIMEOUT
                                                                                                - proxy::OP_MARGIN);
                                refreshPutTimer->async_wait(
                                    std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id));
                                search.puts.emplace(std::piecewise_construct,
                                                    std::forward_as_tuple(id),
                                                    std::forward_as_tuple(val, std::move(refreshPutTimer), sok));
                                search.pendingPuts.erase(it);
                            }
                        }
                    } else {
                        if (logger_)
                            logger_->error("[proxy:client] [put] failed to parse value from server: {}", err);
                    }
                }
            } else {
                if (logger_)
                    logger_->error("[proxy:client] [put] failed with code={:d}", response.status_code);
                if (not response.aborted and response.status_code == 0)
                    opFailed();
            }
            if (cb)
                cb(ok);
            if (not isDestroying_) {
                std::lock_guard l(requestLock_);
                requests_.erase(reqid);
            }
        });
        {
            std::lock_guard l(requestLock_);
            requests_[reqid] = request;
        }
        request->send();
    } catch (const std::exception& e) {
        if (logger_)
            logger_->error("[proxy:client] [put {}] error: {}", key, e.what());
        opFailed();
    }
}

/**
 * Get data currently being put at the given hash.
 */
std::vector<Sp<Value>>
DhtProxyClient::getPut(const InfoHash& key) const
{
    std::vector<Sp<Value>> ret;
    auto search = searches_.find(key);
    if (search != searches_.end()) {
        ret.reserve(search->second.puts.size());
        for (const auto& put : search->second.puts)
            ret.emplace_back(put.second.value);
    }
    return ret;
}

/**
 * Get data currently being put at the given hash with the given id.
 */
Sp<Value>
DhtProxyClient::getPut(const InfoHash& key, const Value::Id& id) const
{
    auto search = searches_.find(key);
    if (search == searches_.end())
        return {};
    auto val = search->second.puts.find(id);
    if (val == search->second.puts.end())
        return {};
    return val->second.value;
}

/**
 * Stop any put/announce operation at the given location,
 * for the value with the given id.
 */
bool
DhtProxyClient::cancelPut(const InfoHash& key, const Value::Id& id)
{
    auto search = searches_.find(key);
    if (search == searches_.end())
        return false;
    if (logger_)
        logger_->debug("[proxy:client] [put] [search {}] cancel", key.to_view());
    return search->second.puts.erase(id) > 0;
}

NodeStats
DhtProxyClient::getNodesStats(sa_family_t af) const
{
    return af == AF_INET ? stats4_ : stats6_;
}

void
DhtProxyClient::getProxyInfos()
{
    if (logger_)
        logger_->debug("[proxy:client] [info] requesting proxy server node information");
    auto infoState = std::make_shared<InfoState>();
    {
        std::lock_guard l(lockCurrentProxyInfos_);
        if (infoState_)
            infoState_->cancel = true;
        infoState_ = infoState;
        if (statusIpv4_ == NodeStatus::Disconnected)
            statusIpv4_ = NodeStatus::Connecting;
        if (statusIpv6_ == NodeStatus::Disconnected)
            statusIpv6_ = NodeStatus::Connecting;
    }
    if (logger_)
        logger_->debug("[proxy:client] [status] sending request");

    auto resolver = std::make_shared<http::Resolver>(httpContext_, proxyUrl_, logger_);
    queryProxyInfo(infoState, resolver, AF_INET);
    queryProxyInfo(infoState, resolver, AF_INET6);
    std::lock_guard l(resolverLock_);
    resolver_ = resolver;
}

void
DhtProxyClient::queryProxyInfo(const Sp<InfoState>& infoState, const Sp<http::Resolver>& resolver, sa_family_t family)
{
    if (logger_)
        logger_->debug("[proxy:client] [status] query ipv{} info", family == AF_INET ? 4 : 6);
    try {
        auto request = std::make_shared<http::Request>(httpContext_, resolver, family);
        if (serverCertificate_)
            request->set_certificate_authority(serverCertificate_);
        auto reqid = request->id();
        request->set_method(restinio::http_method_get());
        setHeaderFields(*request);
        request->add_on_done_callback([this, reqid, family, infoState](const http::Response& response) {
            if (infoState->cancel.load())
                return;
            if (response.status_code != 200) {
                if (logger_)
                    logger_->error("[proxy:client] [status] ipv{} failed with code={}",
                                   family == AF_INET ? 4 : 6,
                                   response.status_code);
                // pass along the failures
                if ((family == AF_INET and infoState->ipv4 == 0) or (family == AF_INET6 and infoState->ipv6 == 0))
                    onProxyInfos(Json::Value {}, family);
            } else {
                std::string err;
                Json::Value proxyInfos;
                if (!jsonReader_->parse(response.body.data(),
                                        response.body.data() + response.body.size(),
                                        &proxyInfos,
                                        &err)) {
                    onProxyInfos(Json::Value {}, family);
                } else if (not infoState->cancel) {
                    if (auto req = response.request.lock()) {
                        if (auto conn = req->get_connection()) {
                            const auto& localAddr = conn->local_address();
                            proxyInfos["local_ip"] = localAddr.to_string();
                        }
                    }
                    onProxyInfos(proxyInfos, family);
                }
            }
            if (not isDestroying_) {
                std::lock_guard l(requestLock_);
                requests_.erase(reqid);
            }
        });

        if (infoState->cancel.load())
            return;
        {
            std::lock_guard l(requestLock_);
            requests_[reqid] = request;
        }
        request->send();
    } catch (const std::exception& e) {
        if (logger_)
            logger_->error("[proxy:client] [status] error sending request: {}", e.what());
    }
}

void
DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t family)
{
    if (isDestroying_)
        return;
    std::unique_lock l(lockCurrentProxyInfos_);
    auto oldStatus = std::max(statusIpv4_, statusIpv6_);
    auto& status = family == AF_INET ? statusIpv4_ : statusIpv6_;
    auto& pubAddress = family == AF_INET ? publicAddressV4_ : publicAddressV6_;
    auto& localAddress = family == AF_INET ? localAddrv4_ : localAddrv6_;
    if (not proxyInfos.isMember("node_id")) {
        if (logger_)
            logger_->error("[proxy:client] [info] request failed for {}", family == AF_INET ? "ipv4" : "ipv6");
        status = NodeStatus::Disconnected;
        if (pubAddress) {
            pubAddress = {};
        }
    } else {
        if (logger_)
            logger_->debug("[proxy:client] [info] got proxy reply for {}", family == AF_INET ? "ipv4" : "ipv6");
        try {
            myid = InfoHash(proxyInfos["node_id"].asString());
            stats4_ = NodeStats(proxyInfos["ipv4"]);
            stats6_ = NodeStats(proxyInfos["ipv6"]);
            auto publicIp = parsePublicAddress(proxyInfos["public_ip"]);
            bool pub = pubAddress != publicIp;
            pubAddress = publicIp;

            if (proxyInfos.isMember("local_ip")) {
                std::string localIp = proxyInfos["local_ip"].asString();
                if (localAddress.toString() != localIp) {
                    localAddress.setAddress(localIp.c_str());
                    pub = (bool) localAddress;
                }
            }

            if (pub && publicAddressChangedCb_) {
                std::vector<SockAddr> addresses;
                if (publicAddressV4_)
                    addresses.emplace_back(publicAddressV4_);
                if (publicAddressV6_)
                    addresses.emplace_back(publicAddressV6_);
                std::lock_guard lock(lockCallbacks_);
                callbacks_.emplace_back(
                    [cb = publicAddressChangedCb_, addresses = std::move(addresses)]() { cb(std::move(addresses)); });
            }

            if (stats4_.good_nodes + stats6_.good_nodes)
                status = NodeStatus::Connected;
            else if (stats4_.dubious_nodes + stats6_.dubious_nodes)
                status = NodeStatus::Connecting;
            else
                status = NodeStatus::Disconnected;

        } catch (const std::exception& e) {
            if (logger_)
                logger_->error("[proxy:client] [info] error processing: {}", e.what());
        }
    }
    auto newStatus = std::max(statusIpv4_, statusIpv6_);
    if (newStatus == NodeStatus::Connected) {
        if (oldStatus == NodeStatus::Disconnected || oldStatus == NodeStatus::Connecting || launchConnectedCbs_) {
            launchConnectedCbs_ = false;
            listenerRestartTimer_->expires_at(std::chrono::steady_clock::now());
            listenerRestartTimer_->async_wait(std::bind(&DhtProxyClient::restartListeners, this, std::placeholders::_1));
            if (not onConnectCallbacks_.empty()) {
                std::lock_guard lock(lockCallbacks_);
                callbacks_.emplace_back([cbs = std::move(onConnectCallbacks_)]() mutable {
                    while (not cbs.empty()) {
                        cbs.front()();
                        cbs.pop();
                    }
                });
            }
        }
        nextProxyConfirmationTimer_->expires_at(std::chrono::steady_clock::now() + std::chrono::minutes(15));
        nextProxyConfirmationTimer_->async_wait(
            std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1));
    } else if (newStatus == NodeStatus::Disconnected) {
        nextProxyConfirmationTimer_->expires_at(std::chrono::steady_clock::now() + std::chrono::minutes(1));
        nextProxyConfirmationTimer_->async_wait(
            std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1));
    }
    l.unlock();
    loopSignal_();
}

SockAddr
DhtProxyClient::parsePublicAddress(const Json::Value& val)
{
    auto public_ip = val.asString();
    auto [host, service] = splitPort(public_ip);
    auto sa = SockAddr::resolve(std::string(host));
    if (sa.empty())
        return {};
    return sa.front().getMappedIPv4();
}

std::vector<SockAddr>
DhtProxyClient::getPublicAddress(sa_family_t family)
{
    std::lock_guard l(lockCurrentProxyInfos_);
    std::vector<SockAddr> result;
    if (publicAddressV6_ && family != AF_INET)
        result.emplace_back(publicAddressV6_);
    if (publicAddressV4_ && family != AF_INET6)
        result.emplace_back(publicAddressV4_);
    return result;
}

size_t
DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filter, Where where)
{
    if (logger_)
        logger_->debug("[proxy:client] [listen] [search {}]", key.to_view());
    if (isDestroying_)
        return 0;

    std::lock_guard lock(searchLock_);
    auto& search = searches_[key];
    auto query = std::make_shared<Query>(Select {}, std::move(where));
    return search.ops
        .listen(cb, query, std::move(filter), [this, key](Sp<Query>, ValueCallback cb, SyncCallback) -> size_t {
            // Find search
            auto search = searches_.find(key);
            if (search == searches_.end()) {
                if (logger_)
                    logger_->error("[proxy:client] [listen] [search {}] search not found", key.to_view());
                return 0;
            }
            if (logger_)
                logger_->debug("[proxy:client] [listen] [search {}] sending {}",
                               key.to_view(),
                               deviceKey_.empty() ? "listen" : "subscribe");
            // Add listener
            auto token = ++listenerToken_;
            auto l = search->second.listeners.find(token);
            if (l == search->second.listeners.end()) {
                l = search->second.listeners
                        .emplace(std::piecewise_construct,
                                 std::forward_as_tuple(token),
                                 std::forward_as_tuple(std::move(cb)))
                        .first;
            } else {
                if (l->second.opstate)
                    l->second.opstate->stop = true;
            }
            // Add cache callback
            auto opstate = std::make_shared<OperationState>();
            l->second.opstate = opstate;
            l->second.cb = [this, key, token, opstate](const std::vector<Sp<Value>>& values,
                                                       bool expired,
                                                       system_clock::time_point t) {
                if (opstate->stop)
                    return false;
                std::lock_guard lock(searchLock_);
                auto s = searches_.find(key);
                if (s != searches_.end()) {
                    auto l = s->second.listeners.find(token);
                    if (l != s->second.listeners.end()) {
                        return l->second.cache.onValue(values, expired, t);
                    }
                }
                return false;
            };
#ifdef OPENDHT_PUSH_NOTIFICATIONS
            if (not deviceKey_.empty()) {
                /*
                 * Relaunch push listeners even if a timeout is not received
                 * (if the proxy crash for any reason)
                 */
                if (!l->second.refreshSubscriberTimer)
                    l->second.refreshSubscriberTimer = std::make_unique<asio::steady_timer>(httpContext_);
                l->second.refreshSubscriberTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT
                                                             - proxy::OP_MARGIN);
                l->second.refreshSubscriberTimer->async_wait(
                    std::bind(&DhtProxyClient::handleResubscribe, this, std::placeholders::_1, key, token, opstate));
            }
#endif
            ListenMethod method;
            restinio::http_request_header_t header;
            if (deviceKey_.empty()) { // listen
                method = ListenMethod::LISTEN;
                header.method(restinio::http_method_get());
                header.request_target(fmt::format("/key/{}/listen", key.to_view()));
            } else {
                method = ListenMethod::SUBSCRIBE;
                header.method(restinio::http_method_subscribe());
                header.request_target(fmt::format("/key/{}", key.to_view()));
            }
            sendListen(header, l->second.cb, opstate, l->second, method);
            return token;
        });
}

void
DhtProxyClient::handleResubscribe(const asio::error_code& ec,
                                  const InfoHash& key,
                                  const size_t token,
                                  std::shared_ptr<OperationState> opstate)
{
    if (ec == asio::error::operation_aborted)
        return;
    else if (ec) {
        if (logger_)
            logger_->error("[proxy:client] [resubscribe {}] {}", key.to_view(), ec.message());
        return;
    }
    if (opstate->stop)
        return;
    std::lock_guard lock(searchLock_);
    auto s = searches_.find(key);
    if (s != searches_.end()) {
        auto l = s->second.listeners.find(token);
        if (l != s->second.listeners.end()) {
            resubscribe(key, token, l->second);
        } else {
            if (logger_)
                logger_->error("[proxy:client] [resubscribe {}] token not found", key.to_view());
        }
    }
}

bool
DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken)
{
    if (logger_)
        logger_->debug("[proxy:client] [search {}] cancel listen {}", key.to_view(), gtoken);

    std::lock_guard lock(searchLock_);
    // find the listener in cache
    auto it = searches_.find(key);
    if (it == searches_.end())
        return false;
    auto& ops = it->second.ops;
    bool canceled = ops.cancelListen(gtoken, std::chrono::steady_clock::now());

    // define real cancel listen only once
    if (not it->second.opExpirationTimer)
        it->second.opExpirationTimer = std::make_unique<asio::steady_timer>(httpContext_, ops.getExpiration());
    else
        it->second.opExpirationTimer->expires_at(ops.getExpiration());
    it->second.opExpirationTimer->async_wait(
        std::bind(&DhtProxyClient::handleExpireListener, this, std::placeholders::_1, key));
    return canceled;
}

void
DhtProxyClient::handleExpireListener(const asio::error_code& ec, const InfoHash& key)
{
    if (ec == asio::error::operation_aborted)
        return;
    else if (ec) {
        if (logger_)
            logger_->error("[proxy:client] [listen {}] error in cancel: {}", key.to_view(), ec.message());
        return;
    }
    if (logger_)
        logger_->debug("[proxy:client] [listen {}] expire listener", key.to_view());

    std::lock_guard lock(searchLock_);
    auto search = searches_.find(key);
    if (search == searches_.end())
        return;

    // everytime a new expiry is set, a previous gets aborted
    time_point next = search->second.ops.expire(std::chrono::steady_clock::now(), [&](size_t ltoken) {
        auto it = search->second.listeners.find(ltoken);
        if (it == search->second.listeners.end())
            return;

        auto& listener = it->second;
        listener.opstate->stop = true;

        if (not deviceKey_.empty()) {
            // UNSUBSCRIBE
            auto request = buildRequest("/key/" + key.toString());
            auto reqid = request->id();
            try {
                request->set_method(restinio::http_method_unsubscribe());
                setHeaderFields(*request);

                Json::Value body;
                body["key"] = deviceKey_;
                body["client_id"] = pushClientId_;
                request->set_body(Json::writeString(jsonBuilder_, body));
                request->add_on_done_callback([this, reqid, key](const http::Response& response) {
                    if (response.status_code != 200) {
                        if (logger_)
                            logger_->error("[proxy:client] [unsubscribe {}] failed with code={}",
                                           key.to_view(),
                                           response.status_code);
                        if (not response.aborted and response.status_code == 0)
                            opFailed();
                    }
                    if (not isDestroying_) {
                        std::lock_guard l(requestLock_);
                        requests_.erase(reqid);
                    }
                });
                {
                    std::lock_guard l(requestLock_);
                    requests_[reqid] = request;
                }
                request->send();
            } catch (const std::exception& e) {
                if (logger_)
                    logger_->error("[proxy:client] [unsubscribe {}] failed: {}", key.to_view(), e.what());
            }
        } else {
            // stop the request
            if (listener.request) {
                listener.request->cancel();
                listener.request.reset();
            }
        }
        search->second.listeners.erase(it);
        if (logger_)
            logger_->debug("[proxy:client] [listen:cancel] [search {}] {} listener remaining",
                           key.to_view(),
                           search->second.listeners.size());
    });
    if (next != time_point::max()) {
        search->second.opExpirationTimer->expires_at(next);
        search->second.opExpirationTimer->async_wait(
            std::bind(&DhtProxyClient::handleExpireListener, this, std::placeholders::_1, key));
    }
    if (search->second.listeners.empty()) {
        searches_.erase(search);
    }
}

void
DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
                           const CacheValueCallback& cb,
                           const Sp<OperationState>& opstate,
                           Listener& listener,
                           ListenMethod method)
{
    if (logger_)
        logger_->error("[proxy:client] [listen] sendListen: {}", (int) method);
    try {
        auto request = buildRequest();
        listener.request = request;
        auto reqid = request->id();
        request->set_header(header);
        setHeaderFields(*request);
        if (method == ListenMethod::LISTEN)
            request->set_connection_type(restinio::http_connection_header_t::keep_alive);
#ifdef OPENDHT_PUSH_NOTIFICATIONS
        std::string body;
        if (method != ListenMethod::LISTEN)
            body = fillBody(method == ListenMethod::RESUBSCRIBE);
        request->set_body(body);
#endif
        auto rxBuf = std::make_shared<LineSplit>();
        request->add_on_body_callback([this, reqid, opstate, rxBuf, cb](const char* at, size_t length) {
            try {
                auto& b = *rxBuf;
                b.append(at, length);

                // one value per body line
                while (b.getLine('\n') and !opstate->stop) {
                    std::string err;
                    Json::Value json;
                    const auto& line = b.line();
                    if (!jsonReader_->parse(line.data(), line.data() + line.size(), &json, &err)) {
                        opstate->ok.store(false);
                        return;
                    }
                    if (json.size() == 0) { // it's the end
                        break;
                    }

                    auto value = std::make_shared<Value>(json);
                    if (cb) {
                        auto expired = json.get("expired", Json::Value(false)).asBool();
                        {
                            std::lock_guard lock(lockCallbacks_);
                            callbacks_.emplace_back([cb, value, opstate, expired]() {
                                if (not opstate->stop.load()
                                    and not cb({value}, expired, system_clock::time_point::min()))
                                    opstate->stop.store(true);
                            });
                        }
                        loopSignal_();
                    }
                }
            } catch (const std::exception& e) {
                if (logger_)
                    logger_->error("[proxy:client] [listen] request #{} error in parsing: {}", reqid, e.what());
                opstate->ok.store(false);
            }
        });
        request->add_on_done_callback([this, opstate, reqid](const http::Response& response) {
            if (response.status_code != 200) {
                if (logger_)
                    logger_->error("[proxy:client] [listen] send request #{} failed with code={}",
                                   reqid,
                                   response.status_code);
                opstate->ok.store(false);
                if (not response.aborted and response.status_code == 0)
                    opFailed();
            }
            if (not isDestroying_) {
                std::lock_guard l(requestLock_);
                requests_.erase(reqid);
            }
        });
        {
            std::lock_guard l(requestLock_);
            requests_[reqid] = request;
        }
        request->add_on_status_callback([request, seconds = this->listenKeepIdle()](unsigned status_code) {
            if (status_code == 200) {
                // increase TCP_KEEPIDLE to save power
                request->get_connection()->set_keepalive(seconds);
            }
        });
        request->send();
    } catch (const std::exception& e) {
        if (logger_)
            logger_->error("[proxy:client] [listen] request failed: {}", e.what());
    }
}

void
DhtProxyClient::opFailed()
{
    if (isDestroying_)
        return;
    if (logger_)
        logger_->error("[proxy:client] proxy request failed");
    {
        std::lock_guard l(lockCurrentProxyInfos_);
        statusIpv4_ = NodeStatus::Disconnected;
        statusIpv6_ = NodeStatus::Disconnected;
    }
    getConnectivityStatus();
    loopSignal_();
}

void
DhtProxyClient::getConnectivityStatus()
{
    if (logger_)
        logger_->d("[proxy:client] [connectivity] get status");
    if (!isDestroying_)
        getProxyInfos();
}

void
DhtProxyClient::restartListeners(const asio::error_code& ec)
{
    if (ec == asio::error::operation_aborted)
        return;
    else if (ec) {
        if (logger_)
            logger_->e("[proxy:client] restart error: %s", ec.message().c_str());
        return;
    }

    if (isDestroying_)
        return;
    if (logger_)
        logger_->d("[proxy:client] [listeners] refresh permanent puts");

    std::lock_guard lock(searchLock_);
    for (auto& search : searches_) {
        const auto& key = search.first;
        for (auto& put : search.second.puts) {
            doPut(key, put.second.value, [ok = put.second.ok](bool result) { *ok = result; }, time_point::max(), true);
            if (!put.second.refreshPutTimer) {
                put.second.refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_);
            }
            put.second.refreshPutTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT
                                                   - proxy::OP_MARGIN);
            put.second.refreshPutTimer->async_wait(
                std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, put.first));
        }
        // Restart failed pending puts
        for (auto& pput : search.second.pendingPuts) {
            doPut(key, pput, {}, time_point::max(), true);
        }
    }
    if (not deviceKey_.empty()) {
        bool pending = false;
        {
            std::lock_guard l(lockCurrentProxyInfos_);
            pending = pushSubscriptionPending_;
            pushSubscriptionPending_ = false;
        }
        if (pending) {
            // setPushNotificationToken() could not send the subscriptions
            // because the proxy connection was not established yet. Send them
            // now, so the server learns the session id of this run: healthy
            // listeners are skipped by the loop below, and would otherwise
            // never be subscribed for push at all.
            if (logger_)
                logger_->debug("[proxy:client] [listeners] sending deferred push subscriptions");
            for (auto& search : searches_)
                for (auto& listener : search.second.listeners)
                    resubscribe(search.first, listener.first, listener.second);
            return;
        }
        if (logger_)
            logger_->debug("[proxy:client] [listeners] resubscribe due to a connectivity change");
        // Connectivity changed, refresh all subscribe
        for (auto& search : searches_)
            for (auto& listener : search.second.listeners)
                if (!listener.second.opstate->ok)
                    resubscribe(search.first, listener.first, listener.second);
        return;
    }
    if (logger_)
        logger_->debug("[proxy:client] [listeners] restarting listeners");
    for (auto& search : searches_) {
        for (auto& l : search.second.listeners) {
            auto& listener = l.second;
            if (auto opstate = listener.opstate)
                opstate->stop = true;
            listener.request->cancel();
            listener.request.reset();
        }
    }
    for (auto& search : searches_) {
        for (auto& l : search.second.listeners) {
            auto& listener = l.second;
            auto opstate = listener.opstate;
            // Redo listen
            opstate->stop.store(false);
            opstate->ok.store(true);
            auto cb = listener.cb;
            // define header
            restinio::http_request_header_t header;
            header.method(restinio::http_method_get());
            header.request_target("/key/" + search.first.toString() + "/listen");
            sendListen(header, cb, opstate, listener, ListenMethod::LISTEN);
        }
    }
}

PushNotificationResult
DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::map<std::string, std::string>& notification)
{
    auto ret = PushNotificationResult::IgnoredNoOp;
#ifdef OPENDHT_PUSH_NOTIFICATIONS
    {
        // If a push notification is received, the proxy is up and running
        std::lock_guard l(lockCurrentProxyInfos_);
        auto oldStatus = std::max(statusIpv4_, statusIpv6_);
        if (oldStatus != NodeStatus::Connected)
            launchConnectedCbs_ = true;
        statusIpv4_ = NodeStatus::Connected;
        statusIpv6_ = NodeStatus::Connected;
    }
    auto launchLoop = false;
    try {
        auto sessionId = notification.find("s");
        if (sessionId != notification.end() and sessionId->second != pushSessionId_) {
            if (logger_)
                logger_->debug("[proxy:client] [push] ignoring push for other session");
            return PushNotificationResult::IgnoredWrongSession;
        }
        std::lock_guard lock(searchLock_);
        auto timeout = notification.find("timeout");
        if (timeout != notification.cend()) {
            InfoHash key(timeout->second);
            auto& search = searches_.at(key);
            auto vidIt = notification.find("vid");
            if (vidIt != notification.end()) {
                // Refresh put
                auto vid = std::stoull(vidIt->second);
                auto& put = search.puts.at(vid);
                if (!put.refreshPutTimer)
                    put.refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_,
                                                                               std::chrono::steady_clock::now());
                else
                    put.refreshPutTimer->expires_at(std::chrono::steady_clock::now());
                put.refreshPutTimer->async_wait(
                    std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, vid));
                ret = PushNotificationResult::PutRefresh;
            } else {
                // Refresh listen
                for (auto& list : search.listeners)
                    resubscribe(key, list.first, list.second);
                ret = search.listeners.empty() ? PushNotificationResult::IgnoredNoOp
                                               : PushNotificationResult::ListenRefresh;
            }
        } else {
            auto key = InfoHash(notification.at("key"));
            system_clock::time_point sendTime = system_clock::time_point::min();
            try {
                sendTime = system_clock::time_point(std::chrono::milliseconds(std::stoull(notification.at("t"))));
            } catch (...) {
            }
            auto& search = searches_.at(key);
            auto expired = notification.find("exp");
            for (auto& list : search.listeners) {
                if (list.second.opstate->stop)
                    continue;
                if (logger_)
                    logger_->debug("[proxy:client] [push] [search {}] received", key);
                auto token = list.first;
                auto opstate = list.second.opstate;
                if (expired == notification.end()) {
                    auto cb = list.second.cb;
                    auto oldValues = list.second.cache.getValues();
                    get(
                        key,
                        [cb, sendTime](const std::vector<Sp<Value>>& vals) { return cb(vals, false, sendTime); },
                        [cb, oldValues, sendTime](bool /*ok*/) {
                            // Decrement old values refcount to expire values not
                            // present in the new list
                            cb(oldValues, true, sendTime);
                        });
                    ret = PushNotificationResult::Values;
                } else {
                    std::stringstream ss(expired->second);
                    std::vector<Value::Id> ids;
                    while (ss.good()) {
                        std::string substr;
                        getline(ss, substr, ',');
                        ids.emplace_back(std::stoull(substr));
                    }
                    {
                        std::lock_guard lockCb(lockCallbacks_);
                        callbacks_.emplace_back([this, key, token, opstate, ids, sendTime]() {
                            if (opstate->stop)
                                return;
                            std::lock_guard lock(searchLock_);
                            auto s = searches_.find(key);
                            if (s == searches_.end())
                                return;
                            auto l = s->second.listeners.find(token);
                            if (l == s->second.listeners.end())
                                return;
                            if (not opstate->stop and not l->second.cache.onValuesExpired(ids, sendTime))
                                opstate->stop = true;
                        });
                    }
                    launchLoop = true;
                    ret = PushNotificationResult::ValuesExpired;
                }
            }
        }
    } catch (const std::out_of_range& e) {
        ret = PushNotificationResult::IgnoredNoOp;
    } catch (const std::exception& e) {
        if (logger_)
            logger_->error("[proxy:client] [push] receive error: {}", e.what());
        ret = PushNotificationResult::Error;
    }
    if (launchLoop)
        loopSignal_();
#endif
    return ret;
}

void
DhtProxyClient::resubscribe([[maybe_unused]] const InfoHash& key,
                            [[maybe_unused]] const size_t token,
                            [[maybe_unused]] Listener& listener)
{
#ifdef OPENDHT_PUSH_NOTIFICATIONS
    if (deviceKey_.empty())
        return;
    if (logger_)
        logger_->debug("[proxy:client] [resubscribe] [search {}]", key);

    auto opstate = listener.opstate;
    listener.request.reset(); // This will update ok to false
    opstate->ok = true;

    restinio::http_request_header_t header;
    header.method(restinio::http_method_subscribe());
    header.request_target("/key/" + key.toString());
    if (!listener.refreshSubscriberTimer) {
        listener.refreshSubscriberTimer = std::make_unique<asio::steady_timer>(httpContext_);
    }
    listener.refreshSubscriberTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
    listener.refreshSubscriberTimer->async_wait(
        std::bind(&DhtProxyClient::handleResubscribe, this, std::placeholders::_1, key, token, opstate));
    auto vcb = listener.cb;
    sendListen(header, vcb, opstate, listener, ListenMethod::RESUBSCRIBE);
#endif
}

#ifdef OPENDHT_PUSH_NOTIFICATIONS
void
DhtProxyClient::getPushRequest(Json::Value& body) const
{
    body["key"] = deviceKey_;
    body["client_id"] = pushClientId_;
    body["session_id"] = pushSessionId_;
    body["topic"] = notificationTopic_;
    body["platform"] = platform_;
}

std::string
DhtProxyClient::fillBody(bool resubscribe)
{
    // Fill body with
    // {
    //   "key":"device_key",
    // }
    Json::Value body;
    getPushRequest(body);
    if (resubscribe) {
        // This is the first listen, we want to retrieve previous values.
        body["refresh"] = true;
    }
    auto content = Json::writeString(jsonBuilder_, body) + "\n";
    std::replace(content.begin(), content.end(), '\n', ' ');
    return content;
}
#endif // OPENDHT_PUSH_NOTIFICATIONS

void
DhtProxyClient::setPushNotificationToken([[maybe_unused]] const std::string& token)
{
#ifdef OPENDHT_PUSH_NOTIFICATIONS
    {
        std::unique_lock l(lockCurrentProxyInfos_);
        if (deviceKey_ == token)
            return;
        deviceKey_ = token;
        if (!(statusIpv4_ == NodeStatus::Connected || statusIpv6_ == NodeStatus::Connected)) {
            // The proxy connection is not up yet, so the subscriptions can't be
            // sent now. Remember that they are owed: clients typically set the
            // token while the account starts, before the first proxy reply has
            // been received, and nothing else would ever push the current
            // session id to the server. The server would then keep notifying
            // this device with the session of a previous run, and every push
            // would be discarded as PushNotificationResult::IgnoredWrongSession.
            pushSubscriptionPending_ = true;
            return;
        }
        pushSubscriptionPending_ = false;
    }
    resubscribeAll();
#endif
}

#ifdef OPENDHT_PUSH_NOTIFICATIONS
void
DhtProxyClient::resubscribeAll()
{
    if (logger_)
        logger_->debug("[proxy:client] [push] resubscribing every listener");
    std::lock_guard lock(searchLock_);
    for (auto& search : searches_) {
        for (auto& listener : search.second.listeners) {
            resubscribe(search.first, listener.first, listener.second);
        }
    }
}
#endif

} // namespace dht