ores.cpp.nats-event-cache.cache_header

{{! GENERATED FILE — tangled from projects/ores.codegen/library/templates/ores.cpp.nats-event-cache.cache_header.org. Edit the org source. }}
{{{cpp_license}}}
{{#domain_entity}}
#ifndef ORES_{{cache_component_upper}}_{{cache_subcomponent_upper}}_SERVICE_CACHE_{{entity_singular_upper}}_CACHE_HPP
#define ORES_{{cache_component_upper}}_{{cache_subcomponent_upper}}_SERVICE_CACHE_{{entity_singular_upper}}_CACHE_HPP

#include "ores.eventing.core/service/cache/partition_token_cache.hpp"
#include "ores.eventing.core/service/cache/partitioned_cache.hpp"
#include "ores.logging/make_logger.hpp"
#include "ores.nats/domain/headers.hpp"
#include "ores.nats/domain/message.hpp"
#include "ores.nats/domain/wire_codec.hpp"
#include "ores.nats/service/client.hpp"
#include "ores.{{component_include}}/domain/{{entity_singular}}.hpp"
#include "ores.{{component_include}}/messaging/{{entity_singular}}_protocol.hpp"
{{#primary_key.is_uuid}}
#include <boost/container_hash/hash.hpp>
#include <boost/uuid/uuid.hpp>
{{^cache_aux_type}}
#include <variant>
{{/cache_aux_type}}
{{/primary_key.is_uuid}}
{{#cache_aux_type}}
#include <immer/map.hpp>
#include <immer/map_transient.hpp>
#include <vector>
{{/cache_aux_type}}
#include <cstdint>
#include <functional>
#include <optional>
#include <string>
#include <unordered_map>
#include <utility>

namespace ores::{{cache_component}}::service::cache {

namespace {
inline auto& {{entity_singular}}_cache_lg() {
    static auto instance = ores::logging::make_logger(
        "ores.{{cache_component}}.service.cache.{{entity_singular}}_cache");
    return instance;
}
}

/**
 * @brief In-process per-tenant cache of {{component}} {{entity_plural}} data.
 *
 * Populated by paging {{component}}'s list_{{entity_plural}} read to its
 * end at startup, and reloaded on receipt of a {{entity_singular}} event
 * for the affected tenant (see the nats-event-cache.registrar_wiring
 * archetype). See the "Generic entity-mirror cache primitive + codegen
 * facet" story.
 *
 * Each partition is one tenant's copy, so it is read inside that tenant:
 * the @c token_provider gives a token for the partition's tenant, obtained
 * by Token Exchange (see partition_token_cache), and row-level security
 * returns exactly the rows that tenant may see. A target that answers
 * @c token_expired gets one renewed token and one retry. Omit the provider
 * only against a producer that has not opted into the auth check.
 */
class {{entity_singular}}_cache {
{{#primary_key.is_uuid}}
    using key_hash = boost::hash<{{primary_key.cpp_type}}>;
{{/primary_key.is_uuid}}
<<paste:ADD930C7-EF33-44B4-A0A2-49254FCC7ACD>>
{{#primary_key.is_uuid}}
    using cache_t = ores::eventing::service::cache::partitioned_cache<std::string,
        {{primary_key.cpp_type}}, ores::{{component}}::domain::{{entity_singular}},
        {{#cache_aux_type}}{{cache_aux_type}}{{/cache_aux_type}}{{^cache_aux_type}}std::monostate{{/cache_aux_type}},
        key_hash>;
{{/primary_key.is_uuid}}
{{^primary_key.is_uuid}}
    using cache_t = ores::eventing::service::cache::partitioned_cache<std::string,
        {{primary_key.cpp_type}}, ores::{{component}}::domain::{{entity_singular}}
        {{#cache_aux_type}}, {{cache_aux_type}}{{/cache_aux_type}}>;
{{/primary_key.is_uuid}}

public:
    explicit {{entity_singular}}_cache(ores::nats::service::client& nats,
        ores::eventing::service::cache::partition_token_provider token_provider = nullptr)
        : nats_(nats)
        , token_provider_(std::move(token_provider)) {}

    {{entity_singular}}_cache(const {{entity_singular}}_cache&) = delete;
    {{entity_singular}}_cache& operator=(const {{entity_singular}}_cache&) = delete;
    {{entity_singular}}_cache({{entity_singular}}_cache&&) = delete;
    {{entity_singular}}_cache& operator=({{entity_singular}}_cache&&) = delete;

    /**
     * @brief Arms (or replaces) the token provider after construction. For a
     * consumer that must construct this cache during its own Phase 1
     * (wiring/readiness) but can only mint a token during Phase 2
     * (post-readiness) — e.g. a self-referential consumer authenticating
     * against its own just-registered subjects. See "Service Bootstrap
     * Phases" in the architecture docs.
     */
    void set_token_provider(
        ores::eventing::service::cache::partition_token_provider token_provider) {
        token_provider_ = std::move(token_provider);
    }

    /**
     * @brief Loads (or reloads) one tenant's partition.
     *
     * A full read is the canonical list paged to its end: the protocol has
     * no bulk read, so the warm-up walks page by page until the response's
     * total is reached.
     *
     * @return An empty string on success. On failure, a human-readable
     * message describing what went wrong -- the failure is still logged
     * (@c warn) either way, but callers that can surface it to a user
     * (e.g. a UI's status/error signal) need it returned, not just
     * buried in a log a user never sees. Silently-swallowed background
     * failures are the right default for the warm-up/refresh path
     * (@c nats-event-cache.registrar_wiring calls this fire-and-forget
     * from a detached thread), but the return value lets any caller
     * that *can* react to a failure opt in.
     */
    [[nodiscard]] std::string load(const std::string& tenant_id) {
        using namespace ores::logging;
        try {
            const auto& codec = ores::nats::default_wire_codec();
            std::unordered_map<std::string, std::string> headers;
            const auto authorise = [&](bool renew) {
                if (!token_provider_)
                    return true;
                const auto token = token_provider_(tenant_id, renew);
                if (token.empty())
                    return false;
                headers[std::string(ores::nats::headers::authorization)] =
                    std::string(ores::nats::headers::bearer_prefix) + token;
                return true;
            };
            if (!authorise(false)) {
                const std::string msg = "no token acts inside this tenant";
                BOOST_LOG_SEV({{entity_singular}}_cache_lg(), warn)
                    << "{{entity_pascal}} cache load failed for tenant " << tenant_id << ": "
                    << msg;
                return msg;
            }
            bool renewed = false;
            auto entries_t = cache_t::entries_map{}.transient();
            std::uint64_t count = 0;
            constexpr std::uint32_t page_size = 100;
            std::uint32_t offset = 0;
            while (true) {
                const auto bytes = codec.encode(
                    ores::{{component}}::messaging::list_{{entity_plural}}_request{
                        .offset = offset, .limit = page_size});
                const auto reply = nats_.request_sync(
                    ores::{{component}}::messaging::list_{{entity_plural}}_request::nats_subject,
                    bytes, headers);
                const auto refusal = reply.headers.find(std::string(ores::nats::headers::x_error));
                if (refusal != reply.headers.end() && refusal->second == "token_expired" &&
                    !renewed && authorise(true)) {
                    renewed = true;
                    continue;
                }
                auto resp = codec.decode<
                    ores::{{component}}::messaging::list_{{entity_plural}}_response>(reply.data);
                if (!resp || resp->result.outcome != ores::utility::domain::outcome::ok) {
                    const auto msg = resp ? resp->result.message
                                          : "parse error (malformed or error reply)";
                    BOOST_LOG_SEV({{entity_singular}}_cache_lg(), warn)
                        << "{{entity_pascal}} cache load failed for tenant " << tenant_id << ": "
                        << msg;
                    return msg;
                }
                const auto page_count = resp->{{entity_plural_short}}.size();
                for (auto& v : resp->{{entity_plural_short}})
                    entries_t.set(v.{{primary_key.column}}, std::move(v));
                count += page_count;
                if (page_count == 0 || count >= resp->total)
                    break;
                offset += page_size;
            }
            auto entries = entries_t.persistent();
<<paste:60520267-94E5-4369-97E9-7AED54CDEDD2>>
            cache_.replace_partition(tenant_id, entries{{#cache_aux_type}}, aux{{/cache_aux_type}});
            BOOST_LOG_SEV({{entity_singular}}_cache_lg(), debug)
                << "Loaded " << count << " {{entity_plural}} for tenant " << tenant_id;
            return {};
        } catch (const std::exception& e) {
            BOOST_LOG_SEV({{entity_singular}}_cache_lg(), warn)
                << "{{entity_pascal}} cache load exception for tenant " << tenant_id << ": "
                << e.what();
            return e.what();
        }
    }

    std::optional<ores::{{component}}::domain::{{entity_singular}}>
    lookup(const std::string& tenant_id, const {{primary_key.cpp_type}}& key) const {
        return cache_.get(tenant_id, key);
    }

<<paste:C0EDFBDF-F38F-4E02-874F-A086FB435FB0>>
private:
    ores::nats::service::client& nats_;
    ores::eventing::service::cache::partition_token_provider token_provider_;
    cache_t cache_;
};

}

#endif
{{/domain_entity}}