VersionedRegistry.h

浏览该文件的文档.

#pragma once
 
#include "common/BorrowedRef.h"
#include "common/EventSequence.h"
#include "common/Export.h"
#include "common/Generation.h"
#include "common/SchemaVersion.h"
#include "common/Subscription.h"
 
#include <algorithm>
#include <cstddef>
#include <exception>
#include <functional>
#include <map>
#include <memory>
#include <optional>
#include <ostream>
#include <string>
#include <type_traits>
#include <utility>
#include <variant>
#include <vector>
 
namespace eve {
namespace detail {
 
[[nodiscard]] EVENGINE_API eve::Result<void> validateRegistryNextSequence(EventSequence nextEventSequence);
 
[[nodiscard]] EVENGINE_API eve::Result<Generation> nextRegistryGeneration(Generation current, bool present);
 
[[nodiscard]] EVENGINE_API eve::Result<EventSequence> nextRegistryEventSequence(EventSequence current);
 
class EVENGINE_API EventLogValidator {
public:
    [[nodiscard]] eve::Result<void> accept(EventSequence sequence, Generation generation, bool remove, bool tombstone);
 
    [[nodiscard]] eve::Result<void> finish(EventSequence nextEventSequence) const;
 
private:
    EventSequence previous_{};
};
 
}  // namespace detail
 
 
enum class RegistryOperation : std::uint8_t {
    Insert,
    Replace,
    Remove,
};
 
inline const char* registryOperationName(RegistryOperation operation) noexcept {
    switch (operation) {
        case RegistryOperation::Insert: return "insert";
        case RegistryOperation::Replace: return "replace";
        case RegistryOperation::Remove: return "remove";
    }
    return "unknown";
}
 
inline std::ostream& operator<<(std::ostream& stream, RegistryOperation operation) {
    return stream << registryOperationName(operation);
}
 
template <class Key, class Value, class EventData = std::monostate, class Compare = std::less<Key>>
class VersionedRegistry {
    static_assert(std::is_copy_constructible_v<Key>, "VersionedRegistry keys must be copyable");
    static_assert(std::is_copy_constructible_v<Value>, "VersionedRegistry values must be copyable");
    static_assert(std::is_copy_constructible_v<EventData>, "VersionedRegistry event data must be copyable");
 
public:
    struct Handle {
        Key        key;
        Generation generation;
 
        [[nodiscard]] bool isValid() const noexcept { return !generation.isZero(); }
    };
 
    struct Event {
        EventSequence     sequence;
        RegistryOperation operation = RegistryOperation::Insert;
        Key               key;
        Generation        generation;
        bool              tombstone = false;
        std::string       label;
        EventData         data{};
 
        [[nodiscard]] bool isTombstone() const noexcept { return tombstone; }
    };
 
    struct Entry {
        Generation           generation;
        std::optional<Value> value;
    };
 
    struct State {
        std::map<Key, Entry, Compare> entries;
        std::vector<Event>            events;
        EventSequence                 nextEventSequence{1};
 
        void swap(State& other) noexcept {
            static_assert(std::is_nothrow_swappable_v<Compare>,
                          "VersionedRegistry comparator must be nothrow-swappable");
            entries.swap(other.entries);
            events.swap(other.events);
            const auto sequence     = nextEventSequence;
            nextEventSequence       = other.nextEventSequence;
            other.nextEventSequence = sequence;
        }
    };
 
    using GenerationProjector = std::function<void(Value&, Generation)>;
    using ChangeCallback      = std::function<void(const Event&)>;
 
    explicit VersionedRegistry(GenerationProjector generationProjector = {})
        : generationProjector_(std::move(generationProjector)), observers_() {}
 
    VersionedRegistry(const VersionedRegistry&)            = delete;
    VersionedRegistry& operator=(const VersionedRegistry&) = delete;
    VersionedRegistry(VersionedRegistry&&)                 = delete;
    VersionedRegistry& operator=(VersionedRegistry&&)      = delete;
    ~VersionedRegistry()                                   = default;
 
    [[nodiscard]] Result<Handle> insert(Key key, Value value, EventData data = {}, std::string label = {}) {
        try {
            return mutate(RegistryOperation::Insert, std::move(key), std::optional<Value>(std::move(value)),
                          std::move(data), std::move(label));
        } catch (const std::exception&) {
            return Result<Handle>::failure(
                Diagnostic::error(DiagnosticCode::Failed, "registry mutation argument preparation failed"));
        } catch (...) {
            return Result<Handle>::failure(
                Diagnostic::error(DiagnosticCode::Failed, "registry mutation argument preparation failed"));
        }
    }
 
    [[nodiscard]] Result<Handle> replace(Key key, Value value, EventData data = {}, std::string label = {}) {
        try {
            return mutate(RegistryOperation::Replace, std::move(key), std::optional<Value>(std::move(value)),
                          std::move(data), std::move(label));
        } catch (const std::exception&) {
            return Result<Handle>::failure(
                Diagnostic::error(DiagnosticCode::Failed, "registry mutation argument preparation failed"));
        } catch (...) {
            return Result<Handle>::failure(
                Diagnostic::error(DiagnosticCode::Failed, "registry mutation argument preparation failed"));
        }
    }
 
    [[nodiscard]] Result<Handle> remove(Key key, EventData data = {}, std::string label = {}) {
        try {
            return mutate(RegistryOperation::Remove, std::move(key), std::nullopt, std::move(data), std::move(label));
        } catch (const std::exception&) {
            return Result<Handle>::failure(
                Diagnostic::error(DiagnosticCode::Failed, "registry mutation argument preparation failed"));
        } catch (...) {
            return Result<Handle>::failure(
                Diagnostic::error(DiagnosticCode::Failed, "registry mutation argument preparation failed"));
        }
    }
 
    [[nodiscard]] ResultRef<const Value> resolve(const Key& key) const {
        const auto it = state_.entries.find(key);
        if (it == state_.entries.end() || !it->second.value.has_value())
            return Result<std::reference_wrapper<const Value>>::failure(
                Diagnostic::error(DiagnosticCode::NotFound, "registry key is not live"));
        return ResultRef<const Value>::success(std::cref(*it->second.value));
    }
 
    [[nodiscard]] ResultRef<const Value> resolve(const Handle& handle) const {
        const auto it = state_.entries.find(handle.key);
        if (it == state_.entries.end())
            return Result<std::reference_wrapper<const Value>>::failure(
                Diagnostic::error(DiagnosticCode::NotFound, "registry key is not present"));
        if (handle.generation.isZero() || it->second.generation != handle.generation || !it->second.value.has_value())
            return Result<std::reference_wrapper<const Value>>::failure(
                Diagnostic::error(DiagnosticCode::StaleHandle, "registry handle is stale"));
        return ResultRef<const Value>::success(std::cref(*it->second.value));
    }
 
    [[nodiscard]] Result<Handle> handle(const Key& key) const {
        const auto it = state_.entries.find(key);
        if (it == state_.entries.end() || !it->second.value.has_value())
            return Result<Handle>::failure(Diagnostic::error(DiagnosticCode::NotFound, "registry key is not live"));
        return Result<Handle>::success(Handle{key, it->second.generation});
    }
 
    [[nodiscard]] Result<Generation> generationOf(const Key& key) const {
        const auto it = state_.entries.find(key);
        if (it == state_.entries.end())
            return Result<Generation>::failure(
                Diagnostic::error(DiagnosticCode::NotFound, "registry key is not present"));
        return Result<Generation>::success(it->second.generation);
    }
 
    [[nodiscard]] bool contains(const Key& key) const noexcept {
        const auto it = state_.entries.find(key);
        return it != state_.entries.end() && it->second.value.has_value();
    }
 
    [[nodiscard]] bool isTombstone(const Key& key) const noexcept {
        const auto it = state_.entries.find(key);
        return it != state_.entries.end() && !it->second.value.has_value();
    }
 
    [[nodiscard]] bool isStale(const Handle& handle) const noexcept {
        const auto it = state_.entries.find(handle.key);
        return it == state_.entries.end() || handle.generation.isZero() || it->second.generation != handle.generation ||
               !it->second.value.has_value();
    }
 
    [[nodiscard]] std::size_t size() const noexcept {
        std::size_t count = 0;
        for (const auto& [key, entry] : state_.entries)
            if (entry.value.has_value()) ++count;
        return count;
    }
 
    [[nodiscard]] std::size_t tombstoneCount() const noexcept {
        std::size_t count = 0;
        for (const auto& [key, entry] : state_.entries)
            if (!entry.value.has_value()) ++count;
        return count;
    }
 
    [[nodiscard]] const Value* at(std::size_t index) const noexcept {
        for (const auto& [key, entry] : state_.entries) {
            if (!entry.value.has_value()) continue;
            if (index-- == 0) return &*entry.value;
        }
        return nullptr;
    }
 
    [[nodiscard]] const Key* keyAt(std::size_t index) const noexcept {
        for (const auto& [key, entry] : state_.entries) {
            if (!entry.value.has_value()) continue;
            if (index-- == 0) return &key;
        }
        return nullptr;
    }
 
    [[nodiscard]] const Event* eventAt(std::size_t index) const noexcept {
        return index < state_.events.size() ? &state_.events[index] : nullptr;
    }
 
    [[nodiscard]] std::size_t eventCount() const noexcept { return state_.events.size(); }
 
    void clearEvents() noexcept { state_.events.clear(); }
 
    [[nodiscard]] State snapshotState() const { return state_; }
 
    [[nodiscard]] Result<void> restoreState(State candidate) {
        auto sequenceOk = detail::validateRegistryNextSequence(candidate.nextEventSequence);
        if (!sequenceOk) return sequenceOk;
 
        try {
            for (auto& [key, entry] : candidate.entries) {
                if (entry.generation.isZero())
                    return Result<void>::failure(
                        Diagnostic::error(DiagnosticCode::ParseError, "registry entry generation must be positive"));
                if (entry.value.has_value()) generationProject(*entry.value, entry.generation);
            }
        } catch (const std::exception&) {
            return Result<void>::failure(
                Diagnostic::error(DiagnosticCode::Failed, "registry generation projection failed"));
        } catch (...) {
            return Result<void>::failure(
                Diagnostic::error(DiagnosticCode::Failed, "registry generation projection failed"));
        }
        detail::EventLogValidator validator;
        for (const auto& event : candidate.events) {
            auto accepted = validator.accept(event.sequence, event.generation,
                                             event.operation == RegistryOperation::Remove, event.tombstone);
            if (!accepted) return accepted;
        }
        auto boundary = validator.finish(candidate.nextEventSequence);
        if (!boundary) return boundary;
 
        state_.swap(candidate);
        return Result<void>::success();
    }
 
    [[nodiscard("retain Subscription or explicitly dispose it")]] Subscription subscribe(ChangeCallback callback) {
        return observers_.subscribe(std::move(callback));
    }
 
private:
    void generationProject(Value& value, Generation generation) const {
        if (generationProjector_) generationProjector_(value, generation);
    }
 
    [[nodiscard]] Result<Handle> mutate(RegistryOperation operation, Key key, std::optional<Value> value,
                                        EventData data, std::string label) {
        try {
            // Every potentially-throwing operation happens on a detached
            // candidate. The live state is touched only by the nothrow swap
            // below, so Value/EventData/projector failures cannot partially
            // publish a registry mutation.
            State      candidate = state_;
            auto       it        = candidate.entries.find(key);
            const bool live      = it != candidate.entries.end() && it->second.value.has_value();
            if (operation == RegistryOperation::Insert && live)
                return Result<Handle>::failure(
                    Diagnostic::error(DiagnosticCode::AlreadyExists, "registry key already exists"));
            if (operation == RegistryOperation::Replace && !live)
                return Result<Handle>::failure(Diagnostic::error(DiagnosticCode::NotFound, "registry key is not live"));
            if (operation == RegistryOperation::Remove && !live)
                return Result<Handle>::failure(Diagnostic::error(DiagnosticCode::NotFound, "registry key is not live"));
            if (operation != RegistryOperation::Remove && !value.has_value())
                return Result<Handle>::failure(
                    Diagnostic::error(DiagnosticCode::InvariantViolation, "live registry mutation has no value"));
 
            const auto current = it == candidate.entries.end() ? nullptr : &it->second;
            auto       next =
                detail::nextRegistryGeneration(current ? current->generation : Generation{}, current != nullptr);
            if (!next.ok()) return Result<Handle>::failure(next.status());
            const auto nextSequence = detail::nextRegistryEventSequence(candidate.nextEventSequence);
            if (!nextSequence.ok()) return Result<Handle>::failure(nextSequence.status());
 
            if (operation != RegistryOperation::Remove) generationProject(*value, next.value());
 
            candidate.events.reserve(candidate.events.size() + 1);
            Event event{candidate.nextEventSequence,
                        operation,
                        key,
                        next.value(),
                        operation == RegistryOperation::Remove,
                        std::move(label),
                        std::move(data)};
 
            if (operation == RegistryOperation::Insert && it == candidate.entries.end()) {
                auto [inserted, wasInserted] =
                    candidate.entries.try_emplace(std::move(key), Entry{next.value(), std::move(value)});
                (void)wasInserted;
                it = inserted;
            } else if (operation == RegistryOperation::Remove) {
                it->second.generation = next.value();
                it->second.value.reset();
            } else {
                it->second.generation = next.value();
                it->second.value      = std::move(value);
            }
 
            candidate.events.push_back(std::move(event));
            candidate.nextEventSequence = nextSequence.value();
            // Keep the notification independent of state_.events. Observer
            // dispatch is reentrant; a nested mutation is allowed to swap or
            // reallocate the canonical event vector while callbacks run.
            const Event  notification = candidate.events.back();
            const Handle handle{notification.key, notification.generation};
            Status       appliedStatus = Status::success(StatusCode::Applied);
            Status       callbackWarningStatus(
                StatusCode::Applied, {Diagnostic::warning(DiagnosticCode::CallbackFailure,
                                                                "registry mutation committed; observer callback failed")});
            state_.swap(candidate);
 
            // State is already committed. A user callback is outside the
            // transaction boundary: never roll back an event that another
            // callback may already have observed. Surface the exception as a
            // successful Applied result carrying a warning instead.
            try {
                observers_.notify(notification);
            } catch (const std::exception&) {
                return Result<Handle>::success(std::move(handle), std::move(callbackWarningStatus));
            } catch (...) {
                return Result<Handle>::success(std::move(handle), std::move(callbackWarningStatus));
            }
            return Result<Handle>::success(std::move(handle), std::move(appliedStatus));
        } catch (const std::exception&) {
            return Result<Handle>::failure(
                Diagnostic::error(DiagnosticCode::Failed, "registry mutation preparation failed"));
        } catch (...) {
            return Result<Handle>::failure(
                Diagnostic::error(DiagnosticCode::Failed, "registry mutation preparation failed"));
        }
    }
 
    State               state_;
    GenerationProjector generationProjector_;
    Observer<Event>     observers_;
};
 
}  // namespace eve