replay_ledger.hpp

// SPDX-License-Identifier: Apache-2.0
 
#pragma once
#include <exception>
#include <map>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <utility>
 
#include "../attributes.hpp"
#include "../core/completion.hpp"
#include "../core/detail/owned_state.hpp"
#include "../core/executor.hpp"
 
namespace morph::offline {
 
/// @brief Interface for an op-id-keyed exactly-once replay ledger.
///
/// Answers one question — "has this operation id already been applied?" —
/// for a host that must dedup a retried write against one it already
/// committed. One framework seam in place of a near-identical table each host
/// would otherwise hand-write. A host either stores a result and replays it
/// verbatim on a hit ("response-replay"), or stores nothing and reports only
/// that the op was seen ("skip-only", as `examples/bookmarks`' import path
/// does). Both are the same mechanism with a different
/// caller-side interpretation of the stored payload — see `lookup()`'s own
/// doc comment — so there is exactly one interface here, not two.
///
/// @par Why the framework cannot supply the storage
/// Every existing occurrence stores its ledger row in the *same database and
/// the same transaction* as the write it guards, so the check-then-set commits
/// atomically with the operation's effect — see `record()`'s doc comment.
/// `include/morph` has no dependency on any SQL library, so it cannot open
/// that connection or join that transaction. What it can supply is the
/// *contract*: the shape every implementation must have, and a conformance
/// suite (`tests/replay_ledger_conformance.hpp`) any implementation can be
/// run against. The table itself stays app-side, per rung, exactly as before.
///
/// @par No base class for consumers
/// A model holds (or is handed) an `IReplayLedger&`; it never derives from
/// one. Every existing occurrence is a free function or a plain member on the
/// model, not a base class — a design that required inheriting from this
/// interface would be un-adoptable by every one of them.
// NOLINTBEGIN(cppcoreguidelines-special-member-functions)
struct IReplayLedger {
    virtual ~IReplayLedger() = default;
 
    /// @brief Reports whether @p opId within @p scope has already been decided.
    ///
    /// @par What a hit means is the caller's decision, not this interface's
    /// The returned string is whatever `record()` stored alongside the id. A
    /// response-replay consumer (a move or a money transfer whose retry must
    /// see the original answer) stores its result's serialised form and, on a
    /// hit, returns that string verbatim instead of re-running the operation. A
    /// skip-only consumer (bookmarks' import path, or an at-most-once replay
    /// of queued captures) stores an empty payload and, on a hit, reports "already applied,
    /// nothing to return" without ever looking at the string. Both read
    /// exactly the same `lookup()` — the value type reports the fact, and the
    /// layer that calls this decides what a hit means.
    ///
    /// @par `scope`
    /// An opaque caller-chosen partition — the three key shapes the seven
    /// occurrences used (`(entity_id, op_id)`, `(owner_principal, op_id)`, a
    /// bare `op_id`) are all this one string plus the id, with the bare-key
    /// case simply passing an empty `scope`. Two calls with the same `opId`
    /// but different `scope` never collide.
    ///
    /// @par Empty `opId`
    /// Never reported as decided, in every conforming implementation — see
    /// `record()`. An empty id is not an identity, and reporting one "already
    /// applied" would make every unkeyed call after the first a silent no-op.
    ///
    /// @param scope Caller-chosen partition; empty is a valid, distinct scope.
    /// @param opId  The operation id to look up. Empty always reports
    ///              `std::nullopt`, regardless of what has been recorded.
    /// @return The payload recorded with @p opId (possibly empty, for a
    ///         skip-only caller), or `std::nullopt` if it has not been
    ///         recorded — or @p opId is empty.
    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters) -- scope/opId, not interchangeable
    [[nodiscard]] std::optional<std::string> lookup(std::string_view scope, std::string_view opId) const {
        if (opId.empty()) {
            return std::nullopt;
        }
        return doLookup(scope, opId);
    }
 
    /// @brief Records that @p opId within @p scope has been decided, with
    ///        @p payload attached.
    ///
    /// @par Atomicity is the caller's responsibility, not this interface's
    /// The framework cannot enforce this — `IReplayLedger` has no database
    /// connection of its own — so it is stated here instead: a conforming
    /// implementation MUST let its caller commit this call in the same
    /// transaction as the write it guards. An implementation backed by a SQL
    /// table does this by operating through the caller's own already-open
    /// connection/transaction handle (passed to the implementation at
    /// construction), never by opening a connection of its own. Recording
    /// outside that transaction reintroduces exactly the defect this ledger
    /// exists to prevent: a crash between the write and the record redelivers
    /// the operation and it is re-applied.
    ///
    /// @par Idempotent: first-write-wins
    /// Recording an @p opId within @p scope that is already decided is a
    /// no-op — the existing payload is kept, `payload` here is discarded, and
    /// no error is raised. This is the same contract
    /// `morph::journal::IActionLog::append()`'s idempotency-key dedup already
    /// ships (`journal/action_log.hpp`'s own doc comment: "if an entry with
    /// the same key was already recorded, treat the call as a no-op"): a
    /// redelivered retry that reaches `record()` a second time (e.g. because
    /// a reply was lost after the first commit) must not overwrite a payload
    /// a caller may already have replayed to someone else.
    ///
    /// @par Empty `opId`
    /// A no-op — there is no identity to record against, so `lookup()` for an
    /// empty id must keep reporting `std::nullopt` regardless of how many
    /// times `record()` is called with one.
    ///
    /// @param scope   Caller-chosen partition; must match the `scope` a later
    ///                `lookup()` uses to find this entry.
    /// @param opId    The operation id being recorded. A no-op if empty.
    /// @param payload Opaque payload to associate with @p opId; empty for a
    ///                skip-only caller that only cares whether a hit occurred.
    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters) -- scope/opId, not interchangeable
    void record(std::string_view scope, std::string_view opId, std::string payload) {
        if (opId.empty()) {
            return;
        }
        doRecord(scope, opId, std::move(payload));
    }
 
    /// @brief Reports whether @p opId within @p scope has already been
    ///        decided, looked up on the ledger's owner and delivered on
    ///        @p replyExec.
    ///
    /// The form of `lookup()` for a caller that is not on the ledger's owner.
    /// A lookup answered elsewhere cannot be atomic with the `record()` that
    /// follows it; a caller that needs check-then-set atomicity runs both on
    /// the owner.
    /// @param replyExec Where the answer is delivered, and where the caller
    ///        attaches to it. Borrowed: it must outlive the completion.
    /// @param scope Caller-chosen partition; empty is a valid, distinct scope.
    /// @param opId  The operation id to look up. Empty always answers
    ///              `std::nullopt`.
    /// @return What `lookup(scope, opId)` would have returned.
    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters) -- scope/opId, not interchangeable
    [[nodiscard]] ::morph::async::Completion<std::optional<std::string>> lookup(
        ::morph::exec::IExecutor& replyExec MORPH_LIFETIMEBOUND, std::string_view scope, std::string_view opId) const {
        if (opId.empty()) {
            auto settleable = ::morph::async::Completion<std::optional<std::string>>::makeSettleable(&replyExec);
            settleable.second.resolve(std::nullopt);
            return std::move(settleable.first);
        }
        return askLookup(replyExec, std::string{scope}, std::string{opId});
    }
 
protected:
    /// @brief What `lookup(replyExec, scope, opId)` asks. The default looks
    ///        up where it is called, for an implementation with no owner; one
    ///        with an owner looks up there.
    /// @param replyExec Where the answer is delivered.
    /// @param scope     Caller-chosen partition.
    /// @param opId      The operation id to look up; never empty here.
    /// @return The payload recorded with @p opId, or `std::nullopt`.
    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters) -- scope/opId, not interchangeable
    [[nodiscard]] virtual ::morph::async::Completion<std::optional<std::string>> askLookup(
        ::morph::exec::IExecutor& replyExec,
        // NOLINTNEXTLINE(performance-unnecessary-value-param): the base only reads them; overrides take them by value and move them.
        std::string scope, std::string opId) const {
        auto settleable = ::morph::async::Completion<std::optional<std::string>>::makeSettleable(&replyExec);
        try {
            settleable.second.resolve(doLookup(scope, opId));
        } catch (...) {
            settleable.second.reject(std::current_exception());
        }
        return std::move(settleable.first);
    }
 
    /// @brief Storage hook for `lookup()`.
    /// @param scope Caller-chosen partition, forwarded verbatim from `lookup()`.
    /// @param opId  The operation id to look up; never empty here -- the
    ///              public wrapper already handled that case.
    /// @return The payload recorded with @p opId, or `std::nullopt` if none.
    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters) -- scope/opId, not interchangeable
    [[nodiscard]] virtual std::optional<std::string> doLookup(std::string_view scope, std::string_view opId) const = 0;
 
    /// @brief Storage hook for `record()`. Must be a no-op if @p scope/@p opId
    ///        is already recorded (first-write-wins).
    /// @param scope   Caller-chosen partition, forwarded verbatim from `record()`.
    /// @param opId    The operation id being recorded; never empty here -- the
    ///                public wrapper already handled that case.
    /// @param payload Opaque payload to associate with @p opId.
    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters) -- scope/opId, not interchangeable
    virtual void doRecord(std::string_view scope, std::string_view opId, std::string payload) = 0;
};
// NOLINTEND(cppcoreguidelines-special-member-functions)
 
// ── In-memory implementation ──────────────────────────────────────────────────
 
/// @brief In-memory implementation of `IReplayLedger`, owned by one executor.
///
/// Suitable for testing and for a host with no durability requirement.
/// Mirrors `morph::offline::InMemoryOfflineQueue`'s shape. Not durable across
/// a process restart, and — since it is not backed by any real transaction —
/// not a stand-in for `record()`'s atomicity requirement: a real
/// implementation must be backed by the same store and transaction as the
/// write it guards.
///
/// @par One owner
/// The entries belong to the executor given at construction, which must run
/// one task at a time — the executor the guarded writes run on, so a
/// `lookup()` and the `record()` after it are one check-then-set. `lookup()`
/// answers only there; `record()` runs there, posted from anywhere else; a
/// caller elsewhere looks up through `lookup(replyExec, scope, opId)`.
class InMemoryReplayLedger : public IReplayLedger {
public:
    /// @brief Constructs an empty ledger belonging to @p owner.
    /// @param owner The executor every access runs on; must run one task at a
    ///        time. Borrowed: it must outlive this ledger and run what it posts.
    explicit InMemoryReplayLedger(::morph::exec::IExecutor& owner MORPH_LIFETIMEBOUND)
        : _owned{owner, std::make_shared<State>()} {}
 
protected:
    /// @brief Looks up @p opId within @p scope in the in-memory map. On the owner.
    ///
    /// @par The two `std::string` constructions below are deliberate
    /// They materialise the key purely to probe an ordered map that could
    /// take a transparent comparator instead. That was profiled rather than
    /// fixed. Measured with clang 22 `-O2`, a counting `operator new`, 2e6
    /// iterations per row:
    ///
    /// @verbatim
    /// -- allocations per lookup() --
    /// scope=bookmarks              ( 9) id=CreateSwimlane       (14) : 0.00
    /// scope=bookmarks              ( 9) id=CreateSwimlane-0001  (19) : 1.00
    /// scope=bookmarks-import-scope (22) id=CreateSwimlane-0001  (19) : 2.00
    ///
    /// -- ns per lookup(), single thread --
    /// both inside libstdc++'s 15-char SSO buffer : 28.2 ns
    /// both past it                               : 31.6 ns
    /// @endverbatim
    ///
    /// So the cost is real and id-length-dependent. What parks it is the call
    /// census, not the size: `InMemoryReplayLedger` is constructed in **one**
    /// file in this tree, `tests/test_replay_ledger.cpp`, and in no shipping
    /// code at all. The only production `IReplayLedger::lookup()` call
    /// (`examples/bookmarks/src/models/bookmark_model.cpp:737`, once per
    /// `ImportBookmarks`) runs against `BookmarksReplayLedger`, whose own
    /// `doLookup` is an ODBC round-trip -- next to which 3 ns is not
    /// measurable.
    ///
    /// It becomes worth fixing the moment a per-request caller of *this* class
    /// appears; the fix is then a transparent comparator on the map,
    /// `std::map`'s `is_transparent` flavour rather than `core/registry.hpp`'s
    /// hash-and-equality pair (which serves an `unordered_map` and does not
    /// apply here).
    ///
    /// @param scope Caller-chosen partition.
    /// @param opId  The operation id to look up; never empty here.
    /// @return The payload recorded with @p opId, or `std::nullopt` if none.
    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters) -- scope/opId, not interchangeable
    [[nodiscard]] std::optional<std::string> doLookup(std::string_view scope, std::string_view opId) const override {
        return find(_owned.read("InMemoryReplayLedger::lookup"), scope, opId);
    }
 
    /// @brief Records @p opId within @p scope in the in-memory map, unless
    ///        already present (first-write-wins). On the owner; posted there
    ///        from anywhere else.
    /// @param scope   Caller-chosen partition.
    /// @param opId    The operation id being recorded; never empty here.
    /// @param payload Opaque payload to associate with @p opId.
    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters) -- scope/opId, not interchangeable
    void doRecord(std::string_view scope, std::string_view opId, std::string payload) override {
        _owned.apply("InMemoryReplayLedger::record", [key = Key{std::string{scope}, std::string{opId}},
                                                      payload = std::move(payload)](State& state) mutable {
            // emplace (not insert_or_assign): first-write-wins, per record()'s
            // own doc comment -- a second record() for the same (scope, opId)
            // must not overwrite a payload a caller may already have replayed.
            state.entries.emplace(std::move(key), std::move(payload));
        });
    }
 
    /// @brief Looks up on the owner, answered on @p replyExec.
    /// @param replyExec Where the answer is delivered.
    /// @param scope     Caller-chosen partition.
    /// @param opId      The operation id to look up; never empty here.
    /// @return The payload recorded with @p opId, or `std::nullopt`.
    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters) -- scope/opId, not interchangeable
    [[nodiscard]] ::morph::async::Completion<std::optional<std::string>> askLookup(::morph::exec::IExecutor& replyExec,
                                                                                   std::string scope,
                                                                                   std::string opId) const override {
        return _owned.ask<std::optional<std::string>>(
            "InMemoryReplayLedger::lookup", replyExec,
            [scope = std::move(scope), opId = std::move(opId)](State& state) { return find(state, scope, opId); });
    }
 
private:
    using Key = std::pair<std::string, std::string>;  // (scope, opId)
 
    /// Everything the ledger holds; touched only on the owner.
    struct State {
        std::map<Key, std::string> entries;
    };
 
    /// The payload recorded under (@p scope, @p opId) in @p state, if any.
    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters) -- scope/opId, not interchangeable
    static std::optional<std::string> find(const State& state, std::string_view scope, std::string_view opId) {
        auto const iter = state.entries.find(Key{std::string{scope}, std::string{opId}});
        if (iter == state.entries.end()) {
            return std::nullopt;
        }
        return iter->second;
    }
 
    ::morph::exec::detail::OwnedState<State> _owned;
};
 
}  // namespace morph::offline