journal.hpp

// SPDX-License-Identifier: Apache-2.0
 
#pragma once
#include <algorithm>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <memory>
#include <stdexcept>
#include <string>
#include <string_view>
#include <unordered_map>
#include <utility>
#include <vector>
 
#include "../attributes.hpp"
#include "../core/detail/owned_state.hpp"
#include "../core/executor.hpp"
#include "../core/model.hpp"
#include "../core/registry.hpp"
#include "action_log.hpp"
 
namespace morph::journal {
 
/// @brief Thrown by `replay()` when a recorded entry's payload fingerprint
///        (`LogEntry::schema`) disagrees with the fingerprint this build
///        computes for the same action, and no migration covers the pair.
///
/// Without this signal the same situation reconstructs, successfully, a state
/// that was never recorded — a renamed field decoding to its default, reported
/// with as much confidence as a correct one. There is no "degraded"
/// reconstruction to fall
/// back to and no way to tell how much of the entry survived the decode, so
/// this throws rather than warning: an audit trail that cannot be reconstructed
/// faithfully must say so.
///
/// @par Who handles it
/// The caller of `replay()` / `SessionLog::undoLast()` — application-level
/// reconstruction code, not the framework. The three answers available to it
/// are: register a migration for the pair (see `PayloadMigrationRegistry`),
/// restore a build whose fingerprint matches, or surface the failure. The
/// framework deliberately offers no fourth.
struct SchemaMismatchError : std::runtime_error {
    /// @brief Builds the message naming the action and both fingerprints.
    /// @param modelType    `LogEntry::modelType` of the offending entry.
    /// @param actionType   `LogEntry::actionType` of the offending entry.
    /// @param recorded     Fingerprint stamped on the entry when it was written.
    /// @param current      Fingerprint this build computes for that action.
    SchemaMismatchError(std::string_view modelType, std::string_view actionType, std::string_view recorded,
                        std::string_view current)
        : std::runtime_error{"journal::replay: payload schema mismatch for " + std::string{modelType} + "/" +
                             std::string{actionType} + ": entry was written with schema " + std::string{recorded} +
                             ", this build decodes it as " + std::string{current} +
                             " (register a migration for this pair, or replay with a build that matches)"} {}
};
 
/// @brief What `replay()` does with an entry carrying no payload fingerprint.
///
/// An entry has an empty `LogEntry::schema` when it was written before the
/// field existed, or when application code appended it directly rather than
/// through an action execution. Such an entry is not "known good" — it is
/// *unverifiable*: there is no record of the shape that produced its payload,
/// so no check can be performed at all.
enum class UnstampedPayloadPolicy : std::uint8_t {
    /// @brief Replay it, without verifying a fingerprint it does not carry.
    ///
    /// The default, because it is the only choice that keeps journals written
    /// by earlier builds replayable at all — refusing them by default would
    /// turn an upgrade into a data-loss event for every retained journal, which
    /// is precisely the failure this feature exists to prevent, inverted.
    /// Choosing this is choosing to accept the *pre-existing* silent-default
    /// risk for *pre-existing* entries; it grants no leniency to anything
    /// written after the stamp exists.
    Replay,
 
    /// @brief Refuse: throw `SchemaMismatchError` on the first unstamped entry.
    ///
    /// For a caller whose correctness claim is "this reconstruction is
    /// faithful" and who would rather have no answer than an unverifiable one —
    /// a regulated audit trail, a conformance run, a test pinning that every
    /// entry in a fixture really is stamped.
    Refuse,
};
 
/// @brief Registry of payload migrations, keyed by `(actionType, fromSchema)`.
///
/// The escape hatch that makes "refuse on mismatch" a livable contract rather
/// than a one-way door. A migration is a pure function over the recorded
/// payload JSON: given the bytes an older build wrote, return the bytes this
/// build's `ActionTraits<A>::fromJson` should see. Reconstruction applies it in
/// memory only — the journal on disk is never rewritten, so the recorded
/// history stays exactly what was recorded and the migration stays reviewable
/// as code.
///
/// @par Thread safety
/// None, matching `morph::model::detail::ActionDispatcher`: register every
/// migration during start-up, before any `replay()` can run.
///
/// @code
/// morph::journal::PayloadMigrationRegistry migrations;
/// migrations.add("setState", oldFingerprint, [](std::string_view payload) {
///     // rename "state" -> "stateCode"
///     return rewriteKey(payload, "state", "stateCode");
/// });
/// auto holder = morph::journal::replay("Sample", log.entries(), registry, dispatcher, migrations);
/// @endcode
class PayloadMigrationRegistry {
public:
    /// @brief Rewrites one recorded payload JSON into the current shape.
    using Migration = std::function<std::string(std::string_view)>;
 
    /// @brief Registers @p migration for entries of @p actionType stamped with
    ///        @p fromSchema. Replaces any migration already registered for that
    ///        exact pair.
    /// @param actionType `LogEntry::actionType` the migration applies to.
    /// @param fromSchema `LogEntry::schema` value the migration reads.
    /// @param migration  Payload rewriter.
    void add(std::string_view actionType, std::string_view fromSchema, Migration migration) {
        _migrations[Key{std::string{actionType}, std::string{fromSchema}}] = std::move(migration);
    }
 
    /// @brief Looks up the migration for `(actionType, fromSchema)`.
    ///
    /// Returns a pointer into the registry, valid until the entry is replaced
    /// or the registry is destroyed (`std::unordered_map` does not invalidate
    /// references to surviving elements on rehash).
    /// @param actionType `LogEntry::actionType` to match.
    /// @param fromSchema `LogEntry::schema` to match.
    /// @return Pointer to the migration, or `nullptr` if none is registered —
    ///         into this registry, so valid only for as long as it is.
    [[nodiscard]] const Migration* find(std::string_view actionType,
                                        std::string_view fromSchema) const MORPH_LIFETIMEBOUND {
        auto iter = _migrations.find(::morph::model::detail::PairKeyView{actionType, fromSchema});
        return iter == _migrations.end() ? nullptr : &iter->second;
    }
 
    /// @brief Removes every registered migration.
    void clear() { _migrations.clear(); }
 
    /// @brief Returns how many migrations are registered.
    /// @return The registration count.
    [[nodiscard]] std::size_t size() const noexcept { return _migrations.size(); }
 
private:
    using Key = std::pair<std::string, std::string>;
    // Both functors, not just the hash: `unordered_map` enables heterogeneous
    // lookup only when the hash *and* the equality are transparent. Name the
    // hash alone and every `find` silently builds a `Key` to probe with, while
    // looking like it does not. `add` builds one regardless, because it
    // inserts.
    std::unordered_map<Key, Migration, ::morph::model::detail::PairKeyHash, ::morph::model::detail::PairKeyEqual>
        _migrations;
};
 
/// @brief Returns the process-level migration registry `replay()` uses by
///        default. Empty until an application adds to it.
/// @return Reference to the process-wide registry.
[[nodiscard]] inline PayloadMigrationRegistry& defaultPayloadMigrations() {
    static PayloadMigrationRegistry registry;
    return registry;
}
 
namespace detail {
 
/// @brief Thread-local flag telling executing model code whether the current
///        dispatch is happening inside `replay()`.
///
/// Installed by `replay()` around its dispatch loop via `ScopedReplayFlag`,
/// mirroring how `morph::session::detail::tlsCurrent()`/`ScopedContext` thread
/// a per-call `Context` through dispatch -- same shape (a thread-local slot
/// plus an RAII guard that restores the previous value on scope exit), applied
/// to a `bool` instead of a `const Context*`. Model/rule code never touches
/// this directly; it reads the public accessor `isReplaying()`.
inline bool& tlsIsReplaying() {
    thread_local bool tls = false;
    return tls;
}
 
/// @brief RAII helper that sets the thread-local replay flag for its scope,
///        restoring the previous value on destruction.
///
/// Nests correctly: a `replay()` call that itself triggers another `replay()`
/// (not a pattern this framework uses today, but not precluded) leaves the
/// flag `true` for the whole nested extent and restores the outer call's value
/// on the inner guard's destruction -- the same nesting behavior
/// `session::detail::ScopedContext` already has for `Context`.
class ScopedReplayFlag {
public:
    /// @brief Sets the thread-local replay flag to `true`, saving whatever
    ///        value was there before.
    ScopedReplayFlag() : _previous{tlsIsReplaying()} { tlsIsReplaying() = true; }
    /// @brief Restores the saved value.
    ~ScopedReplayFlag() { tlsIsReplaying() = _previous; }
 
    ScopedReplayFlag(const ScopedReplayFlag&) = delete;
    ScopedReplayFlag& operator=(const ScopedReplayFlag&) = delete;
    ScopedReplayFlag(ScopedReplayFlag&&) = delete;
    ScopedReplayFlag& operator=(ScopedReplayFlag&&) = delete;
 
private:
    bool _previous;
};
 
}  // namespace detail
 
/// @brief Returns `true` if the calling thread is currently inside a
///        `replay()` dispatch, `false` otherwise.
///
/// This is the signal Phase 6's automation-rules engine (and any other model
/// code that reacts to its own actions) checks before evaluating a rule: a
/// rule that fires again while `replay()` re-applies its recorded trigger
/// entry would double-apply a cascade that is also being replayed from its own
/// recorded entry (see `docs/spec/journal/journal.md`'s cascade-journaling
/// section). Reading this outside of any `replay()` call (the ordinary,
/// live-dispatch case) always returns `false`.
/// @return `true` during `replay()`'s dispatch loop on this thread, `false` otherwise.
[[nodiscard]] inline bool isReplaying() noexcept { return detail::tlsIsReplaying(); }
 
/// @brief Reconstructs model state by replaying @p entries, in order, against a
///        freshly created model instance.
///
/// Builds on the same `ModelRegistryFactory` / `ActionDispatcher` machinery
/// `RemoteServer` already uses to dispatch incoming requests — no separate
/// replay engine is needed. Because a model's entire state is exactly "initial
/// state plus the ordered actions replayed against it", this both reconstructs
/// state from a durable log and powers `SessionLog::undoLast()` below.
///
/// Sets `isReplaying()` to `true` for the duration of the dispatch loop below
/// (via `detail::ScopedReplayFlag`), so any model/rule code executed as part of
/// a replayed dispatch can tell it is being replayed rather than live-dispatched
/// -- this is what lets Phase 6's rules engine suppress rule evaluation on
/// replay while `replay()` re-applies the cascade's own recorded entry
/// unchanged. The flag is restored to its prior value (`false`, for any
/// ordinary top-level caller) once this function returns, so it never leaks
/// into dispatches that happen after `replay()` completes.
///
/// @par Payload-schema verification
/// Before dispatching an entry, `replay()` compares the entry's
/// `LogEntry::schema` against `ActionDispatcher::schemaFor(modelType,
/// actionType)` — the fingerprint *this* build computes for that action:
///
/// - **Equal** — dispatch, unchanged.
/// - **Different, with a migration registered** for `(actionType, entry.schema)`
///   in @p migrations — the migration rewrites the payload JSON in memory, and
///   the rewritten bytes are dispatched. The stored entry is untouched.
/// - **Different, with no migration** — throw `SchemaMismatchError`. Decoding
///   it anyway is what would reconstruct a state nobody ever recorded.
/// - **Entry unstamped** (`schema` empty) — governed by @p unstamped; see
///   `UnstampedPayloadPolicy`.
///
/// An entry naming an action this build has not registered is *not* rejected
/// here (`schemaFor` returns empty for it); it falls through to `dispatch()`,
/// which throws "unknown action" — the more precise diagnostic of the two.
///
/// @param modelTypeId String type-id of the model to reconstruct (`ModelTraits<M>::typeId()`).
/// @param entries     Ordered entries to replay, typically from `IActionLog::entries()`. Entries
///                    with `outcome == Outcome::Failed` are skipped (see below).
/// @param registry    Model factory registry; defaults to the process-level singleton.
/// @param dispatcher  Action dispatcher; defaults to the process-level singleton.
/// @param migrations  Payload migrations consulted on a fingerprint mismatch;
///                    defaults to the process-level registry, which is empty
///                    unless the application populated it.
/// @param unstamped   What to do with an entry carrying no fingerprint;
///                    defaults to `UnstampedPayloadPolicy::Replay`.
/// @return A freshly created holder with @p entries replayed against it.
/// @throws std::runtime_error if @p modelTypeId or any entry's action type is unregistered.
/// @throws SchemaMismatchError if an entry's payload fingerprint disagrees with this
///         build's and no migration covers it, or if an unstamped entry is met
///         under `UnstampedPayloadPolicy::Refuse`.
inline std::unique_ptr<::morph::model::detail::IModelHolder> replay(
    std::string_view modelTypeId, const std::vector<LogEntry>& entries,
    ::morph::model::detail::ModelRegistryFactory& registry = ::morph::model::detail::defaultRegistry(),
    ::morph::model::detail::ActionDispatcher& dispatcher = ::morph::model::detail::defaultDispatcher(),
    const PayloadMigrationRegistry& migrations = defaultPayloadMigrations(),
    UnstampedPayloadPolicy unstamped = UnstampedPayloadPolicy::Replay) {
    auto holder = registry.create(modelTypeId);
    // `registry.create` (via `ModelFactory::create`) auto-attaches the process
    // default action log. Detach it before replaying: reconstruction re-runs the
    // recorded actions, and without this each replayed dispatch would re-record
    // into the live sink, corrupting the very audit trail we are reading from.
    holder->attachActionLog(nullptr, {});
    const detail::ScopedReplayFlag replayFlag;
    for (const auto& entry : entries) {
        // A Failed entry (see action_log.hpp's `Outcome`) never mutated model
        // state -- Model::execute threw or the validator rejected it before any
        // state change -- so there is nothing to reconstruct from it. Worse,
        // re-dispatching it would very likely throw the same exception again
        // (the same rejected precondition), aborting reconstruction outright.
        // Skipping it is exactly "replay only committed facts", which is what
        // this function promises.
        if (entry.outcome == Outcome::Failed) {
            continue;
        }
        // Payload-schema gate. `current` is empty for an action this build never
        // registered; leaving that to dispatch()'s "unknown action" keeps the
        // two failures distinguishable instead of reporting the missing
        // registration as a schema change.
        const std::string current = dispatcher.schemaFor(entry.modelType, entry.actionType);
        std::string_view payload{entry.payload};
        std::string migrated;
        if (entry.schema.empty()) {
            if (unstamped == UnstampedPayloadPolicy::Refuse) {
                throw SchemaMismatchError{entry.modelType, entry.actionType, "(unstamped)", current};
            }
        } else if (!current.empty() && entry.schema != current) {
            const auto* migration = migrations.find(entry.actionType, entry.schema);
            if (migration == nullptr) {
                throw SchemaMismatchError{entry.modelType, entry.actionType, entry.schema, current};
            }
            migrated = (*migration)(entry.payload);
            payload = migrated;
        }
        dispatcher.dispatch(entry.modelType, entry.actionType, *holder, payload);
    }
    return holder;
}
 
/// @brief Full-fidelity, in-memory log of one model instance's executed actions.
///
/// Attach directly via `IModelHolder::attachActionLog` — every successfully
/// executed loggable action is appended here in order, regardless of any
/// `ActionLogPolicy::coalesce` setting. This full history is the raw material
/// for `undoLast()`. `checkpoint()` is the one place coalescing actually
/// happens: entries accumulated since the last checkpoint are reduced by
/// `(modelType, entityKey, actionType)` — keeping only the latest occurrence
/// where the action's policy says `coalesce == true`, keeping every occurrence
/// otherwise — and only that reduced set reaches the durable sink.
///
/// @par One owner
/// The history and the checkpoint watermark belong to the executor given at
/// construction, which must run one task at a time. `append()` runs there —
/// at once when called on it, posted from a model's strand. `entries()`,
/// `undoLast()` and `checkpoint()` run only there; a caller elsewhere reads
/// through `entries(replyExec, entityKey)`. Because a checkpoint and every
/// append are tasks of the one owner, a checkpoint's slice-and-forward is
/// never interleaved with an append or with another checkpoint. See
/// `docs/spec/journal/journal.md`, "One owner".
class SessionLog : public IActionLog {
public:
    using IActionLog::entries;
    using IActionLog::flush;
 
    /// @brief Constructs an empty session log belonging to @p owner.
    /// @param owner The executor every access runs on; must run one task at a
    ///        time. Borrowed: it must outlive this log and run what it posts.
    explicit SessionLog(::morph::exec::IExecutor& owner MORPH_LIFETIMEBOUND)
        : _owned{owner, std::make_shared<State>()} {}
 
    /// @brief Appends @p entry, assigning a monotonically increasing `seq`, on
    ///        the owner.
    ///
    /// Always full fidelity: this is what `undoLast()` walks back through, so
    /// nothing is coalesced or dropped here.
    /// @param entry Entry to append; `seq` is overwritten regardless of the input value.
    void append(LogEntry entry) override {
        _owned.apply("SessionLog::append", [entry = std::move(entry)](State& state) mutable {
            entry.seq = ++state.nextSeq;
            state.all.push_back(std::move(entry));
        });
    }
 
    /// @brief No-op — `checkpoint()` is this class's real commit point. Callable anywhere.
    void flush() override {}
 
    /// @brief Returns the full history (or one entity's slice of it) in append
    ///        order. On the owner.
    /// @param entityKey If non-empty, restricts the result to that entity's entries.
    /// @return Matching entries, in append order.
    [[nodiscard]] std::vector<LogEntry> entries(std::string_view entityKey = {}) const override {
        return detail::entriesFor(_owned.read("SessionLog::entries").all, entityKey);
    }
 
    /// @brief The executor given at construction.
    /// @return The owner.
    [[nodiscard]] ::morph::exec::IExecutor* owner() const noexcept override { return &_owned.owner(); }
 
    /// @brief Drops the most recently appended entry and replays everything that
    ///        remains against a fresh model instance, reconstructing the state
    ///        the model was in immediately before that entry executed. On the owner.
    ///
    /// No inverse/undo operations are needed on the action types themselves —
    /// this reuses `replay()` over a shorter prefix of the same log. A no-op
    /// (returns a freshly created, un-replayed holder) if the log is empty.
    ///
    /// Undo only rewinds this in-memory history; it never moves the checkpoint
    /// watermark, which is keyed by entry `seq` (see `checkpoint()`). Undoing an
    /// entry a prior `checkpoint()` already forwarded does **not** un-forward it:
    /// the durable sink is append-only, so that entry stays durable while the
    /// reconstructed history diverges from it. A later `checkpoint()` never
    /// re-forwards a coalesced-away, already-committed entry. Applications that
    /// need to durably reverse a checkpointed action must record a compensating
    /// action, not rely on `undoLast()`.
    ///
    /// Replays through `replay()` with its default payload-schema settings.
    /// That gate cannot fire here in practice: the entries being replayed were
    /// appended by this same process, so their fingerprints are this build's by
    /// construction. Undo is not a cross-version path.
    ///
    /// Synchronous: it returns the reconstructed holder, and its caller — the
    /// application — is the owner.
    ///
    /// @param modelTypeId String type-id of the model to reconstruct.
    /// @param registry    Model factory registry; defaults to the process-level singleton.
    /// @param dispatcher  Action dispatcher; defaults to the process-level singleton.
    /// @return A freshly created holder with the pre-undo state replayed onto it.
    std::unique_ptr<::morph::model::detail::IModelHolder> undoLast(
        std::string_view modelTypeId,
        ::morph::model::detail::ModelRegistryFactory& registry = ::morph::model::detail::defaultRegistry(),
        ::morph::model::detail::ActionDispatcher& dispatcher = ::morph::model::detail::defaultDispatcher()) {
        State& state = _owned.read("SessionLog::undoLast");
        if (!state.all.empty()) {
            // The checkpoint watermark is a *seq* threshold, not an index into
            // `all`, so dropping the tail entry cannot lower it: undo never
            // rewinds the watermark, and a later checkpoint never re-forwards
            // a coalesced-away, already-committed entry.
            state.all.pop_back();
        }
        return replay(modelTypeId, state.all, registry, dispatcher);
    }
 
    /// @brief Coalesces entries appended since the last checkpoint and forwards
    ///        the reduced set, in order, to @p durableSink; then flushes it. On
    ///        the owner.
    ///
    /// Advances the checkpoint watermark (the `seq` of the last committed entry)
    /// *before* forwarding, so the batch is consumed even if @p durableSink's
    /// `append()` or `flush()` throws: a checkpoint is a forward-only commit
    /// point (at-most-once), not a transaction to retry. A no-op if nothing has
    /// been appended since the last checkpoint.
    ///
    /// @par Ordering
    /// The whole body is one task of the owner, where every `append()` runs
    /// too, so no append lands between taking the slice and forwarding it, and
    /// two checkpoints forward their batches in the order they ran. That is
    /// the append-order identity the sink relies on.
    ///
    /// @param durableSink Receives only the coalesced entries — never the raw
    ///        stream. Driven from this log's owner, so it must share it: its
    ///        `flush()` answers only on its own owner.
    /// @param dispatcher  Supplies each action's `coalesce` policy; defaults to
    ///                    the process-level singleton (the same one actions were
    ///                    registered against via `BRIDGE_REGISTER_ACTION`).
    void checkpoint(IActionLog& durableSink, ::morph::model::detail::ActionDispatcher& dispatcher =
                                                 ::morph::model::detail::defaultDispatcher()) {
        State& state = _owned.read("SessionLog::checkpoint");
        // Forward every entry whose seq is strictly past the last committed
        // seq. `seq` is assigned once at append and never reused, so this is
        // stable under coalescing and under `undoLast()` removing tail
        // entries — unlike a raw index into `all`, which those mutate.
        std::vector<LogEntry> pending;
        uint64_t highestSeq = state.committedUpToSeq;
        for (const auto& entry : state.all) {
            if (entry.seq > state.committedUpToSeq) {
                pending.push_back(entry);
                highestSeq = std::max(highestSeq, entry.seq);
            }
        }
        if (pending.empty()) {
            return;
        }
        state.committedUpToSeq = highestSeq;
        for (auto& entry : coalesced(pending, dispatcher)) {
            durableSink.append(std::move(entry));
        }
        durableSink.flush();
    }
 
protected:
    /// @brief Reads the history on the owner, delivered on @p replyExec.
    /// @param replyExec Where the answer is delivered.
    /// @param entityKey If non-empty, restricts the result to that entity's entries.
    /// @return The matching entries.
    [[nodiscard]] ::morph::async::Completion<std::vector<LogEntry>> askEntries(::morph::exec::IExecutor& replyExec,
                                                                               std::string entityKey) const override {
        return _owned.ask<std::vector<LogEntry>>(
            "SessionLog::entries", replyExec,
            [entityKey = std::move(entityKey)](State& state) { return detail::entriesFor(state.all, entityKey); });
    }
 
private:
    /// @brief Reduces @p pending by `(modelType, entityKey, actionType)`: the
    ///        latest occurrence overwrites earlier ones (in place, preserving
    ///        first-seen position) wherever @p dispatcher says the action
    ///        coalesces; every other entry is kept as-is.
    static std::vector<LogEntry> coalesced(const std::vector<LogEntry>& pending,
                                           ::morph::model::detail::ActionDispatcher& dispatcher) {
        std::vector<LogEntry> out;
        std::unordered_map<std::string, std::size_t> lastIndexForKey;
        for (const auto& entry : pending) {
            if (!dispatcher.coalesce(entry.modelType, entry.actionType)) {
                out.push_back(entry);
                continue;
            }
            std::string key = entry.modelType + '\0' + entry.entityKey + '\0' + entry.actionType;
            auto iter = lastIndexForKey.find(key);
            if (iter == lastIndexForKey.end()) {
                lastIndexForKey.emplace(std::move(key), out.size());
                out.push_back(entry);
            } else {
                // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-avoid-unchecked-container-access) — index from same loop
                out[iter->second] = entry;
            }
        }
        return out;
    }
 
    /// Everything the log holds; touched only on the owner.
    struct State {
        std::vector<LogEntry> all;
        uint64_t committedUpToSeq{0};
        uint64_t nextSeq{0};
    };
 
    ::morph::exec::detail::OwnedState<State> _owned;
};
 
}  // namespace morph::journal