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

#include "dht.h"
#include "rng.h"
#include "search.h"
#include "storage.h"
#include "request.h"

#include <msgpack.hpp>

#include <algorithm>
#include <filesystem>
#include <random>
#include <sstream>
#include <fstream>

#include <inttypes.h>

namespace dht {

using namespace std::placeholders;

constexpr std::chrono::minutes Dht::MAX_STORAGE_MAINTENANCE_EXPIRE_TIME;
constexpr std::chrono::minutes Dht::SEARCH_EXPIRE_TIME;
constexpr std::chrono::seconds Dht::BOOTSTRAP_PERIOD;
constexpr duration Dht::LISTEN_EXPIRE_TIME;
constexpr duration Dht::LISTEN_EXPIRE_TIME_PUBLIC;
constexpr duration Dht::REANNOUNCE_MARGIN;
static constexpr size_t MAX_REQUESTS_PER_SEC {8 * 1024};
static constexpr size_t MAX_CRYPTO_PER_SEC {512};
static constexpr duration BOOTSTRAP_PERIOD_MAX {std::chrono::hours(24)};

NodeStatus
Dht::updateStatus(sa_family_t af)
{
    auto& d = dht(af);
    auto old = d.status;
    d.status = d.getStatus(scheduler.time());
    if (d.status == NodeStatus::Disconnected && bootstrap_pending && bootstrapJob && !bootstrap_nodes.empty())
        d.status = NodeStatus::Connecting;
    if (d.status != old) {
        auto& other = dht(af == AF_INET ? AF_INET6 : AF_INET);
        if (other.status == NodeStatus::Disconnected && d.status == NodeStatus::Disconnected) {
            onDisconnected();
        } else if (other.status == NodeStatus::Connected || d.status == NodeStatus::Connected) {
            onConnected();
        }
    }
    return d.status;
}

NodeStatus
Dht::Kad::getStatus(time_point now) const
{
    unsigned dubious = 0;
    for (const auto& b : buckets) {
        for (auto& n : b.nodes) {
            if (n->isGood(now)) {
                return NodeStatus::Connected;
            } else if (not n->isExpired())
                dubious++;
        }
    }
    auto& ping = pending_pings;
    if (dubious or ping)
        return NodeStatus::Connecting;
    return NodeStatus::Disconnected;
}

void
Dht::shutdown(ShutdownCallback cb, bool stop)
{
    if (not persistPath.empty())
        saveState(persistPath);

    if (stop) {
        for (auto dht : {&dht4, &dht6}) {
            for (auto& sr : dht->searches) {
                for (const auto& r : sr.second->callbacks)
                    r.second.done_cb(false, {});
                sr.second->callbacks.clear();
                for (const auto& a : sr.second->announce) {
                    for (auto& cb : a.callbacks)
                        cb(false, {});
                }
                sr.second->announce.clear();
                sr.second->listeners.clear();
            }
        }
        network_engine.clear();
    }

    if (not maintain_storage) {
        if (cb)
            cb();
        return;
    }

    // Last store maintenance
    scheduler.syncTime();
    auto remaining = std::make_shared<int>(0);
    auto str_donecb = [=](bool, const std::vector<Sp<Node>>&) {
        --*remaining;
        if (logger_)
            logger_->warn("Shutting down node: {} ops remaining", *remaining);
        if (!*remaining && cb) {
            cb();
        }
    };

    for (auto& str : store)
        *remaining += maintainStorage(str, true, str_donecb);

    if (logger_)
        logger_->warn("Shutting down node: after storage, {} ops", *remaining);

    if (!*remaining) {
        if (cb)
            cb();
    }
}

bool
Dht::isRunning(sa_family_t af) const
{
    return network_engine.isRunning(af);
}

/* Every bucket contains an unordered list of nodes. */
const Sp<Node>
Dht::findNode(const InfoHash& id, sa_family_t af) const
{
    if (const Bucket* b = findBucket(id, af))
        for (const auto& n : b->nodes)
            if (n->id == id)
                return n;
    return {};
}

/* Every bucket caches the address of a likely node.  Ping it. */
void
Dht::sendCachedPing(Bucket& b)
{
    if (b.cached)
        if (logger_)
            logger_->debug("[node {}] Sending ping to cached node", b.cached->toString());
    b.sendCachedPing(network_engine);
}

std::vector<SockAddr>
Dht::getPublicAddress(sa_family_t family)
{
    std::vector<SockAddr> ret;
    if (family == AF_UNSPEC) {
        auto& d4 = dht(AF_INET).reported_addr;
        auto& d6 = dht(AF_INET6).reported_addr;
        ret.reserve(d4.size() + d6.size());
        for (const auto& a : d4)
            ret.emplace_back(a.second);
        for (const auto& a : d6)
            ret.emplace_back(a.second);
        return ret;
    }
    auto& reported_addr = dht(family).reported_addr;
    ret.reserve(!family ? reported_addr.size() : reported_addr.size() / 2);
    for (const auto& addr : reported_addr)
        if (!family || family == addr.second.getFamily())
            ret.emplace_back(addr.second);
    return ret;
}

std::pair<size_t, size_t>
Dht::getLocalStoreSize() const
{
    return {local_store_quota->size(), local_store_quota->valueCount()};
}

bool
Dht::trySearchInsert(const Sp<Node>& node)
{
    const auto& now = scheduler.time();
    if (not node)
        return false;

    auto& srs = searches(node->getFamily());
    auto closest = srs.lower_bound(node->id);
    bool inserted {false};

    // insert forward
    for (auto it = closest; it != srs.end(); it++) {
        auto& s = *it->second;
        if (s.insertNode(node, now)) {
            inserted = true;
            scheduler.edit(s.nextSearchStep, now);
        } else if (not s.expired and not s.done)
            break;
    }
    // insert backward
    for (auto it = closest; it != srs.begin();) {
        --it;
        auto& s = *it->second;
        if (s.insertNode(node, now)) {
            inserted = true;
            scheduler.edit(s.nextSearchStep, now);
        } else if (not s.expired and not s.done)
            break;
    }
    return inserted;
}

void
Dht::reportedAddr(const SockAddr& addr)
{
    auto& reported_addr = dht(addr.getFamily()).reported_addr;
    auto firstBefore = reported_addr.empty() ? nullptr : reported_addr.begin()->second.get();
    auto it = std::find_if(reported_addr.begin(), reported_addr.end(), [&](const ReportedAddr& a) {
        return a.second == addr;
    });
    if (it == reported_addr.end()) {
        if (reported_addr.size() < 32)
            reported_addr.emplace_back(1, addr);
    } else
        it->first++;
    std::sort(reported_addr.begin(), reported_addr.end(), [](const ReportedAddr& a, const ReportedAddr& b) {
        return a.first > b.first;
    });
    if (publicAddressChangedCb_) {
        auto firstAfter = reported_addr.begin()->second.get();
        if (firstBefore != firstAfter) {
            auto& otherDht = dht(addr.getFamily() == AF_INET ? AF_INET6 : AF_INET).reported_addr;
            std::vector<SockAddr> v;
            v.reserve(otherDht.empty() ? 1 : 2);
            v.emplace_back(reported_addr.begin()->second);
            if (not otherDht.empty())
                v.emplace_back(otherDht.begin()->second);
            publicAddressChangedCb_(std::move(v));
        }
    }
}

/* We just learnt about a node, not necessarily a new one.  Confirm is 1 if
   the node sent a message, 2 if it sent us a reply. */
void
Dht::onNewNode(const Sp<Node>& node, int confirm)
{
    const auto& now = scheduler.time();
    auto& b = buckets(node->getFamily());
    auto wasEmpty = confirm < 2 && b.grow_time < now - std::chrono::minutes(5);
    if (b.onNewNode(node, confirm, now, myid, network_engine) or confirm) {
        trySearchInsert(node);
        if (wasEmpty) {
            scheduler.edit(nextNodesConfirmation, now + std::chrono::seconds(1));
        }
    }
}

/* Called periodically to purge known-bad nodes.  Note that we're very
   conservative here: broken nodes in the table don't do much harm, we'll
   recover as soon as we find better ones. */
void
Dht::expireBuckets(RoutingTable& list)
{
    for (auto& b : list) {
        bool changed = false;
        b.nodes.remove_if([&changed](const Sp<Node>& n) {
            if (n->isExpired()) {
                changed = true;
                return true;
            }
            return false;
        });
        if (changed)
            sendCachedPing(b);
    }
}

void
Dht::expireSearches()
{
    auto t = scheduler.time() - SEARCH_EXPIRE_TIME;
    auto expired = [&](std::pair<const InfoHash, Sp<Search>>& srp) {
        auto& sr = *srp.second;
        auto b = sr.callbacks.empty() && sr.announce.empty() && sr.listeners.empty() && sr.step_time < t;
        if (b) {
            if (logger_)
                logger_->debug("[search {}] Removing search", srp.first.to_view());
            sr.clear();
            return b;
        } else {
            return false;
        }
    };
    erase_if(dht4.searches, expired);
    erase_if(dht6.searches, expired);
}

void
Dht::searchNodeGetDone(const net::Request& req, net::RequestAnswer&& answer, std::weak_ptr<Search> ws, Sp<Query> query)
{
    const auto& now = scheduler.time();
    if (auto sr = ws.lock()) {
        sr->insertNode(req.node, now, answer.ntoken);
        if (auto srn = sr->getNode(req.node)) {
            /* all other get requests which are satisfied by this answer
               should not be sent anymore */
            for (auto& g : sr->callbacks) {
                auto& q = g.second.query;
                if (q != query and g.second.isSatisfiedBy(*query)) {
                    auto dummy_req = std::make_shared<net::Request>();
                    dummy_req->cancel();
                    srn->getStatus[q] = std::move(dummy_req);
                }
            }
            auto syncTime = srn->getSyncTime(scheduler.time());
            if (srn->syncJob)
                scheduler.edit(srn->syncJob, syncTime);
            else
                srn->syncJob = scheduler.add(syncTime, std::bind(&Dht::searchStep, this, ws));
        }
        onGetValuesDone(req.node, answer, sr, query);
    }
}

void
Dht::searchNodeGetExpired(const net::Request& status, bool over, std::weak_ptr<Search> ws, Sp<Query> query)
{
    if (auto sr = ws.lock()) {
        if (auto srn = sr->getNode(status.node)) {
            srn->candidate = not over;
            if (over)
                srn->getStatus.erase(query);
        }
        scheduler.edit(sr->nextSearchStep, scheduler.time());
    }
}

void
Dht::searchNodeRoutingDone(const net::Request& req, net::RequestAnswer&& answer, std::weak_ptr<Search> ws)
{
    if (auto sr = ws.lock()) {
        if (auto sn = sr->getNode(req.node); sn and sn->routingProbe.get() == &req) {
            sr->insertNode(req.node, scheduler.time(), answer.ntoken);
            scheduler.edit(sr->nextSearchStep, scheduler.time());
        }
    }
}

void
Dht::searchNodeRoutingExpired(const net::Request& req, bool over, std::weak_ptr<Search> ws)
{
    if (auto sr = ws.lock()) {
        if (auto sn = sr->getNode(req.node); sn and sn->routingProbe.get() == &req) {
            sn->candidate = not over;
            if (over)
                sn->routingProbe.reset();
            scheduler.edit(sr->nextSearchStep, scheduler.time());
        }
    }
}

void
Dht::paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n)
{
    auto sr = ws.lock();
    if (not sr)
        return;
    auto select_q = std::make_shared<Query>(Select {}.field(Value::Field::Id), query ? query->where : Where {});
    auto onSelectDone = [this, ws, query](const net::Request& status, net::RequestAnswer&& answer) mutable {
        // Retrieve search
        auto sr = ws.lock();
        if (not sr)
            return;
        const auto& id = sr->id;
        // Retrieve search node
        auto sn = sr->getNode(status.node);
        if (not sn)
            return;
        // backward compatibility
        if (answer.fields.empty()) {
            searchNodeGetDone(status, std::move(answer), ws, query);
            return;
        }
        for (const auto& fvi : answer.fields) {
            try {
                auto vid = fvi->index.at(Value::Field::Id).getInt();
                if (vid == Value::INVALID_ID)
                    continue;
                auto query_for_vid = std::make_shared<Query>(Select {}, Where {}.id(vid));
                sn->pagination_queries[query].push_back(query_for_vid);
                if (logger_)
                    logger_->debug("[search {}] [node {}] Sending {}",
                                   id.toString(),
                                   sn->node->toString(),
                                   query_for_vid->toString());
                sn->getStatus[query_for_vid] = network_engine.sendGetValues(
                    status.node,
                    id,
                    *query_for_vid,
                    -1,
                    std::bind(&Dht::searchNodeGetDone, this, _1, _2, ws, query),
                    std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, query_for_vid));
            } catch (const std::out_of_range&) {
                if (logger_)
                    logger_->error("[search {}] [node {}] Received non-id field in response to "
                                   "'SELECT id' request…",
                                   id.toString(),
                                   sn->node->toString());
            }
        }
    };
    /* add pagination query key for tracking ongoing requests. */
    n->pagination_queries[query].push_back(select_q);

    if (logger_)
        logger_->debug("[search {}] [node {}] Sending {}", sr->id.toString(), n->node->toString(), select_q->toString());
    n->getStatus[select_q] = network_engine.sendGetValues(
        n->node, sr->id, *select_q, -1, onSelectDone, std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, select_q));
}

Dht::SearchNode*
Dht::searchSendGetValues(Sp<Search> sr, SearchNode* pn, bool update)
{
    if (sr->done or sr->currentlySolicitedNodeCount() >= MAX_REQUESTED_SEARCH_NODES)
        return nullptr;

    const auto& now = scheduler.time();

    std::weak_ptr<Search> ws = sr;
    auto cb = sr->callbacks.begin();
    static const Get ANY_GET {time_point::min(), {}, std::make_shared<Query>(Select {}, Where {}, true), {}, {}, {}};
    do { /* for all requests to send */
        SearchNode* n = nullptr;
        const auto& get = sr->callbacks.empty() ? ANY_GET : cb->second;
        const auto& query = get.query;
        const time_point up = (not sr->callbacks.empty() and update) ? sr->getLastGetTime(*query) : time_point::min();

        if (pn and pn->canGet(now, up, get)) {
            n = pn;
        } else {
            for (auto& sn : sr->nodes) {
                if (sn->canGet(now, up, get)) {
                    n = sn.get();
                    break;
                }
            }
        }

        if (sr->callbacks.empty()) { /* 'find_node' request */
            if (not n)
                return nullptr;

            /* if (logger_)
                   logger_->d(sr->id, n->node->id, "[search %s] [node %s] Sending 'find_node'",
                        sr->id.toString().c_str(), n->node->toString().c_str());*/
            n->getStatus[query] = network_engine
                                      .sendFindNode(n->node,
                                                    sr->id,
                                                    -1,
                                                    std::bind(&Dht::searchNodeGetDone, this, _1, _2, ws, query),
                                                    std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, query));

        } else { /* 'get' request */
            if (not n)
                continue;

            if (network_engine.needsSession(*n->node)) {
                n->routingProbe = network_engine
                                      .sendFindNode(n->node,
                                                    sr->id,
                                                    -1,
                                                    std::bind(&Dht::searchNodeRoutingDone, this, _1, _2, ws),
                                                    std::bind(&Dht::searchNodeRoutingExpired, this, _1, _2, ws));
                return n;
            }

            if (query and not query->select.empty()) {
                /* The request contains a select. No need to paginate… */
                /* if (logger_)
                       logger_->d(sr->id, n->node->id, "[search %s] [node %s] Sending 'get'",
                            sr->id.toString().c_str(), n->node->toString().c_str());*/
                n->getStatus[query]
                    = network_engine.sendGetValues(n->node,
                                                   sr->id,
                                                   *query,
                                                   -1,
                                                   std::bind(&Dht::searchNodeGetDone, this, _1, _2, ws, query),
                                                   std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, query));
            } else
                paginate(ws, query, n);
        }

        /* We only attempt to send one request. Return. */
        return n;

    } while (++cb != sr->callbacks.end());

    /* no request were sent */
    return nullptr;
}

void
Dht::searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel)
{
    if (sr->announce.empty())
        return;
    std::weak_ptr<Search> ws = sr;

    auto onDone = [this, ws](const net::Request& req,
                             net::RequestAnswer&& answer) { /* when put done */
                                                            if (auto sr = ws.lock()) {
                                                                onAnnounceDone(req.node, answer, sr);
                                                                scheduler.edit(sr->nextSearchStep, scheduler.time());
                                                            }
    };

    auto onExpired = [this, ws](const net::Request&, bool over) { /* when put expired */
                                                                  if (over)
                                                                      if (auto sr = ws.lock())
                                                                          scheduler.edit(sr->nextSearchStep,
                                                                                         scheduler.time());
    };

    auto onSelectDone = [this, ws, onDone, onExpired](const net::Request& req, net::RequestAnswer&& answer) mutable {
        /* on probing done */
        auto sr = ws.lock();
        if (not sr)
            return;
        const auto& now = scheduler.time();
        sr->insertNode(req.node, now, answer.ntoken);
        auto sn = sr->getNode(req.node);
        if (not sn)
            return;

        if (not sn->isSynced(now)) {
            /* Search is now unsynced. Let's call searchStep to sync again. */
            scheduler.edit(sr->nextSearchStep, now);
            return;
        }
        for (auto& a : sr->announce) {
            if (sn->getAnnounceTime(a.value->id) > now)
                continue;
            bool hasValue {false};
            uint16_t seq_no = 0;
            try {
                const auto& f = std::find_if(answer.fields.cbegin(),
                                             answer.fields.cend(),
                                             [&a](const Sp<FieldValueIndex>& i) {
                                                 return i->index.at(Value::Field::Id).getInt() == a.value->id;
                                             });
                if (f != answer.fields.cend() and *f) {
                    hasValue = true;
                    seq_no = static_cast<uint16_t>((*f)->index.at(Value::Field::SeqNum).getInt());
                }
            } catch (std::out_of_range&) {
            }

            auto next_refresh_time = now + getType(a.value->type).expiration;
            auto& acked = sn->acked[a.value->id];
            scheduler.cancel(acked.refresh);
            /* only put the value if the node doesn't already have it */
            if (not hasValue or seq_no < a.value->seq) {
                if (logger_)
                    logger_->debug("[search {}] [node {}] Sending 'put' (vid: {:016x})",
                                   sr->id.toString(),
                                   sn->node->toString(),
                                   a.value->id);
                auto created = a.permanent ? time_point::max() : a.created;
                acked = {network_engine.sendAnnounceValue(
                             sn->node, sr->id, a.value, created, sn->token.value_or(Blob {}), onDone, onExpired),
                         next_refresh_time};
            } else if (hasValue and a.permanent) {
                if (logger_)
                    logger_->warn("[search {}] [node {}] Sending 'refresh' (vid: {:016x})",
                                  sr->id.toString(),
                                  sn->node->toString(),
                                  a.value->id);
                acked = {network_engine.sendRefreshValue(
                             sn->node,
                             sr->id,
                             a.value->id,
                             sn->token.value_or(Blob {}),
                             onDone,
                             [this,
                              ws,
                              node = sn->node,
                              v = a.value,
                              onDone,
                              onExpired,
                              created = a.permanent ? time_point::max() : a.created,
                              next_refresh_time](const net::Request& /*req*/, net::DhtProtocolException&& e) {
                                 if (e.getCode() == net::DhtProtocolException::NOT_FOUND) {
                                     if (logger_)
                                         logger_->error("[node {}] Returned error 404: storage not found",
                                                        node->toString());
                                     if (auto sr = ws.lock()) {
                                         if (auto sn = sr->getNode(node)) {
                                             sn->acked[v->id] = {network_engine.sendAnnounceValue(sn->node,
                                                                                                  sr->id,
                                                                                                  v,
                                                                                                  created,
                                                                                                  sn->token.value_or(
                                                                                                      Blob {}),
                                                                                                  onDone,
                                                                                                  onExpired),
                                                                 next_refresh_time};
                                             scheduler.edit(sr->nextSearchStep, scheduler.time());
                                             return true;
                                         }
                                     }
                                 }
                                 return false;
                             },
                             onExpired),
                         next_refresh_time};
            } else {
                if (logger_)
                    logger_->warn("[search {}] [node {}] Already has value (vid: {:016x}). Aborting.",
                                  sr->id.toString(),
                                  sn->node->toString(),
                                  a.value->id);
                auto ack_req = std::make_shared<net::Request>(net::Request::State::COMPLETED);
                ack_req->reply_time = now;
                acked = {std::move(ack_req), next_refresh_time};

                /* step to clear announces */
                scheduler.edit(sr->nextSearchStep, now);
            }
            if (a.permanent) {
                acked.refresh = scheduler.add(next_refresh_time - REANNOUNCE_MARGIN,
                                              std::bind(&Dht::searchStep, this, ws));
            }
        }
    };

    static const auto PROBE_QUERY = std::make_shared<Query>(
        Select {}.field(Value::Field::Id).field(Value::Field::SeqNum));

    const auto& now = scheduler.time();
    unsigned i = 0;
    for (auto& np : sr->nodes) {
        auto& n = *np;
        if (not n.isSynced(now))
            continue;

        auto gs = n.probe_query ? n.getStatus.find(n.probe_query) : n.getStatus.end();
        if (gs != n.getStatus.end() and gs->second and gs->second->pending()) {
            continue;
        }

        bool sendQuery = false;
        for (auto& a : sr->announce) {
            if (n.getAnnounceTime(a.value->id) <= now) {
                if (a.permanent) {
                    sendQuery = true;
                } else {
                    if (logger_)
                        logger_->warn("[search {}] [node {}] Sending 'put' (vid: {:016x})",
                                      sr->id.toString(),
                                      n.node->toString(),
                                      a.value->id);
                    n.acked[a.value->id] = {network_engine.sendAnnounceValue(n.node,
                                                                             sr->id,
                                                                             a.value,
                                                                             a.created,
                                                                             n.token.value_or(Blob {}),
                                                                             onDone,
                                                                             onExpired),
                                            now + getType(a.value->type).expiration};
                }
            }
        }

        if (sendQuery) {
            n.probe_query = PROBE_QUERY;
            if (logger_)
                logger_->debug("[search {}] [node {}] Sending {}",
                               sr->id.toString(),
                               n.node->toString(),
                               n.probe_query->toString());
            auto req = network_engine.sendGetValues(n.node,
                                                    sr->id,
                                                    *PROBE_QUERY,
                                                    -1,
                                                    onSelectDone,
                                                    std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, PROBE_QUERY));
            n.getStatus[PROBE_QUERY] = std::move(req);
        }
        if (not n.candidate and ++i == syncLevel)
            break;
    }
}

void
Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
{
    const auto& listenExp = getListenExpiration();
    std::weak_ptr<Search> ws = sr;
    for (const auto& l : sr->listeners) {
        const auto& query = l.second.query;

        auto openListenSocket = [this, ws, query, node = n.node] {
            return node->openSocket([this, ws, query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable {
                /* on new values */
                if (auto sr = ws.lock()) {
                    scheduler.edit(sr->nextSearchStep, scheduler.time());
                    if (auto sn = sr->getNode(node)) {
                        if (answer.ntoken)
                            sn->refreshToken(*answer.ntoken);
                        sn->onValues(query, std::move(answer), types, scheduler);
                    }
                }
            });
        };

        auto r = n.listenStatus.find(query);
        if (n.getListenTime(r, listenExp) > scheduler.time())
            continue;
        // if (logger_)
        //     logger_->d(sr->id, n.node->id, "[search %s] [node %s] Sending 'listen'",
        //        sr->id.toString().c_str(), n.node->toString().c_str());

        if (r == n.listenStatus.end()) {
            r = n.listenStatus
                    .emplace(std::piecewise_construct,
                             std::forward_as_tuple(query),
                             std::forward_as_tuple(l.second.get_cb, l.second.sync_cb, openListenSocket()))
                    .first;
            r->second.cacheExpirationJob = scheduler.add(time_point::max(), [this, ws, query, node = n.node] {
                if (auto sr = ws.lock()) {
                    if (auto sn = sr->getNode(node)) {
                        sn->expireValues(query, scheduler);
                    }
                }
            });
        } else if (not n.node->getSocket(r->second.socketId)) {
            r->second.socketId = openListenSocket();
        }
        auto new_req = network_engine.sendListen(
            n.node,
            sr->id,
            *query,
            n.token.value_or(Blob {}),
            r->second.socketId,
            [this, ws, query](const net::Request& req, net::RequestAnswer&& answer) mutable {
                /* on done */
                if (auto sr = ws.lock()) {
                    scheduler.edit(sr->nextSearchStep, scheduler.time());
                    if (auto sn = sr->getNode(req.node)) {
                        auto job = scheduler.add(sn->getListenTime(query, getListenExpiration()),
                                                 std::bind(&Dht::searchStep, this, ws));
                        sn->onListenSynced(query, true, std::move(job));
                    }
                    onListenDone(req.node, answer, sr);
                }
            },
            [this, ws, query](const net::Request& req, bool over) mutable {
                /* on request expired */
                if (auto sr = ws.lock()) {
                    scheduler.edit(sr->nextSearchStep, scheduler.time());
                    if (over)
                        if (auto sn = sr->getNode(req.node))
                            sn->listenStatus.erase(query);
                }
            });
        // Here the request may have failed and the CachedListenStatus removed
        r = n.listenStatus.find(query);
        if (r != n.listenStatus.end()) {
            r->second.req = new_req;
        }
    }
}

/* When a search is in progress, we periodically call search_step to send
   further requests. */
void
Dht::searchStep(std::weak_ptr<Search> ws)
{
    auto sr = ws.lock();
    if (not sr or sr->expired or sr->done)
        return;

    const auto& now = scheduler.time();
    auto level = sr->syncLevel(now);
    constexpr auto MARGIN = 1;
    bool preSynced = level > MARGIN;
    auto syncLevel = preSynced ? level - MARGIN : 0;
    /*if (auto req_count = sr->currentlySolicitedNodeCount())
        if (logger_)
            logger_->d(sr->id, "[search %s IPv%c] Step (%d requests)",
                sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', req_count);*/
    sr->step_time = now;

    if (sr->refill_time + Node::NODE_EXPIRE_TIME < now and sr->nodes.size() - sr->getNumberOfBadNodes() < SEARCH_NODES)
        refill(*sr);

    /* Check if the first TARGET_NODES (8) live nodes have replied. */
    if (sr->isSynced(now)) {
        if (not(sr->callbacks.empty() and sr->announce.empty())) {
            // search is synced but some (newer) get operations are not complete
            // Call callbacks when done
            std::vector<Get> completed_gets;
            for (auto b = sr->callbacks.begin(); b != sr->callbacks.end();) {
                if (sr->isDone(b->second)) {
                    sr->setDone(b->second);
                    completed_gets.emplace_back(std::move(b->second));
                    b = sr->callbacks.erase(b);
                } else
                    ++b;
            }
            // clear corresponding queries
            for (const auto& get : completed_gets)
                for (auto& sn : sr->nodes) {
                    sn->getStatus.erase(get.query);
                    sn->pagination_queries.erase(get.query);
                }

            /* clearing callbacks for announced values */
            sr->checkAnnounced();

            if (sr->callbacks.empty() && sr->announce.empty() && sr->listeners.empty())
                sr->setDone();
        }

        // true if this node is part of the target nodes cluter.
        /*bool in = sr->id.xorCmp(myid, sr->nodes.back().node->id) < 0;

        logger__DBG("[search %s IPv%c] synced%s",
                sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', in ? ", in" : "");*/

        if (not sr->listeners.empty()) {
            unsigned i = 0;
            for (auto& n : sr->nodes) {
                if (not n->isSynced(now))
                    continue;
                searchSynchedNodeListen(sr, *n);
                if (not n->candidate and ++i == LISTEN_NODES)
                    break;
            }
        }

        // Announce requests
        searchSendAnnounceValue(sr);

        if (sr->callbacks.empty() && sr->announce.empty() && sr->listeners.empty())
            sr->setDone();
    } else if (preSynced) {
        if (not sr->listeners.empty()) {
            auto nl = std::min(syncLevel, LISTEN_NODES);
            unsigned i = 0;
            for (auto& n : sr->nodes) {
                if (not n->isSynced(now))
                    break;
                searchSynchedNodeListen(sr, *n);
                if (++i == nl)
                    break;
            }
        }

        // Announce requests
        searchSendAnnounceValue(sr, syncLevel);
    }

    while (sr->currentlySolicitedNodeCount() < MAX_REQUESTED_SEARCH_NODES and searchSendGetValues(sr))
        ;

    if (sr->getNumberOfConsecutiveBadNodes() >= std::min<size_t>(sr->nodes.size(), SEARCH_MAX_BAD_NODES)) {
        if (logger_)
            logger_->warn("[search {} IPv{}] Expired", sr->id.to_view(), sr->af == AF_INET ? '4' : '6');
        sr->expire();
        if (not public_stable)
            connectivityChanged(sr->af);
    }

    /* dumpSearch(*sr, std::cout); */
}

unsigned
Dht::refill(Dht::Search& sr)
{
    const auto& now = scheduler.time();
    sr.refill_time = now;
    /* we search for up to SEARCH_NODES good nodes. */
    auto cached_nodes = network_engine.getCachedNodes(sr.id, sr.af, SEARCH_NODES);

    if (cached_nodes.empty()) {
        if (logger_)
            logger_->error("[search {} IPv{}] No nodes from cache while refilling search",
                           sr.id.toString(),
                           (sr.af == AF_INET) ? '4' : '6');
        return 0;
    }

    unsigned inserted = 0;
    for (auto& i : cached_nodes) {
        /* try to insert the nodes. Search::insertNode will know how many to insert. */
        if (sr.insertNode(i, now))
            ++inserted;
    }
    if (logger_)
        logger_->debug("[search {} IPv{}] Refilled search with {} nodes from node cache",
                       sr.id.toString(),
                       (sr.af == AF_INET) ? '4' : '6',
                       inserted);
    return inserted;
}

/* Start a search. */
Sp<Dht::Search>
Dht::search(const InfoHash& id,
            sa_family_t af,
            GetCallback gcb,
            QueryCallback qcb,
            DoneCallback dcb,
            Value::Filter f,
            const Sp<Query>& q)
{
    if (!isRunning(af)) {
        if (logger_)
            logger_->error("[search {} IPv{}] Unsupported protocol", id.to_view(), (af == AF_INET) ? '4' : '6');
        if (dcb)
            dcb(false, {});
        return {};
    }

    auto& srs = searches(af);
    const auto& srp = srs.find(id);
    Sp<Search> sr {};

    if (srp != srs.end()) {
        sr = srp->second;
        sr->done = false;
        sr->expired = false;
    } else {
        if (srs.size() < max_searches) {
            sr = std::make_shared<Search>();
            srs.emplace(id, sr);
        } else {
            for (auto it = srs.begin(); it != srs.end(); ++it) {
                auto& s = *it->second;
                if ((s.done or s.expired) and s.announce.empty() and s.listeners.empty()) {
                    sr = it->second;
                    if (sr->nextSearchStep)
                        scheduler.cancel(sr->nextSearchStep);
                    if (sr->opExpirationJob)
                        scheduler.cancel(sr->opExpirationJob);
                    sr->clear();
                    srs.erase(it);
                    break;
                }
            }
            if (not sr) {
                if (logger_)
                    logger_->error("[search {} IPv{}] Maximum number of searches reached!",
                                   id.toString(),
                                   (af == AF_INET) ? '4' : '6');
                if (dcb)
                    dcb(false, {});
                return {};
            }
            srs.emplace(id, sr);
        }
        sr->af = af;
        sr->tid = search_id++;
        sr->step_time = time_point::min();
        sr->id = id;
        sr->done = false;
        sr->expired = false;
        sr->nodes.clear();
        sr->nodes.reserve(SEARCH_NODES + 1);
        sr->nextSearchStep = scheduler.add(time_point::max(),
                                           std::bind(&Dht::searchStep, this, std::weak_ptr<Search>(sr)));
        if (logger_)
            logger_->warn("[search {} IPv{}] New search", id.to_view(), (af == AF_INET) ? '4' : '6');
        if (search_id == 0)
            search_id++;
    }

    sr->get(f, q, qcb, gcb, dcb, scheduler);
    refill(*sr);

    return sr;
}

void
Dht::announce(
    const InfoHash& id, sa_family_t af, Sp<Value> value, DoneCallback callback, time_point created, bool permanent)
{
    auto& srs = searches(af);
    auto srp = srs.find(id);
    if (auto sr = srp == srs.end() ? search(id, af) : srp->second) {
        sr->put(value, callback, created, permanent);
        scheduler.edit(sr->nextSearchStep, scheduler.time());
    } else if (callback) {
        callback(false, {});
    }
}

size_t
Dht::listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filter f, const Sp<Query>& q)
{
    if (!isRunning(af))
        return 0;
    // logger__ERR("[search %s IPv%c] Search_time is now in %lfs", sr->id.toString().c_str(), (sr->af == AF_INET) ? '4'
    // : '6', print_duration(tm-clock::now()));

    // logger__WARN("listenTo %s", id.toString().c_str());
    auto& srs = searches(af);
    auto srp = srs.find(id);
    Sp<Search> sr = (srp == srs.end()) ? search(id, af) : srp->second;
    if (!sr)
        throw DhtException("Unable to create search");
    if (logger_)
        logger_->warn("[search {} IPv{}] Listen", id.to_view(), (af == AF_INET) ? '4' : '6');
    return sr->listen(cb, std::move(f), q, scheduler);
}

size_t
Dht::listen(const InfoHash& id, ValueCallback cb, Value::Filter f, Where where)
{
    if (not id) {
        if (logger_)
            logger_->warn("Listen called with invalid key");
        return 0;
    }
    scheduler.syncTime();

    auto token = ++listener_token;
    auto gcb = OpValueCache::cacheCallback(std::move(cb), [this, id, token] { cancelListen(id, token); });

    auto query = std::make_shared<Query>(Select {}, std::move(where));
    auto filter = Value::Filter::chain(std::move(f), query->where.getFilter());
    auto st = store.find(id);
    if (st == store.end() && store.size() < max_store_keys)
        st = store.emplace(id, scheduler.time() + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME).first;

    size_t tokenlocal = 0;
    if (st != store.end()) {
        tokenlocal = st->second.listen(gcb, filter, query);
        if (tokenlocal == 0)
            return 0;
    }

    auto token4 = Dht::listenTo(id, AF_INET, gcb, filter, query);
    auto token6 = token4 == 0 ? 0 : Dht::listenTo(id, AF_INET6, gcb, filter, query);
    if (token6 == 0 && st != store.end()) {
        st->second.cancelListen(tokenlocal);
        return 0;
    }

    listeners.emplace(token, std::make_tuple(tokenlocal, token4, token6));
    return token;
}

bool
Dht::cancelListen(const InfoHash& id, size_t token)
{
    scheduler.syncTime();

    auto it = listeners.find(token);
    if (it == listeners.end()) {
        if (logger_)
            logger_->warn("Listen token not found: {}", token);
        return false;
    }
    if (logger_)
        logger_->debug("cancelListen {} with token {}", id.to_view(), token);
    if (auto tokenlocal = std::get<0>(it->second)) {
        auto st = store.find(id);
        if (st != store.end())
            st->second.cancelListen(tokenlocal);
    }
    auto searches_cancel_listen = [this, &id](std::map<InfoHash, Sp<Search>>& srs, size_t token) {
        if (token) {
            auto srp = srs.find(id);
            if (srp != srs.end())
                srp->second->cancelListen(token, scheduler);
        }
    };
    searches_cancel_listen(dht4.searches, std::get<1>(it->second));
    searches_cancel_listen(dht6.searches, std::get<2>(it->second));
    listeners.erase(it);
    return true;
}

struct OpStatus
{
    struct Status
    {
        bool done {false};
        bool ok {false};
        Status(bool done = false, bool ok = false)
            : done(done)
            , ok(ok)
        {}
    };
    Status status;
    Status status4;
    Status status6;
};

template<typename T>
struct GetStatus : public OpStatus
{
    T values;
    std::vector<Sp<Node>> nodes;
};

void
Dht::put(const InfoHash& id, Sp<Value> val, DoneCallback callback, time_point created, bool permanent)
{
    if (not id or not val) {
        if (logger_)
            logger_->warn("Put called with invalid key or value");
        if (callback)
            callback(false, {});
        return;
    }
    if (val->id == Value::INVALID_ID)
        val->id = std::uniform_int_distribution<Value::Id> {1}(rd);
    scheduler.syncTime();
    const auto& now = scheduler.time();
    created = std::min(now, created);
    storageStore(id, val, created, {}, permanent);

    if (logger_)
        logger_->debug("put: adding {} → {}", id.to_view(), val->toString());

    auto op = std::make_shared<OpStatus>();
    auto donecb = [callback](const std::vector<Sp<Node>>& nodes, OpStatus& op) {
        // Callback as soon as the value is announced on one of the available networks
        if (callback and not op.status.done and (op.status4.done && op.status6.done)) {
            callback(op.status4.ok or op.status6.ok, nodes);
            op.status.done = true;
        }
    };
    announce(
        id,
        AF_INET,
        val,
        [=](bool ok4, const std::vector<Sp<Node>>& nodes) {
            if (logger_)
                logger_->debug("Announce done IPv4 {}", ok4);
            auto& o = *op;
            o.status4 = {true, ok4};
            donecb(nodes, o);
        },
        created,
        permanent);
    announce(
        id,
        AF_INET6,
        val,
        [=](bool ok6, const std::vector<Sp<Node>>& nodes) {
            if (logger_)
                logger_->debug("Announce done IPv6 {}", ok6);
            auto& o = *op;
            o.status6 = {true, ok6};
            donecb(nodes, o);
        },
        created,
        permanent);
}

template<typename T>
void
doneCallbackWrapper(DoneCallback dcb, const std::vector<Sp<Node>>& nodes, GetStatus<T>& op)
{
    if (op.status.done)
        return;
    op.nodes.insert(op.nodes.end(), nodes.begin(), nodes.end());
    if (op.status.ok or (op.status4.done and op.status6.done)) {
        bool ok = op.status.ok or op.status4.ok or op.status6.ok;
        op.status.done = true;
        if (dcb)
            dcb(ok, op.nodes);
    }
}

template<typename T, typename St, typename Cb, typename Av, typename Cv>
bool
callbackWrapper(
    Cb get_cb, DoneCallback done_cb, const std::vector<Sp<T>>& values, Av add_values, Cv cache_values, GetStatus<St>& op)
{
    if (op.status.done)
        return false;
    auto newvals = add_values(values);
    if (not newvals.empty()) {
        op.status.ok = !get_cb(newvals);
        cache_values(newvals);
    }
    doneCallbackWrapper(done_cb, {}, op);
    return !op.status.ok;
}

void
Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filter&& filter, Where&& where)
{
    if (not id) {
        if (logger_)
            logger_->warn("Get called with invalid key");
        if (donecb)
            donecb(false, {});
        return;
    }
    scheduler.syncTime();

    auto op = std::make_shared<GetStatus<std::map<Value::Id, Sp<Value>>>>();
    auto gcb = [getcb, donecb, op](const std::vector<Sp<Value>>& vals) {
        auto& o = *op;
        return callbackWrapper(
            getcb,
            donecb,
            vals,
            [&o](const std::vector<Sp<Value>>& values) {
                std::vector<Sp<Value>> newvals {};
                for (const auto& v : values) {
                    auto it = o.values.find(v->id);
                    if (it == o.values.cend() or (it->second != v && !(*it->second == *v))) {
                        newvals.push_back(v);
                    }
                }
                return newvals;
            },
            [&o](const std::vector<Sp<Value>>& newvals) {
                for (const auto& v : newvals)
                    o.values[v->id] = v;
            },
            o);
    };

    auto q = std::make_shared<Query>(Select {}, std::move(where));
    auto f = filter.chain(q->where.getFilter());

    /* Try to answer this search locally. */
    gcb(getLocal(id, f));

    Dht::search(
        id,
        AF_INET,
        gcb,
        {},
        [=](bool ok, const std::vector<Sp<Node>>& nodes) {
            // logger__WARN("DHT done IPv4");
            op->status4 = {true, ok};
            doneCallbackWrapper(donecb, nodes, *op);
        },
        f,
        q);
    Dht::search(
        id,
        AF_INET6,
        gcb,
        {},
        [=](bool ok, const std::vector<Sp<Node>>& nodes) {
            // logger__WARN("DHT done IPv6");
            op->status6 = {true, ok};
            doneCallbackWrapper(donecb, nodes, *op);
        },
        f,
        q);
}

void
Dht::query(const InfoHash& id, QueryCallback cb, DoneCallback done_cb, Query&& q)
{
    if (not id) {
        if (logger_)
            logger_->warn("Query called with invalid key");
        if (done_cb)
            done_cb(false, {});
        return;
    }
    scheduler.syncTime();
    auto op = std::make_shared<GetStatus<std::vector<Sp<FieldValueIndex>>>>();
    auto f = q.where.getFilter();
    auto qcb = [cb, done_cb, op](const std::vector<Sp<FieldValueIndex>>& fields) {
        auto& o = *op;
        return callbackWrapper(
            cb,
            done_cb,
            fields,
            [&](const std::vector<Sp<FieldValueIndex>>& fields) {
                std::vector<Sp<FieldValueIndex>> newvals {};
                for (const auto& f : fields) {
                    auto it = std::find_if(o.values.cbegin(), o.values.cend(), [&](const Sp<FieldValueIndex>& sf) {
                        return sf == f or f->containedIn(*sf);
                    });
                    if (it == o.values.cend()) {
                        auto lesser = std::find_if(o.values.begin(),
                                                   o.values.end(),
                                                   [&](const Sp<FieldValueIndex>& sf) { return sf->containedIn(*f); });
                        if (lesser != o.values.end())
                            o.values.erase(lesser);
                        newvals.push_back(f);
                    }
                }
                return newvals;
            },
            [&](const std::vector<Sp<FieldValueIndex>>& fields) {
                o.values.insert(o.values.end(), fields.begin(), fields.end());
            },
            o);
    };

    /* Try to answer this search locally. */
    auto values = getLocal(id, f);
    std::vector<Sp<FieldValueIndex>> local_fields(values.size());
    std::transform(values.begin(), values.end(), local_fields.begin(), [&q](const Sp<Value>& v) {
        return std::make_shared<FieldValueIndex>(*v, q.select);
    });
    qcb(local_fields);

    auto sq = std::make_shared<Query>(std::move(q));
    Dht::search(
        id,
        AF_INET,
        {},
        qcb,
        [=](bool ok, const std::vector<Sp<Node>>& nodes) {
            // logger__WARN("DHT done IPv4");
            op->status4 = {true, ok};
            doneCallbackWrapper(done_cb, nodes, *op);
        },
        f,
        sq);
    Dht::search(
        id,
        AF_INET6,
        {},
        qcb,
        [=](bool ok, const std::vector<Sp<Node>>& nodes) {
            // logger__WARN("DHT done IPv6");
            op->status6 = {true, ok};
            doneCallbackWrapper(done_cb, nodes, *op);
        },
        f,
        sq);
}

std::vector<Sp<Value>>
Dht::getLocal(const InfoHash& id, const Value::Filter& f) const
{
    auto s = store.find(id);
    if (s == store.end())
        return {};
    return s->second.get(f);
}

Sp<Value>
Dht::getLocalById(const InfoHash& id, Value::Id vid) const
{
    auto s = store.find(id);
    if (s != store.end())
        return s->second.getById(vid);
    return {};
}

std::vector<Sp<Value>>
Dht::getPut(const InfoHash& id) const
{
    std::vector<Sp<Value>> ret;
    auto find_values = [&](const std::map<InfoHash, Sp<Search>>& srs) {
        auto srp = srs.find(id);
        if (srp == srs.end())
            return;
        auto vals = srp->second->getPut();
        ret.insert(ret.end(), vals.begin(), vals.end());
    };
    find_values(dht4.searches);
    find_values(dht6.searches);
    return ret;
}

Sp<Value>
Dht::getPut(const InfoHash& id, const Value::Id& vid) const
{
    auto find_value = [&](const std::map<InfoHash, Sp<Search>>& srs) {
        auto srp = srs.find(id);
        return (srp != srs.end()) ? srp->second->getPut(vid) : Sp<Value> {};
    };
    if (auto v4 = find_value(dht4.searches))
        return v4;
    if (auto v6 = find_value(dht6.searches))
        return v6;
    return {};
}

bool
Dht::cancelPut(const InfoHash& id, const Value::Id& vid)
{
    bool canceled {false};
    auto sr_cancel_put = [&](std::map<InfoHash, Sp<Search>>& srs) {
        auto srp = srs.find(id);
        return (srp != srs.end()) ? srp->second->cancelPut(vid) : false;
    };
    canceled |= sr_cancel_put(dht4.searches);
    canceled |= sr_cancel_put(dht6.searches);
    if (canceled) {
        auto st = store.find(id);
        if (st != store.end()) {
            if (auto value = st->second.remove(id, vid))
                storageRemoved(id, st->second, {value}, value->size());
        }
    }
    return canceled;
}

// Storage

void
Dht::storageChanged(const InfoHash& id, Storage& st, const Sp<Value>& v, bool newValue)
{
    if (newValue) {
        if (not st.local_listeners.empty()) {
            if (logger_)
                logger_->debug("[store {}] {} local listeners", id.to_view(), st.local_listeners.size());
            std::vector<std::pair<ValueCallback, std::vector<Sp<Value>>>> cbs;
            cbs.reserve(st.local_listeners.size());
            for (const auto& l : st.local_listeners) {
                std::vector<Sp<Value>> vals;
                if (not l.second.filter or l.second.filter(*v))
                    vals.push_back(v);
                if (not vals.empty()) {
                    if (logger_)
                        logger_->debug("[store {}] Sending update local listener with token {}", id.to_view(), l.first);
                    cbs.emplace_back(l.second.get_cb, std::move(vals));
                }
            }
            // listeners are copied: they may be deleted by the callback
            for (auto& cb : cbs)
                cb.first(cb.second, false);
        }
    }

    if (not st.listeners.empty()) {
        if (logger_)
            logger_->debug("[store {}] {} remote listeners", id.to_view(), st.listeners.size());
        for (const auto& node_listeners : st.listeners) {
            for (const auto& l : node_listeners.second) {
                auto f = l.second.query.where.getFilter();
                if (f and not f(*v))
                    continue;
                if (logger_)
                    logger_->debug("[store {}] [node {}] Sending update",
                                   id.to_view(),
                                   node_listeners.first->toString());
                std::vector<Sp<Value>> vals;
                vals.push_back(v);
                network_engine.tellListener(node_listeners.first,
                                            l.first,
                                            id,
                                            0,
                                            {},
                                            {},
                                            std::move(vals),
                                            l.second.query,
                                            l.second.version);
            }
        }
    }
}

bool
Dht::storageStore(const InfoHash& id,
                  const Sp<Value>& value,
                  time_point created,
                  const SockAddr& sa,
                  bool permanent,
                  time_point expiration)
{
    const auto& now = scheduler.time();
    created = std::min(created, now);
    if (expiration == time_point::min())
        expiration = permanent ? time_point::max() : created + getType(value->type).expiration;
    else if (not permanent)
        expiration = std::min(expiration, created + getType(value->type).expiration);
    if (expiration < now)
        return false;

    auto st = store.find(id);
    if (st == store.end()) {
        if (store.size() >= max_store_keys)
            return false;
        auto st_i = store.emplace(id, now);
        st = st_i.first;
        if (maintain_storage and st_i.second)
            scheduler.add(st->second.maintenance_time, std::bind(&Dht::dataPersistence, this, id));
    }

    StorageBucket* store_bucket {nullptr};
    bool is_local = !sa;
    if (sa)
        store_bucket = &store_quota.try_emplace(sa, sa).first->second;
    else
        store_bucket = local_store_quota.get();

    // Reject new local values if local storage limit is exceeded
    if (is_local && max_local_store_size != STORAGE_LIMIT_UNLIMITED
        && local_store_quota->size() + value->size() > max_local_store_size) {
        if (logger_)
            logger_->warn("[store {}] Rejecting local put: local storage limit exceeded ({}/{})",
                          id.to_view(),
                          local_store_quota->size(),
                          max_local_store_size);
        return false;
    }

    auto store = st->second.store(id, value, created, expiration, store_bucket);
    if (auto vs = store.first) {
        total_store_size += store.second.size_diff;
        total_values += store.second.values_diff;
        scheduler.cancel(vs->expiration_job);
        if (not permanent) {
            vs->expiration_job = scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
        }
        if (total_store_size - local_store_quota->size() > max_store_size) {
            auto value = vs->data;
            auto value_diff = store.second.values_diff;
            auto value_edit = store.second.edited_values;
            expireStore();
            storageChanged(id, st->second, value, value_diff > 0 || value_edit > 0);
        } else {
            storageChanged(id, st->second, vs->data, store.second.values_diff > 0 || store.second.edited_values > 0);
        }
    }

    return std::get<0>(store);
}

void
Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query, int version)
{
    const auto& now = scheduler.time();
    auto st = store.find(id);
    if (st == store.end()) {
        if (store.size() >= max_store_keys)
            return;
        st = store.emplace(id, now).first;
    }
    auto& node_listeners = st->second.listeners[node];
    auto l = node_listeners.find(socket_id);
    if (l == node_listeners.end()) {
        auto vals = st->second.get(query.where.getFilter());
        if (not vals.empty()) {
            network_engine.tellListener(node,
                                        socket_id,
                                        id,
                                        WANT4 | WANT6,
                                        dht4.buckets.findClosestNodes(id, now, TARGET_NODES),
                                        dht6.buckets.findClosestNodes(id, now, TARGET_NODES),
                                        std::move(vals),
                                        query,
                                        version);
        }
        node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query), version});
    } else
        l->second.refresh(now, std::forward<Query>(query));
}

void
Dht::expireStore(decltype(store)::iterator i)
{
    const auto& id = i->first;
    auto& st = i->second;
    auto stats = st.expire(id, scheduler.time());
    if (not stats.expired_values.empty()) {
        storageRemoved(id, st, stats.expired_values, -stats.size_diff);
    }
}

void
Dht::expireStorage(InfoHash h)
{
    auto i = store.find(h);
    if (i != store.end())
        expireStore(i);
}

void
Dht::storageRemoved(const InfoHash& id, Storage& st, const std::vector<Sp<Value>>& values, size_t totalSize)
{
    if (logger_)
        logger_->debug("[store {}] Discarded {} values ({} bytes)", id.to_view(), values.size(), totalSize);

    total_store_size -= totalSize;
    total_values -= values.size();

    if (not st.listeners.empty()) {
        if (logger_)
            logger_->debug("[store {}] {} remote listeners", id.to_view(), st.listeners.size());

        std::vector<Value::Id> ids;
        ids.reserve(values.size());
        for (const auto& v : values)
            ids.emplace_back(v->id);

        for (const auto& node_listeners : st.listeners) {
            for (const auto& l : node_listeners.second) {
                if (logger_)
                    logger_->warn("[store {}] [node {}] Sending expired",
                                  id.to_view(),
                                  node_listeners.first->toString());
                network_engine.tellListenerExpired(node_listeners.first, l.first, id, ids, l.second.version);
            }
        }
    }
    for (const auto& local_listeners : st.local_listeners) {
        local_listeners.second.get_cb(values, true);
    }
}

void
Dht::expireStore()
{
    // removing expired values
    for (auto i = store.begin(); i != store.end();) {
        expireStore(i);

        if (i->second.empty() && i->second.listeners.empty() && i->second.local_listeners.empty()) {
            if (logger_)
                logger_->debug("[store {}] Discarding empty storage", i->first.to_view());
            i = store.erase(i);
        } else
            ++i;
    }

    // remove more values if remote storage limit is exceeded
    while (total_store_size - local_store_quota->size() > max_store_size) {
        // find IP using the most storage
        if (store_quota.empty()) {
            if (logger_)
                logger_->warn("No space left: local data consumes all the quota!");
            break;
        }
        auto largest = store_quota.begin();
        for (auto it = std::next(largest); it != store_quota.end(); ++it) {
            if (it->second.size() > largest->second.size())
                largest = it;
        }
        if (largest != store_quota.end() && !largest->second.empty()) {
            while (true) {
                auto exp_value = largest->second.getOldest();
                auto storage = store.find(exp_value.first);
                if (storage != store.end()) {
                    if (logger_)
                        logger_->warn("Storage quota full: discarding value from {} at {} {:016x}",
                                      largest->first.toString(),
                                      exp_value.first.to_view(),
                                      exp_value.second);

                    if (auto value = storage->second.remove(exp_value.first, exp_value.second)) {
                        storageRemoved(storage->first, storage->second, {value}, value->size());
                        break;
                    }
                }
            }
        } else {
            if (logger_)
                logger_->warn("No space left: remote data consumes all the quota!");
            break;
        }
    }

    // remove unused quota entires
    for (auto i = store_quota.begin(); i != store_quota.end();) {
        if (i->second.empty())
            i = store_quota.erase(i);
        else
            ++i;
    }
}

void
Dht::connectivityChanged(sa_family_t af)
{
    const auto& now = scheduler.time();
    scheduler.edit(nextNodesConfirmation, now);
    auto& dht = this->dht(af);
    dht.buckets.connectivityChanged(now);
    dht.reported_addr.clear();
    network_engine.connectivityChanged(af);
    startBootstrap(); // will only happen if disconnected
}

NodeStats
Dht::getNodesStats(sa_family_t af) const
{
    NodeStats stats = dht(af).getNodesStats(scheduler.time(), myid);
    stats.node_cache_size = network_engine.getNodeCacheSize(af);
    return stats;
}

NodeStats
Dht::Kad::getNodesStats(time_point now, const InfoHash& myid) const
{
    NodeStats stats {};
    for (const auto& b : buckets) {
        for (auto& n : b.nodes) {
            if (n->isGood(now)) {
                stats.good_nodes++;
                if (n->isIncoming())
                    stats.incoming_nodes++;
            } else if (not n->isExpired())
                stats.dubious_nodes++;
        }
        if (b.cached)
            stats.cached_nodes++;
    }
    stats.table_depth = buckets.depth(buckets.findBucket(myid));
    stats.searches = searches.size();
    return stats;
}

void
Dht::dumpBucket(const Bucket& b, std::ostream& out) const
{
    const auto& now = scheduler.time();
    using namespace std::chrono;
    out << b.first << " count: " << b.nodes.size() << " updated: " << print_time_relative(now, b.time);
    if (b.cached)
        out << " (cached)";
    out << std::endl;
    for (auto& n : b.nodes) {
        out << "    Node " << n->toString();
        const auto& t = n->getTime();
        const auto& r = n->getReplyTime();
        if (t != r)
            out << " updated: " << print_time_relative(now, t) << ", replied: " << print_time_relative(now, r);
        else
            out << " updated: " << print_time_relative(now, t);
        if (n->isExpired())
            out << " [expired]";
        else if (n->isGood(now))
            out << " [good]";
        out << std::endl;
    }
}

void
Dht::dumpSearch(const Search& sr, std::ostream& out) const
{
    const auto& now = scheduler.time();
    const auto& listen_expire = getListenExpiration();
    using namespace std::chrono;
    fmt::print(out,
               "Search IPv{} {} gets: {} last step: {}{}{}{}{}\n",
               (sr.af == AF_INET6 ? '6' : '4'),
               sr.id,
               sr.callbacks.size(),
               print_time_relative(now, sr.step_time),
               (sr.done ? " [done]"sv : ""sv),
               (sr.expired ? " [expired]"sv : ""sv),
               (sr.isSynced(now) ? " [synced]"sv : " [not synced]"sv),
               (sr.isListening(now, listen_expire) ? " [listening]"sv : ""sv));

    // printing the queries
    if (sr.callbacks.size() + sr.listeners.size() > 0)
        fmt::print(out, "Queries:\n");
    for (const auto& cb : sr.callbacks) {
        out << *cb.second.query << std::endl;
    }
    for (const auto& l : sr.listeners) {
        out << *l.second.query << std::endl;
    }

    for (const auto& a : sr.announce) {
        bool announced = sr.isAnnounced(a.value->id);
        out << "Announcement: " << *a.value << (announced ? " [announced]" : "") << std::endl;
    }

    fmt::print(out, " Common bits    InfoHash                       Conn. Get   Ops  IP\n");
    auto last_get = sr.getLastGetTime();
    for (const auto& np : sr.nodes) {
        auto& n = *np;
        auto listen = (not sr.listeners.empty()
                           ? (n.listenStatus.empty() ? "    "
                                                     : fmt::format("[{}] ",
                                                                   (n.isListening(now, listen_expire)
                                                                        ? 'l'
                                                                        : (n.pending(n.listenStatus) ? 'f' : ' '))))
                           : "");
        std::string announce;
        if (not sr.announce.empty()) {
            if (n.acked.empty()) {
                announce = std::string(sr.announce.size() + 3, ' ');
            } else {
                announce = "[";
                for (const auto& a : sr.announce) {
                    auto ack = n.acked.find(a.value->id);
                    if (ack == n.acked.end() or not ack->second.req) {
                        announce += ' ';
                    } else {
                        announce += ack->second.req->getStateChar();
                    }
                }
                announce += "] ";
            }
        }

        fmt::print(out,
                   "{:3} {} {} [{}{}] {}{}{}\n",
                   InfoHash::commonBits(sr.id, n.node->id),
                   n.node->id,
                   (findNode(n.node->id, sr.af) ? '*' : ' '),
                   (n.pending(n.getStatus) ? (n.candidate ? 'c' : 'f') : ' '),
                   (n.isSynced(now) ? (n.last_get_reply > last_get ? 'u' : 's') : '-'),
                   listen,
                   announce,
                   n.node->getAddrStr());
    }
}

void
Dht::dumpTables() const
{
    std::ostringstream out;
    out << "My id " << myid << std::endl;

    out << "Buckets IPv4 :" << std::endl;
    for (const auto& b : dht4.buckets)
        dumpBucket(b, out);
    out << "Buckets IPv6 :" << std::endl;
    for (const auto& b : dht6.buckets)
        dumpBucket(b, out);

    auto dump_searches = [&](std::map<InfoHash, Sp<Search>> srs) {
        for (auto& srp : srs)
            dumpSearch(*srp.second, out);
    };
    dump_searches(dht4.searches);
    dump_searches(dht6.searches);
    out << std::endl;

    out << getStorageLog() << std::endl;

    if (logger_)
        logger_->debug("{}", out.str());
}

std::string
Dht::getStorageLog() const
{
    std::ostringstream out;
    for (const auto& s : store) {
        const auto& st = s.second;
        out << "Storage " << s.first << " " << st.listeners.size() << " list., " << st.valueCount() << " values, "
            << printByteCount(st.totalSize()) << "\n";
    }
    out << std::endl << std::endl;
    std::multimap<size_t, const SockAddr*> q_map;
    for (const auto& ip : store_quota)
        if (ip.second.size())
            q_map.emplace(ip.second.size(), &ip.first);
    for (auto ip = q_map.rbegin(); ip != q_map.rend(); ++ip)
        out << "IP " << ip->second->toString() << " uses " << printByteCount(ip->first) << std::endl;
    out << std::endl;
    out << "Total " << store.size() << " storages, " << total_values << " values (";
    if (total_store_size < 1024)
        out << total_store_size << " bytes)";
    else {
        auto remote_store_size = total_store_size - local_store_quota->size();
        out << (remote_store_size / 1024) << " / " << (max_store_size / 1024) << " KB remote, "
            << (local_store_quota->size() / 1024) << " / ";
        if (max_local_store_size == STORAGE_LIMIT_UNLIMITED)
            out << "unlimited";
        else
            out << (max_local_store_size / 1024) << " KB";
        out << " local)";
    }
    out << std::endl;
    return out.str();
}

std::string
Dht::getStorageLog(const InfoHash& h) const
{
    auto s = store.find(h);
    if (s == store.end())
        return fmt::format("Storage {} empty\n", h);
    return printStorageLog(*s);
}

std::string
Dht::printStorageLog(const decltype(store)::value_type& s) const
{
    std::ostringstream out;
    using namespace std::chrono;
    const auto& st = s.second;
    const auto& now = scheduler.time();
    out << "Storage " << s.first << " " << st.listeners.size() << " list., " << st.valueCount() << " values, "
        << printByteCount(st.totalSize()) << std::endl;
    for (const auto& v : st.getValues()) {
        out << "   Value " << v.data->id << " size: " << printByteCount(v.data->size())
            << " created: " << print_time_relative(now, v.created)
            << " expires: " << print_time_relative(now, v.expiration) << std::endl;
    }
    if (not st.local_listeners.empty())
        out << "   " << st.local_listeners.size() << " local listeners" << std::endl;
    for (const auto& node_listeners : st.listeners) {
        const auto& node = node_listeners.first;
        out << "   " << "Listener " << node->toString() << " : " << node_listeners.second.size() << " entries"
            << std::endl;
    }
    return out.str();
}

std::string
Dht::getRoutingTablesLog(sa_family_t af) const
{
    std::ostringstream out;
    for (const auto& b : buckets(af))
        dumpBucket(b, out);
    return out.str();
}

std::string
Dht::getSearchesLog(sa_family_t af) const
{
    std::ostringstream out;
    auto num_searches = dht4.searches.size() + dht6.searches.size();
    if (num_searches > 8) {
        if (not af or af == AF_INET)
            for (const auto& sr : dht4.searches)
                out << "[search " << sr.first << " IPv4]" << std::endl;
        if (not af or af == AF_INET6)
            for (const auto& sr : dht6.searches)
                out << "[search " << sr.first << " IPv6]" << std::endl;
    } else {
        out << "s:synched, u:updated, a:announced, c:candidate, f:cur req, x:expired, *:known" << std::endl;
        if (not af or af == AF_INET)
            for (const auto& sr : dht4.searches)
                dumpSearch(*sr.second, out);
        if (not af or af == AF_INET6)
            for (const auto& sr : dht6.searches)
                dumpSearch(*sr.second, out);
    }
    out << "Total: " << num_searches << " searches (" << dht4.searches.size() << " IPv4, " << dht6.searches.size()
        << " IPv6)." << std::endl;
    return out.str();
}

std::string
Dht::getSearchLog(const InfoHash& id, sa_family_t af) const
{
    std::ostringstream out;
    if (af == AF_UNSPEC) {
        out << getSearchLog(id, AF_INET) << getSearchLog(id, AF_INET6);
    } else {
        auto& srs = searches(af);
        auto sr = srs.find(id);
        if (sr != srs.end())
            dumpSearch(*sr->second, out);
    }
    return out.str();
}

Dht::~Dht()
{
    for (auto& s : dht4.searches)
        s.second->clear();
    for (auto& s : dht6.searches)
        s.second->clear();
}

net::NetworkConfig
fromDhtConfig(const Config& config)
{
    net::NetworkConfig netConf;
    netConf.network = config.network;
    // global limiting should be triggered by at least 8 different IPs
    auto peerLimit = [](ssize_t global, ssize_t peer) -> ssize_t {
        return peer ? peer : (global < 0 ? -1 : global / 8);
    };
    netConf.max_req_per_sec = config.max_req_per_sec ? config.max_req_per_sec : MAX_REQUESTS_PER_SEC;
    netConf.max_peer_req_per_sec = peerLimit(netConf.max_req_per_sec, config.max_peer_req_per_sec);
    // left at 0, the engine uses the limits above
    netConf.max_unauth_req_per_sec = config.max_unauth_req_per_sec;
    netConf.max_peer_unauth_req_per_sec = config.max_unauth_req_per_sec
                                              ? peerLimit(config.max_unauth_req_per_sec,
                                                          config.max_peer_unauth_req_per_sec)
                                              : config.max_peer_unauth_req_per_sec;
    netConf.max_crypto_per_sec = config.max_crypto_per_sec ? config.max_crypto_per_sec : MAX_CRYPTO_PER_SEC;
    netConf.max_peer_crypto_per_sec = peerLimit(netConf.max_crypto_per_sec, config.max_peer_crypto_per_sec);
    netConf.is_client = config.client_mode;
    netConf.transport = config.transport;
    auto& transport = netConf.transport;
    if (transport.isPrivate() and not transport.enabled)
        throw DhtException("Private networks require transport encryption");
    if (transport.enabled and not transport.key)
        transport.key = transport::KeyPair::generate();
    if (transport.isPrivate()) {
        if (transport.credential.key != transport.key.pub)
            throw DhtException("Transport credential doesn't match the transport key");
        if (not transport.credential.verify(transport.authorities))
            throw DhtException("Invalid or expired transport credential");
    }
    return netConf;
}

Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock,
         const Config& config,
         const Sp<Logger>& l,
         std::unique_ptr<std::mt19937_64>&& rda)
    : DhtInterface(l)
    , rd(rda ? std::move(*rda) : crypto::getSeededRandomEngine<std::mt19937_64>())
    , myid(config.node_id ? config.node_id : InfoHash::getRandom(rd))
    , store()
    , store_quota()
    , local_store_quota(std::make_unique<StorageBucket>())
    , max_store_keys(config.max_store_keys ? (int) config.max_store_keys : MAX_HASHES)
    , max_store_size(config.max_store_size ? (size_t) config.max_store_size : STORAGE_LIMIT_DEFAULT)
    , max_local_store_size(config.max_local_store_size ? (size_t) config.max_local_store_size : STORAGE_LIMIT_UNLIMITED)
    , max_searches(config.max_searches ? (int) config.max_searches : MAX_SEARCHES)
    , network_engine(myid,
                     fromDhtConfig(config),
                     std::move(sock),
                     logger_,
                     rd,
                     scheduler,
                     std::bind(&Dht::onError, this, _1, _2),
                     std::bind(&Dht::onNewNode, this, _1, _2),
                     std::bind(&Dht::onReportedAddr, this, _1, _2),
                     std::bind(&Dht::onPing, this, _1),
                     std::bind(&Dht::onFindNode, this, _1, _2, _3),
                     std::bind(&Dht::onGetValues, this, _1, _2, _3, _4),
                     std::bind(&Dht::onListen, this, _1, _2, _3, _4, _5),
                     std::bind(&Dht::onAnnounce, this, _1, _2, _3, _4),
                     std::bind(&Dht::onRefresh, this, _1, _2, _3))
    , persistPath(config.persist_path)
    , is_bootstrap(config.is_bootstrap)
    , maintain_storage(config.maintain_storage)
    , public_stable(config.public_stable)
{
    if (auto key = network_engine.getTransportKey())
        myid = transport::nodeId(key->pub);
    persistTransportKey = config.transport.enabled and not config.transport.key and not persistPath.empty();
    scheduler.syncTime();
    auto s = network_engine.getSocket();
    if (not s or (not s->hasIPv4() and not s->hasIPv6()))
        throw DhtException("Opened socket required");
    if (s->hasIPv4()) {
        dht4.buckets = {Bucket {AF_INET}};
        dht4.buckets.is_client = config.is_bootstrap;
    }
    if (s->hasIPv6()) {
        dht6.buckets = {Bucket {AF_INET6}};
        dht6.buckets.is_client = config.is_bootstrap;
    }

    search_id = std::uniform_int_distribution<decltype(search_id)> {}(rd);

    uniform_duration_distribution<> time_dis {std::chrono::seconds(3), std::chrono::seconds(5)};
    nextNodesConfirmation = scheduler.add(scheduler.time() + time_dis(rd), std::bind(&Dht::confirmNodes, this));

    if (not persistPath.empty())
        loadState(persistPath);

    expire();

    if (logger_)
        logger_->debug("DHT node initialised with ID {:s}", myid);
}

bool
Dht::neighbourhoodMaintenance(RoutingTable& list)
{
    // logger__DBG("neighbourhoodMaintenance");
    auto b = list.findBucket(myid);
    if (b == list.end())
        return false;

    InfoHash id = myid;
#ifdef _WIN32
    std::uniform_int_distribution<int> rand_byte {0, std::numeric_limits<uint8_t>::max()};
#else
    std::uniform_int_distribution<uint8_t> rand_byte;
#endif
    id[HASH_LEN - 1] = rand_byte(rd);

    std::bernoulli_distribution rand_trial(1. / 8.);
    auto q = b;
    if (std::next(q) != list.end() && (q->nodes.empty() || rand_trial(rd)))
        q = std::next(q);
    if (b != list.begin() && (q->nodes.empty() || rand_trial(rd))) {
        auto r = std::prev(b);
        if (!r->nodes.empty())
            q = r;
    }

    auto n = q->randomNode(rd);
    if (n) {
        if (logger_)
            logger_->debug("[node {}] Sending [find {}] for neighborhood maintenance", n->toString(), id.to_view());
        /* Since our node-id is the same in both DHTs, it's probably
           profitable to query both families. */
        network_engine.sendFindNode(n, id, network_engine.want());
    }

    return true;
}

bool
Dht::bucketMaintenance(RoutingTable& list)
{
    std::bernoulli_distribution rand_trial(1. / 8.);
    std::bernoulli_distribution rand_trial_38(1. / 38.);

    bool sent {false};
    for (auto b = list.begin(); b != list.end(); ++b) {
        if (b->time < scheduler.time() - std::chrono::minutes(10) || b->nodes.empty()) {
            /* This bucket hasn't seen any positive confirmation for a long
               time. Pick a random id in this bucket's range, and send a request
               to a random node. */
            InfoHash id = list.randomId(b, rd);
            auto q = b;
            /* If the bucket is empty, we try to fill it from a neighbour.
               We also sometimes do it gratuitiously to recover from
               buckets full of broken nodes. */
            if (std::next(b) != list.end() && (q->nodes.empty() || rand_trial(rd)))
                q = std::next(b);
            if (b != list.begin() && (q->nodes.empty() || rand_trial(rd))) {
                auto r = std::prev(b);
                if (!r->nodes.empty())
                    q = r;
            }

            auto n = q->randomNode(rd);
            if (n and not n->isPendingMessage()) {
                want_t want = -1;

                if (network_engine.want() != want) {
                    auto otherbucket = findBucket(id, q->af == AF_INET ? AF_INET6 : AF_INET);
                    if (otherbucket && otherbucket->nodes.size() < TARGET_NODES)
                        /* The corresponding bucket in the other family
                           is emptyish -- querying both is useful. */
                        want = WANT4 | WANT6;
                    else if (rand_trial_38(rd))
                        /* Most of the time, this just adds overhead.
                           However, it might help stitch back one of
                           the DHTs after a network collapse, so query
                           both, but only very occasionally. */
                        want = WANT4 | WANT6;
                }

                if (logger_)
                    logger_->debug("[node {}] Sending find {} for bucket maintenance", n->toString(), id.to_view());
                // auto start = scheduler.time();
                network_engine.sendFindNode(n, id, want, nullptr, [this, n](const net::Request&, bool over) {
                    if (over) {
                        const auto& end = scheduler.time();
                        // using namespace std::chrono;
                        // if (logger_)
                        //     logger_->d(n->id, "[node %s] Bucket maintenance op expired after %s",
                        //     n->toString().c_str(), print_duration(end-start).c_str());
                        scheduler.edit(nextNodesConfirmation, end + Node::MAX_RESPONSE_TIME);
                    }
                });
                sent = true;
            }
        }
    }
    return sent;
}

void
Dht::dataPersistence(InfoHash id)
{
    const auto& now = scheduler.time();
    auto str = store.find(id);
    if (str != store.end() and now > str->second.maintenance_time) {
        if (logger_)
            logger_->debug("[storage {}] Maintenance ({} values, {} bytes)",
                           id.to_view(),
                           str->second.valueCount(),
                           str->second.totalSize());
        maintainStorage(*str);
        str->second.maintenance_time = now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME;
        scheduler.add(str->second.maintenance_time, std::bind(&Dht::dataPersistence, this, id));
    }
}

size_t
Dht::maintainStorage(decltype(store)::value_type& storage, bool force, const DoneCallback& donecb)
{
    const auto& now = scheduler.time();
    size_t announce_per_af = 0;

    auto maintain = [&](sa_family_t af) {
        bool want = true;
        auto nodes = buckets(af).findClosestNodes(storage.first, now);
        if (!nodes.empty()) {
            if (force || storage.first.xorCmp(nodes.back()->id, myid) < 0) {
                for (auto& value : storage.second.getValues()) {
                    const auto& vt = getType(value.data->type);
                    if (force || value.created + vt.expiration > now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME) {
                        // gotta put that value there
                        announce(storage.first, af, value.data, donecb, value.created);
                        ++announce_per_af;
                    }
                }
                want = false;
            }
        }
        return want;
    };
    bool want4 = maintain(AF_INET), want6 = maintain(AF_INET6);

    if (not want4 and not want6) {
        if (logger_)
            logger_->debug("Discarding storage values {}", storage.first.to_view());
        auto diff = storage.second.clear(storage.first);
        total_store_size += diff.size_diff;
        total_values += diff.values_diff;
    }

    return announce_per_af;
}

time_point
Dht::periodic(const uint8_t* buf, size_t buflen, SockAddr from, const time_point& now)
{
    scheduler.syncTime(now);
    if (buflen) {
        try {
            network_engine.processMessage(buf, buflen, std::move(from));
        } catch (const std::exception& e) {
            if (logger_)
                logger_->warn("Unable to process message: {}", e.what());
        }
    }
    return scheduler.run();
}

void
Dht::expire()
{
    uniform_duration_distribution<> time_dis(std::chrono::minutes(2), std::chrono::minutes(6));
    auto expire_stuff_time = scheduler.time() + duration(time_dis(rd));

    expireBuckets(dht4.buckets);
    expireBuckets(dht6.buckets);
    expireStore();
    expireSearches();
    scheduler.add(expire_stuff_time, std::bind(&Dht::expire, this));
}

void
Dht::onConnected()
{
    stopBootstrap();

    auto restartSearches = [this](std::map<InfoHash, Sp<Search>>& searches) {
        const auto& now = scheduler.time();
        for (auto& srp : searches) {
            auto& sr = *srp.second;
            if (sr.done)
                continue;
            scheduler.edit(sr.nextSearchStep, now);
        }
    };
    restartSearches(dht4.searches);
    restartSearches(dht6.searches);
    auto callbacks = std::move(onConnectCallbacks_);
    while (not callbacks.empty()) {
        callbacks.front()();
        callbacks.pop();
    }
}

void
Dht::onDisconnected()
{
    if (not bootstrapJob)
        bootstrap();
}

void
Dht::bootstrap()
{
    if (dht4.status != NodeStatus::Disconnected || dht6.status != NodeStatus::Disconnected) {
        // The job that led here has already run, and the scheduler does not
        // clear the handle to a job it has executed. onDisconnected() reads a
        // set handle as "a bootstrap is already scheduled" and does nothing,
        // so leaving it set here means the node never tries to rejoin the
        // network once it has lost it. Clearing it is what says that no
        // bootstrap is pending any more.
        bootstrapJob.reset();
        return;
    }
    if (logger_)
        logger_->debug("[{}] Bootstraping", myid.to_view());
    bootstrap_pending = false;
    for (const auto& boootstrap : bootstrap_nodes) {
        try {
            auto ips = network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second);
            for (auto& ip : ips) {
                if (ip.getPort() == 0)
                    ip.setPort(net::DHT_DEFAULT_PORT);
                pingNode(ip);
            }
        } catch (const std::exception& e) {
            if (logger_)
                logger_->error("[{}] Unable to resolve {}:{}: {}",
                               myid.to_view(),
                               boootstrap.first,
                               boootstrap.second,
                               e.what());
        }
    }
    scheduler.cancel(bootstrapJob);
    bootstrapJob = scheduler.add(scheduler.time() + bootstrap_period, std::bind(&Dht::bootstrap, this));
    bootstrap_period = std::min(bootstrap_period * 2, BOOTSTRAP_PERIOD_MAX);
}

void
Dht::startBootstrap()
{
    stopBootstrap();
    bootstrap_pending = !bootstrap_nodes.empty();
    bootstrapJob = scheduler.add(scheduler.time(), std::bind(&Dht::bootstrap, this));
}

void
Dht::stopBootstrap()
{
    scheduler.cancel(bootstrapJob);
    bootstrap_pending = false;
    bootstrap_period = BOOTSTRAP_PERIOD;
}

void
Dht::confirmNodes()
{
    using namespace std::chrono;
    bool soon = false;
    const auto& now = scheduler.time();

    if (dht4.searches.empty() and dht4.status == NodeStatus::Connected) {
        if (logger_)
            logger_->debug("[confirm nodes] Initial IPv4 'get' for my id ({})", myid.to_view());
        search(myid, AF_INET);
    }
    if (dht6.searches.empty() and dht6.status == NodeStatus::Connected) {
        if (logger_)
            logger_->debug("[confirm nodes] Initial IPv6 'get' for my id ({})", myid.to_view());
        search(myid, AF_INET6);
    }

    soon |= bucketMaintenance(dht4.buckets);
    soon |= bucketMaintenance(dht6.buckets);

    if (!soon) {
        if (dht4.buckets.grow_time >= now - seconds(150))
            soon |= neighbourhoodMaintenance(dht4.buckets);
        if (dht6.buckets.grow_time >= now - seconds(150))
            soon |= neighbourhoodMaintenance(dht6.buckets);
    }

    /* In order to maintain all buckets' age within 600 seconds, worst
       case is roughly 27 seconds, assuming the table is 22 bits deep.
       We want to keep a margin for neighborhood maintenance, so keep
       this within 25 seconds. */
    auto time_dis = soon ? uniform_duration_distribution<> {seconds(5), seconds(25)}
                         : uniform_duration_distribution<> {seconds(60), seconds(180)};
    auto confirm_nodes_time = now + time_dis(rd);

    scheduler.edit(nextNodesConfirmation, confirm_nodes_time);
}

std::vector<ValuesExport>
Dht::exportValues() const
{
    const auto& now = scheduler.time();
    const auto system_now = system_clock::now();

    std::vector<ValuesExport> e {};
    e.reserve(store.size());
    for (const auto& h : store) {
        ValuesExport ve;
        ve.first = h.first;

        msgpack::sbuffer buffer;
        msgpack::packer<msgpack::sbuffer> pk(&buffer);
        const auto& vals = h.second.getValues();
        pk.pack_array(vals.size());
        for (const auto& v : vals) {
            pk.pack_array(4);
            pk.pack(to_system_time(v.created, now, system_now));
            v.data->msgpack_pack(pk);
            if (const auto& store_addr = v.store_bucket ? v.store_bucket->getAddr() : SockAddr {}) {
                pk.pack_bin(store_addr.getLength());
                pk.pack_bin_body(reinterpret_cast<const char*>(store_addr.get()), store_addr.getLength());
            } else {
                pk.pack_bin(0);
            }
            pk.pack(to_system_time(v.expiration, now, system_now));
        }
        ve.second = {buffer.data(), buffer.data() + buffer.size()};
        e.push_back(std::move(ve));
    }
    return e;
}

void
Dht::importValues(const std::vector<ValuesExport>& import)
{
    const auto& now = scheduler.time();
    const auto system_now = system_clock::now();

    unsigned imported = 0, ignored = 0;
    size_t imported_size = 0;
    for (const auto& value : import) {
        if (value.second.empty()) {
            ignored++;
            continue;
        }

        try {
            msgpack::unpacked msg;
            msgpack::unpack(msg, (const char*) value.second.data(), value.second.size());
            auto valarr = msg.get();
            if (valarr.type != msgpack::type::ARRAY)
                throw msgpack::type_error();
            for (unsigned i = 0; i < valarr.via.array.size; i++) {
                auto& valel = valarr.via.array.ptr[i];
                if (valel.type != msgpack::type::ARRAY or valel.via.array.size < 2)
                    throw msgpack::type_error();
                time_point val_time;
                time_point expiration = time_point::min();
                Value tmp_val;
                SockAddr store_addr;
                try {
                    const bool uses_system_time = valel.via.array.ptr[0].type == msgpack::type::EXT;
                    if (uses_system_time) {
                        val_time = from_system_time(valel.via.array.ptr[0].as<system_clock::time_point>(),
                                                    now,
                                                    system_now);
                    } else {
                        val_time = time_point {
                            time_point::duration {valel.via.array.ptr[0].as<time_point::duration::rep>()}};
                    }
                    tmp_val.msgpack_unpack(valel.via.array.ptr[1]);
                    if (valel.via.array.size >= 3) {
                        auto addr_blob = valel.via.array.ptr[2].as<Blob>();
                        if (not addr_blob.empty()) {
                            store_addr = SockAddr(reinterpret_cast<const sockaddr*>(addr_blob.data()),
                                                  static_cast<socklen_t>(addr_blob.size()));
                        }
                    }
                    if (valel.via.array.size >= 4) {
                        if (uses_system_time) {
                            expiration = from_system_time(valel.via.array.ptr[3].as<system_clock::time_point>(),
                                                          now,
                                                          system_now);
                        } else {
                            expiration = time_point {
                                time_point::duration {valel.via.array.ptr[3].as<time_point::duration::rep>()}};
                        }
                    }
                } catch (const std::exception&) {
                    if (logger_)
                        logger_->error("Error reading value at {}", value.first.to_view());
                    ignored++;
                    continue;
                }
                val_time = std::min(val_time, now);
                auto val_size = tmp_val.size();
                if (storageStore(value.first,
                                 std::make_shared<Value>(std::move(tmp_val)),
                                 val_time,
                                 store_addr,
                                 false,
                                 expiration)) {
                    imported++;
                    imported_size += val_size;
                } else {
                    ignored++;
                }
            }
        } catch (const std::exception&) {
            ignored++;
            if (logger_)
                logger_->error("Error reading values at {}", value.first.to_view());
            continue;
        }
    }
    if (logger_)
        logger_->debug("Imported {} values, {}, ignored {}", imported, dht::printByteCount(imported_size), ignored);
}

std::vector<NodeExport>
Dht::exportNodes() const
{
    const auto& now = scheduler.time();
    std::vector<NodeExport> nodes;
    const auto b4 = dht4.buckets.findBucket(myid);
    if (b4 != dht4.buckets.end()) {
        for (auto& n : b4->nodes)
            if (n->isGood(now))
                nodes.emplace_back(n->exportNode());
    }
    const auto b6 = dht6.buckets.findBucket(myid);
    if (b6 != dht6.buckets.end()) {
        for (auto& n : b6->nodes)
            if (n->isGood(now))
                nodes.emplace_back(n->exportNode());
    }
    for (auto b = dht4.buckets.begin(); b != dht4.buckets.end(); ++b) {
        if (b == b4)
            continue;
        for (auto& n : b->nodes)
            if (n->isGood(now))
                nodes.emplace_back(n->exportNode());
    }
    for (auto b = dht6.buckets.begin(); b != dht6.buckets.end(); ++b) {
        if (b == b6)
            continue;
        for (auto& n : b->nodes)
            if (n->isGood(now))
                nodes.emplace_back(n->exportNode());
    }
    return nodes;
}

void
Dht::insertNode(const InfoHash& id, const SockAddr& addr)
{
    if (addr.getFamily() != AF_INET && addr.getFamily() != AF_INET6)
        return;
    scheduler.syncTime();
    network_engine.insertNode(id, addr);
}

void
Dht::pingNode(SockAddr sa, DoneCallbackSimple&& cb)
{
    scheduler.syncTime();
    if (logger_)
        logger_->debug("Sending ping to {}", sa);
    auto& count = dht(sa.getFamily()).pending_pings;
    count++;
    network_engine.sendPing(
        std::move(sa),
        [&count, cb](const net::Request&, net::RequestAnswer&&) {
            count--;
            if (cb)
                cb(true);
        },
        [&count, cb](const net::Request&, bool last) {
            if (last) {
                count--;
                if (cb)
                    cb(false);
            }
        });
}

void
Dht::onError(Sp<net::Request> req, net::DhtProtocolException e)
{
    const auto& node = req->node;
    if (e.getCode() == net::DhtProtocolException::UNAUTHORIZED) {
        if (logger_)
            logger_->error("[node {}] Token flush", node->toString());
        node->authError();
        for (auto& srp : searches(node->getFamily())) {
            auto& sr = srp.second;
            for (auto& n : sr->nodes) {
                if (n->node != node)
                    continue;
                n->token.reset();
                n->last_get_reply = time_point::min();
                for (auto& listen : n->listenStatus) {
                    if (listen.second.req and listen.second.req->pending())
                        n->node->cancelRequest(listen.second.req);
                    listen.second.req = {};
                    n->node->closeSocket(listen.second.socketId);
                    if (listen.second.refresh) {
                        listen.second.refresh->cancel();
                        listen.second.refresh = {};
                    }
                    listen.second.cache.onSynced(false);
                }
                searchSendGetValues(sr);
                scheduler.edit(sr->nextSearchStep, scheduler.time());
                break;
            }
        }
    } else if (e.getCode() == net::DhtProtocolException::NOT_FOUND) {
        if (logger_)
            logger_->error("[node {}] Returned error 404: storage not found", node->toString());
        node->cancelRequest(req);
    }
}

void
Dht::onReportedAddr(const InfoHash& /*id*/, const SockAddr& addr)
{
    if (addr)
        reportedAddr(addr);
}

net::RequestAnswer
Dht::onPing(Sp<Node>)
{
    return {};
}

net::RequestAnswer
Dht::onFindNode(Sp<Node>, const InfoHash& target, want_t want)
{
    const auto& now = scheduler.time();
    net::RequestAnswer answer;
    if (want & WANT4)
        answer.nodes4 = dht4.buckets.findClosestNodes(target, now, TARGET_NODES);
    if (want & WANT6)
        answer.nodes6 = dht6.buckets.findClosestNodes(target, now, TARGET_NODES);
    return answer;
}

net::RequestAnswer
Dht::onGetValues(Sp<Node> node, const InfoHash& hash, want_t, const Query& query)
{
    if (not hash) {
        if (logger_)
            logger_->warn("[node {}] Eek! Got get_values with no info_hash", node->toString());
        throw net::DhtProtocolException {net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
                                         net::DhtProtocolException::GET_NO_INFOHASH};
    }
    const auto& now = scheduler.time();
    net::RequestAnswer answer {};
    auto st = store.find(hash);
    answer.nodes4 = dht4.buckets.findClosestNodes(hash, now, TARGET_NODES);
    answer.nodes6 = dht6.buckets.findClosestNodes(hash, now, TARGET_NODES);
    if (st != store.end() && not st->second.empty()) {
        answer.values = st->second.get(query.where.getFilter());
        if (logger_)
            logger_->debug("[node {}] Sending {} values", node->toString(), answer.values.size());
    }
    return answer;
}

void
Dht::onGetValuesDone(const Sp<Node>& node, net::RequestAnswer& a, Sp<Search>& sr, const Sp<Query>& orig_query)
{
    if (not sr) {
        if (logger_)
            logger_->warn("[search unknown] Got reply to 'get'. Ignoring.");
        return;
    }

    /* if (logger_)
           logger_->d(sr->id, "[search %s] [node %s] Got reply to 'get' with %u nodes",
            sr->id.toString().c_str(), node->toString().c_str(), a.nodes4.size()+a.nodes6.size());*/

    if (a.ntoken) {
        if (not a.values.empty() or not a.fields.empty()) {
            if (logger_)
                logger_->debug("[search {}] [node {}] Found {} values",
                               sr->id.to_view(),
                               node->toString(),
                               a.values.size());
            for (auto& getp : sr->callbacks) { /* call all callbacks for this search */
                auto& get = getp.second;
                if (not(get.get_cb or get.query_cb)
                    or (orig_query and get.query and not get.isSatisfiedBy(*orig_query)))
                    continue;

                if (get.query_cb) { /* in case of a request with query */
                    if (not a.fields.empty()) {
                        get.query_cb(a.fields);
                    } else if (not a.values.empty()) {
                        std::vector<Sp<FieldValueIndex>> fields;
                        fields.reserve(a.values.size());
                        for (const auto& v : a.values)
                            fields.emplace_back(
                                std::make_shared<FieldValueIndex>(*v, orig_query ? orig_query->select : Select {}));
                        get.query_cb(fields);
                    }
                } else if (get.get_cb) { /* in case of a vanilla get request */
                    std::vector<Sp<Value>> tmp;
                    for (const auto& v : a.values)
                        if (not get.filter or get.filter(*v))
                            tmp.emplace_back(v);
                    if (not tmp.empty())
                        get.get_cb(tmp);
                }
            }

            /* callbacks for local search listeners */
            /*std::vector<std::pair<ValueCallback, std::vector<Sp<Value>>>> tmp_lists;
            for (auto& l : sr->listeners) {
                if (!l.second.get_cb or (orig_query and l.second.query and not
            l.second.query->isSatisfiedBy(*orig_query))) continue; std::vector<Sp<Value>> tmp; for (const auto& v :
            a.values) if (not l.second.filter or l.second.filter(*v)) tmp.emplace_back(v); if (not tmp.empty())
                    tmp_lists.emplace_back(l.second.get_cb, std::move(tmp));
            }
            for (auto& l : tmp_lists)
                l.first(l.second, false);*/
        } else if (not a.expired_values.empty()) {
            if (logger_)
                logger_->warn("[search {}] [node {}] {} expired values",
                              sr->id.toString(),
                              node->toString(),
                              a.expired_values.size());
        }
    } else {
        if (logger_)
            logger_->warn("[node {}] No token provided. Ignoring response content.", node->toString());
        network_engine.blacklistNode(node);
    }

    if (not sr->done) {
        searchSendGetValues(sr);

        // Force to recompute the next step time
        scheduler.edit(sr->nextSearchStep, scheduler.time());
    }
}

net::RequestAnswer
Dht::onListen(Sp<Node> node, const InfoHash& hash, size_t socket_id, const Query& query, int version)
{
    if (not hash) {
        if (logger_)
            logger_->warn("[node {}] Listen with no info_hash", node->toString());
        throw net::DhtProtocolException {net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
                                         net::DhtProtocolException::LISTEN_NO_INFOHASH};
    }
    Query q = query;
    storageAddListener(hash, node, socket_id, std::move(q), version);
    return {};
}

void
Dht::onListenDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>& sr)
{
    // if (logger_)
    //     logger_->d(sr->id, node->id, "[search %s] [node %s] Got listen confirmation",
    //            sr->id.toString().c_str(), node->toString().c_str(), answer.values.size());

    if (not sr->done) {
        const auto& now = scheduler.time();
        if (answer.ntoken)
            if (auto sn = sr->getNode(node))
                sn->refreshToken(*answer.ntoken);
        searchSendGetValues(sr);
        scheduler.edit(sr->nextSearchStep, now);
    }
}

net::RequestAnswer
Dht::onAnnounce(Sp<Node> n,
                const InfoHash& hash,
                const std::vector<Sp<Value>>& values,
                const time_point& creation_date)
{
    auto& node = *n;
    if (not hash) {
        if (logger_)
            logger_->warn("Put with no info_hash");
        throw net::DhtProtocolException {net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
                                         net::DhtProtocolException::PUT_NO_INFOHASH};
    }
    {
        // We store a value only if we think we're part of the
        // SEARCH_NODES nodes around the target id.
        auto closest_nodes = buckets(node.getFamily()).findClosestNodes(hash, scheduler.time(), SEARCH_NODES);
        if (closest_nodes.size() >= TARGET_NODES and hash.xorCmp(closest_nodes.back()->id, myid) < 0) {
            if (logger_)
                logger_->warn("[node {}] Announce too far from the target. Dropping value.", node.toString());
            return {};
        }
    }

    auto created = std::min(creation_date, scheduler.time());
    for (const auto& v : values) {
        if (v->id == Value::INVALID_ID) {
            if (logger_)
                logger_->warn("[value {}] Incorrect value id", hash.to_view());
            throw net::DhtProtocolException {net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
                                             net::DhtProtocolException::PUT_INVALID_ID};
        }
        auto lv = getLocalById(hash, v->id);
        Sp<Value> vc = v;
        if (lv) {
            if (*lv == *vc) {
                storageRefresh(hash, v->id);
                if (logger_)
                    logger_->debug("[store {}] [node {}] Refreshed value {:016x}",
                                   hash.toString(),
                                   node.toString(),
                                   v->id);
            } else {
                const auto& type = getType(lv->type);
                if (type.editPolicy(hash, lv, vc, node.id, node.getAddr())) {
                    if (logger_)
                        logger_->debug("[store {}] Editing {}", hash.toString(), vc->toString());
                    storageStore(hash, vc, created, node.getAddr());
                } else {
                    if (logger_)
                        logger_->debug("[store {}] Rejecting edition of {} because of storage policy",
                                       hash.toString(),
                                       vc->toString());
                }
            }
        } else {
            // Allow the value to be edited by the storage policy
            const auto& type = getType(vc->type);
            if (type.storePolicy(hash, vc, node.id, node.getAddr())) {
                // if (logger_)
                //     logger_->d(hash, node.id, "[store %s] Storing %s", hash.toString().c_str(),
                //     std::to_string(vc->id).c_str());
                storageStore(hash, vc, created, node.getAddr());
            } else {
                if (logger_)
                    logger_->debug("[store {}] Rejecting storage of {}", hash.toString(), vc->toString());
            }
        }
    }
    return {};
}

net::RequestAnswer
Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Value::Id& vid)
{
    using namespace net;

    if (storageRefresh(hash, vid)) {
        if (logger_)
            logger_->debug("[store {}] [node {}] Refreshed value {:016x}", hash.toString(), node->toString(), vid);
    } else {
        if (logger_)
            logger_->debug("[store {}] [node {}] Got refresh for unknown value", hash.toString(), node->toString());
        throw DhtProtocolException {DhtProtocolException::NOT_FOUND, DhtProtocolException::STORAGE_NOT_FOUND};
    }
    return {};
}

bool
Dht::storageRefresh(const InfoHash& id, Value::Id vid)
{
    const auto& now = scheduler.time();
    auto s = store.find(id);
    if (s != store.end()) {
        // Values like for a permanent put can be refreshed. So, inform remote listeners that the value
        // need to be refreshed
        auto& st = s->second;
        if (not st.listeners.empty()) {
            if (logger_)
                logger_->debug("[store {}] {} remote listeners", id.to_view(), st.listeners.size());
            std::vector<Value::Id> ids = {vid};
            for (const auto& node_listeners : st.listeners) {
                for (const auto& l : node_listeners.second) {
                    if (logger_)
                        logger_->warn("[store {}] [node {}] Sending refresh",
                                      id.toString(),
                                      node_listeners.first->toString());
                    network_engine.tellListenerRefreshed(node_listeners.first, l.first, id, ids, l.second.version);
                }
            }
        }

        auto expiration = s->second.refresh(id, now, vid, types);
        if (expiration.first) {
            scheduler.cancel(expiration.first->expiration_job);
            if (expiration.second != time_point::max()) {
                expiration.first->expiration_job = scheduler.add(expiration.second,
                                                                 std::bind(&Dht::expireStorage, this, id));
            }
        }
        return true;
    }
    return false;
}

void
Dht::onAnnounceDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>& sr)
{
    if (logger_)
        logger_->debug("[search {}] [node {}] Got reply to put!", sr->id.toString(), node->toString());
    searchSendGetValues(sr);
    sr->checkAnnounced(answer.vid);
}

struct DhtState
{
    unsigned v {1};
    InfoHash id;
    std::vector<NodeExport> nodes;
    std::vector<ValuesExport> values;
    /* secret transport key, if generated by the node */
    Blob transport_key;

    MSGPACK_DEFINE_MAP(v, id, nodes, values, transport_key)
};

void
Dht::saveState(const std::string& path) const
{
    DhtState state;
    state.id = myid;
    state.nodes = exportNodes();
    state.values = exportValues();
    auto key = persistTransportKey ? network_engine.getTransportKey() : nullptr;
    if (key)
        state.transport_key.assign(key->secret.begin(), key->secret.end());
    auto temporary = path + "." + InfoHash::getRandom().toString() + ".tmp";
    try {
        std::ofstream file;
        file.exceptions(std::ios::failbit | std::ios::badbit);
        file.open(temporary, std::ios::binary | std::ios::trunc);
        if (key)
            std::filesystem::permissions(temporary,
                                         std::filesystem::perms::owner_read | std::filesystem::perms::owner_write);
        msgpack::pack(file, state);
        file.close();
        std::filesystem::rename(temporary, path);
    } catch (const std::exception& e) {
        std::error_code ec;
        std::filesystem::remove(temporary, ec);
        if (logger_)
            logger_->warn("Error saving state to {}: {}", path, e.what());
    }
}

void
Dht::loadState(const std::string& path)
{
    if (logger_)
        logger_->debug("Importing state from {}", path);
    try {
        // Import nodes from binary file
        msgpack::unpacker pac;
        {
            // Read whole file
            std::ifstream file(path, std::ios::binary | std::ios::ate);
            if (!file.is_open()) {
                return;
            }
            auto size = file.tellg();
            file.seekg(0, std::ios::beg);
            pac.reserve_buffer(size);
            file.read(pac.buffer(), size);
            pac.buffer_consumed(size);
        }
        // Import nodes
        msgpack::object_handle oh;
        if (pac.next(oh)) {
            auto state = oh.get().as<DhtState>();
            if (logger_)
                logger_->debug("Importing {} nodes", state.nodes.size());
            if (persistTransportKey and state.transport_key.size() == transport::KEY_SIZE) {
                transport::Key secret;
                std::copy(state.transport_key.begin(), state.transport_key.end(), secret.begin());
                auto key = transport::KeyPair::fromSecret(secret);
                network_engine.setTransportKey(key);
                myid = transport::nodeId(key.pub);
            } else if (state.id and not network_engine.getTransportKey()) {
                myid = state.id;
            }
            std::vector<Sp<Node>> tmpNodes;
            tmpNodes.reserve(state.nodes.size());
            for (const auto& node : state.nodes)
                tmpNodes.emplace_back(network_engine.insertNode(node.id, node.addr));
            importValues(state.values);
        }
    } catch (const std::exception& e) {
        if (logger_)
            logger_->warn("Error importing state from {}: {}", path, e.what());
    }
}

} // namespace dht