opendht/src/op_cache.h

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

#include "value.h"
#include "value_cache.h"
#include "listener.h"

namespace dht {

struct OpCacheValueStorage
{
    Sp<Value> data {};
    unsigned refCount {1};
    system_clock::time_point updated {system_clock::time_point::min()};
    explicit OpCacheValueStorage(Sp<Value> val)
        : data(std::move(val))
    {}
};

class OpValueCache
{
public:
    /**
     * @param countSources  When true, a value is kept until every source
     *                      that reported it has expired it (per-node caches,
     *                      IPv4/IPv6 searches, local storage). When false,
     *                      re-deliveries of the same value are not counted
     *                      and a single expiration removes it (single
     *                      logical source, e.g. a proxy server).
     */
    OpValueCache(ValueCallback&& cb, bool countSources = true) noexcept
        : callback(std::forward<ValueCallback>(cb))
        , countSources(countSources)
    {}
    explicit OpValueCache(OpValueCache&& o) noexcept
        : values(std::move(o.values))
        , callback(std::move(o.callback))
        , countSources(o.countSources)
    {
        o.callback = {};
    }

    static ValueCallback cacheCallback(ValueCallback&& cb, std::function<void()>&& onCancel, bool countSources = true)
    {
        return [cache = std::make_shared<OpValueCache>(std::forward<ValueCallback>(cb), countSources),
                onCancel = std::move(onCancel)](const std::vector<Sp<Value>>& vals, bool expired) {
            auto ret = cache->onValue(vals, expired);
            if (not ret)
                onCancel();
            return ret;
        };
    }

    bool onValue(const std::vector<Sp<Value>>& vals,
                 bool expired,
                 const system_clock::time_point& t = system_clock::time_point::min())
    {
        return expired ? onValuesExpired(vals, t) : onValuesAdded(vals, t);
    }

    bool onValuesAdded(const std::vector<Sp<Value>>& vals,
                       const system_clock::time_point& t = system_clock::time_point::min());
    bool onValuesExpired(const std::vector<Sp<Value>>& vals,
                         const system_clock::time_point& t = system_clock::time_point::min());
    bool onValuesExpired(const std::vector<Value::Id>& vals,
                         const system_clock::time_point& t = system_clock::time_point::min());

    void onNodeChanged(ListenSyncStatus status)
    {
        switch (status) {
        case ListenSyncStatus::ADDED:
            nodes++;
            break;
        case ListenSyncStatus::REMOVED:
            nodes--;
            break;
        case ListenSyncStatus::SYNCED:
            syncedNodes++;
            break;
        case ListenSyncStatus::UNSYNCED:
            syncedNodes--;
            break;
        }
    }

    bool isSynced() const { return nodes > 0 and syncedNodes == nodes; }

    std::vector<Sp<Value>> get(const Value::Filter& filter) const;
    Sp<Value> get(Value::Id id) const;
    std::vector<Sp<Value>> getValues() const;
    size_t size() const { return values.size(); }

private:
    OpValueCache(const OpValueCache&) = delete;
    OpValueCache& operator=(const OpValueCache&) = delete;

    size_t nodes {0};
    size_t syncedNodes {0};
    std::map<Value::Id, OpCacheValueStorage> values {};
    ValueCallback callback;
    bool countSources {true};
};

class OpCache
{
public:
    explicit OpCache(bool countSources = true)
        : cache(
              [this](const std::vector<Sp<Value>>& vals, bool expired) {
                  if (expired)
                      onValuesExpired(vals);
                  else
                      onValuesAdded(vals);
                  return true;
              },
              countSources)
    {}

    bool onValue(const std::vector<Sp<Value>>& vals, bool expired) { return cache.onValue(vals, expired); }
    void onNodeChanged(ListenSyncStatus status) { cache.onNodeChanged(status); }

    void onValuesAdded(const std::vector<Sp<Value>>& vals);
    void onValuesExpired(const std::vector<Sp<Value>>& vals);

    bool addListener(size_t token, const ValueCallback& cb, const Sp<Query>& q, Value::Filter&& filter)
    {
        auto cached = cache.get(filter);
        if (not cached.empty() and not cb(cached, false)) {
            return false;
        }
        listeners.emplace(token, LocalListener {q, std::move(filter), cb});
        return true;
    }

    bool removeListener(size_t token, const time_point& now)
    {
        lastRemoved = now;
        return listeners.erase(token) > 0;
    }

    void removeAll() { listeners.clear(); }

    bool isDone() { return listeners.empty(); }

    std::vector<Sp<Value>> get(const Value::Filter& filter) const { return cache.get(filter); }

    Sp<Value> get(Value::Id id) const { return cache.get(id); }

    bool isSynced() const { return cache.isSynced(); }

    bool isExpired(const time_point& now) const { return listeners.empty() and (lastRemoved + EXPIRATION < now); }
    time_point getExpiration() const;

    size_t size() const { return cache.size(); }

    size_t searchToken {0};

private:
    constexpr static const std::chrono::seconds EXPIRATION {60};
    OpCache(const OpCache&) = delete;
    OpCache& operator=(const OpCache&) = delete;

    OpValueCache cache;
    std::map<size_t, LocalListener> listeners;
    time_point lastRemoved {clock::now()};
};

class SearchCache
{
public:
    explicit SearchCache(bool countSources = true)
        : countSources_(countSources)
    {}
    SearchCache(SearchCache&&) = default;

    using OnListen = std::function<size_t(Sp<Query>, ValueCallback, SyncCallback)>;
    size_t listen(const ValueCallback& get_cb, const Sp<Query>& q, Value::Filter&& filter, const OnListen& onListen);

    bool cancelListen(size_t gtoken, const time_point& now);
    void cancelAll(const std::function<void(size_t)>& onCancel);
    void clear();

    time_point expire(const time_point& now, const std::function<void(size_t)>& onCancel);
    time_point getExpiration() const { return nextExpiration_; }

    bool get(const Value::Filter& f, const Sp<Query>& q, const GetCallback& gcb, const DoneCallback& dcb) const;
    std::vector<Sp<Value>> get(const Value::Filter& filter) const;
    Sp<Value> get(Value::Id id) const;

    std::pair<size_t, size_t> size() const
    {
        size_t tot = 0;
        for (const auto& c : ops)
            tot += c.second->size();
        return {ops.size(), tot};
    }

private:
    SearchCache(const SearchCache&) = delete;
    SearchCache& operator=(const SearchCache&) = delete;

    using OpMap = std::map<Sp<Query>, std::unique_ptr<OpCache>>;
    OpMap ops {};
    OpMap::iterator getOp(const Sp<Query>& q);
    OpMap::const_iterator getOp(const Sp<Query>& q) const;

    size_t nextToken_ {1};
    time_point nextExpiration_ {time_point::max()};
    bool countSources_ {true};
};

} // namespace dht