%% @doc Streams API facade for reckon-db %% %% Provides the public API for stream operations: %% - append: Write events to a stream with optimistic concurrency %% - read: Read events from a stream %% - get_version: Get current stream version %% - exists: Check if stream exists %% - list_streams: List all streams in the store %% %% @author rgfaber -module(reckon_db_streams). -include("reckon_db.hrl"). -include("reckon_db_telemetry.hrl"). -include_lib("khepri/include/khepri.hrl"). %% API -export([ append/4, append/5, append_if_no_tag_matches/4, read/5, read/6, read_all/4, read_all_global/3, read_by_event_types/3, read_by_tags/4, read_by_metadata/3, get_version/2, exists/2, has_events/1, list_streams/1, delete/2 ]). %% Internal exports for workers -export([ do_append/4, do_read/5 ]). %%==================================================================== %% Types %%==================================================================== -type new_event() :: #{ event_type := binary(), data := map() | binary(), metadata => map(), tags => [binary()], event_id => binary() }. -type direction() :: forward | backward. -export_type([new_event/0, direction/0]). %%==================================================================== %% API %%==================================================================== %% @doc Append events to a stream with expected version check %% %% Expected version semantics: %% -1 (NO_STREAM) - Stream must not exist (first write) %% -2 (ANY_VERSION) - No version check, always append %% N >= 0 - Stream version must equal N %% %% Returns {ok, NewVersion} on success or {error, Reason} on failure. -spec append(atom(), binary(), integer(), [new_event()]) -> {ok, non_neg_integer()} | {error, term()}. append(StoreId, StreamId, ExpectedVersion, Events) -> append(StoreId, StreamId, ExpectedVersion, Events, #{}). %% @doc Conditionally append events under the DCB pseudo-stream %% (Dynamic Consistency Boundary, Phase 3, 2.4.0+). %% %% Unlike append/4,5, the precondition is NOT a stream-version check; %% it is a tag-filter context query. Returns %% {error, {context_changed, MaxSeq}} when any event matching %% TagFilter has seq above SeqCutoff. %% %% v1 refuses on stores with integrity enabled (DCB v1 lacks HMAC %% chain). Returns {error, integrity_not_supported_in_dcb_v1} in %% that case. %% %% See: plans/PLAN_DCB_IMPLEMENTATION.md %% %% NOTE: This facade calls reckon_db_dcb directly. P3.4 will route via %% reckon_db_gateway_worker for transport-layer consistency with %% append/4,5. -spec append_if_no_tag_matches( StoreId :: atom(), TagFilter :: reckon_gater_types:tag_filter(), SeqCutoff :: reckon_gater_types:seq_cutoff(), Events :: [reckon_db_log_backend:new_event()] ) -> {ok, LastSeq :: non_neg_integer()} | {error, {context_changed, non_neg_integer()}} | {error, no_events} | {error, integrity_not_supported_in_dcb_v1} | {error, term()}. append_if_no_tag_matches(StoreId, TagFilter, SeqCutoff, Events) -> reckon_db_dcb:append_if_no_tag_matches(StoreId, TagFilter, SeqCutoff, Events). -spec append(atom(), binary(), integer(), [new_event()], map()) -> {ok, non_neg_integer()} | {error, term()}. append(StoreId, StreamId, ExpectedVersion, Events, _Opts) -> %% Stream-id format gate. Rejecting at the head of append/4 %% means no malformed id reaches Khepri, so the store can't %% accumulate polluted paths from misbehaving tests / clients. %% See reckon_gater_stream_id for the format rules (moved out %% of reckon-db in 3.0.0 — protocol contract belongs in the %% gateway layer, shared with reckon-evoq). case reckon_gater_stream_id:validate(StreamId) of ok -> do_append_with_telemetry(StoreId, StreamId, ExpectedVersion, Events); {error, Reason} -> {error, {invalid_stream_id, Reason, StreamId}} end. %% @private validation-gated wrapper around do_append that also %% emits the write-lifecycle telemetry. do_append_with_telemetry(StoreId, StreamId, ExpectedVersion, Events) -> StartTime = erlang:monotonic_time(), %% Emit start telemetry telemetry:execute( ?STREAM_WRITE_START, #{system_time => erlang:system_time(millisecond)}, #{store_id => StoreId, stream_id => StreamId, event_count => length(Events), expected_version => ExpectedVersion} ), Result = do_append(StoreId, StreamId, ExpectedVersion, Events), Duration = erlang:monotonic_time() - StartTime, %% Emit stop/error telemetry case Result of {ok, NewVersion} -> telemetry:execute( ?STREAM_WRITE_STOP, #{duration => Duration, event_count => length(Events)}, #{store_id => StoreId, stream_id => StreamId, new_version => NewVersion} ), Result; {error, Reason} -> telemetry:execute( ?STREAM_WRITE_ERROR, #{duration => Duration}, #{store_id => StoreId, stream_id => StreamId, reason => Reason} ), Result end. %% @doc Read events from a stream %% %% Parameters: %% StoreId - The store identifier %% StreamId - The stream identifier %% StartVersion - Starting version (0-based) %% Count - Maximum number of events to read %% Direction - forward or backward %% %% Returns {ok, [Event]} or {error, Reason} -spec read(atom(), binary(), non_neg_integer(), pos_integer(), direction()) -> {ok, [event()]} | {error, term()}. read(StoreId, StreamId, StartVersion, Count, Direction) -> read(StoreId, StreamId, StartVersion, Count, Direction, #{}). %% @doc Read events from a stream with explicit options. %% %% Currently supported options: %% %% `verify' :: `skip_legacy' | `strict' | `skip_all' %% Tamper-resistance enforcement mode. Default: `skip_legacy'. %% - `skip_legacy' (default): events with version below the %% per-stream chain_start watermark are returned untouched %% (legacy data); events at or above the watermark are %% verified strictly and an integrity_violation is returned %% on any failure. %% - `strict': every event must carry integrity fields and %% verify; legacy events surface as missing_integrity. %% - `skip_all': no verification (dangerous; intended for %% migration tooling only). %% %% Backward-direction reads always bypass chain verification in 2.1.0; %% the MAC alone could still be checked but is not in this release. %% Forward reads receive full chain + MAC verification. -type verify_mode() :: skip_legacy | strict | skip_all. -type read_opts() :: #{verify => verify_mode()}. -spec read( atom(), binary(), non_neg_integer(), pos_integer(), direction(), read_opts() ) -> {ok, [event()]} | {error, term()}. read(StoreId, StreamId, StartVersion, Count, Direction, Opts) -> StartTime = erlang:monotonic_time(), %% Emit start telemetry telemetry:execute( ?STREAM_READ_START, #{system_time => erlang:system_time(millisecond)}, #{store_id => StoreId, stream_id => StreamId, start_version => StartVersion, count => Count, direction => Direction} ), Result = do_read_with_verify( StoreId, StreamId, StartVersion, Count, Direction, Opts), Duration = erlang:monotonic_time() - StartTime, %% Emit stop telemetry case Result of {ok, Events} -> telemetry:execute( ?STREAM_READ_STOP, #{duration => Duration, event_count => length(Events)}, #{store_id => StoreId, stream_id => StreamId} ), Result; Error -> Error end. %% @doc Read all events from a stream -spec read_all(atom(), binary(), pos_integer(), direction()) -> {ok, [event()]} | {error, term()}. read_all(StoreId, StreamId, BatchSize, Direction) -> read(StoreId, StreamId, 0, BatchSize, Direction). %% @doc Read all events across all streams in global epoch_us order. %% %% Returns events sorted by epoch_us, skipping `Offset' events and %% returning up to `BatchSize' events. Used by catch-up subscriptions %% to replay historical events to a subscriber. %% %% Parameters: %% StoreId - The store identifier %% Offset - Number of events to skip (0-based) %% BatchSize - Maximum number of events to return %% %% Returns events sorted by epoch_us (global ordering). -spec read_all_global(atom(), non_neg_integer(), pos_integer()) -> {ok, [event()]} | {error, term()}. read_all_global(StoreId, Offset, BatchSize) -> case query_event_results(StoreId, has_data_leaf()) of {ok, Results} -> Events = [convert_result_to_event(PathKey, Value) || {PathKey, Value} <- Results], ValidEvents = [E || E <- Events, E =/= undefined], SortedEvents = sort_by_epoch(ValidEvents), Skipped = safe_nthtail(Offset, SortedEvents), {ok, lists:sublist(Skipped, BatchSize)}; {error, _} = Error -> Error end. %% @private Leaf condition matching any version/seq node that carries %% event data: `#if_all{[*, has_data]}'. Shared by the global readers. -spec has_data_leaf() -> khepri:condition(). has_data_leaf() -> #if_all{conditions = [ ?KHEPRI_WILDCARD_STAR, #if_has_data{has_data = true} ]}. %% @private Sort events into global epoch_us order. -spec sort_by_epoch([event()]) -> [event()]. sort_by_epoch(Events) -> lists:sort( fun(#event{epoch_us = E1}, #event{epoch_us = E2}) -> E1 =< E2 end, Events). %% @private Query every event node across the store. Model C stores %% regular events at 4 levels ([streams, Type, Id, Version]) and the DCB %% pseudo-stream at 2 ([streams, _dcb, SeqKey]), so a single get_many %% cannot span both depths — we run one query per depth and merge. %% %% The regular query is authoritative for error propagation (the %% `[streams]' root always exists); the DCB subtree is best-effort %% (absent when no DCB events have been written → empty, not an error). -spec query_event_results(atom(), khepri:condition()) -> {ok, [{[atom() | binary()], term()}]} | {error, term()}. query_event_results(StoreId, LeafMatch) -> RegularPattern = [streams, ?KHEPRI_WILDCARD_STAR, ?KHEPRI_WILDCARD_STAR, LeafMatch], case khepri:get_many(StoreId, RegularPattern) of {ok, Regular} when is_map(Regular) -> DcbPattern = ?DCB_STREAM_PATH ++ [LeafMatch], {ok, maps:to_list(Regular) ++ get_many_list(StoreId, DcbPattern)}; {ok, _} -> {ok, []}; {error, _} = Error -> Error end. %% @private Best-effort get_many returning a plain list ([] on any miss). -spec get_many_list(atom(), khepri_path:native_pattern()) -> [{[atom() | binary()], term()}]. get_many_list(StoreId, Pattern) -> case khepri:get_many(StoreId, Pattern) of {ok, Results} when is_map(Results) -> maps:to_list(Results); _ -> [] end. %% @doc Read all events of specific types from all streams using Khepri native filtering. %% %% This function uses Khepri's built-in #if_data_matches condition to filter %% events by type at the database level, avoiding loading all events into memory. %% %% Parameters: %% StoreId - The store identifier %% EventTypes - List of event type binaries to match %% BatchSize - Maximum number of events to return (for pagination) %% %% Returns events sorted by epoch_us (global ordering). -spec read_by_event_types(atom(), [binary()], pos_integer()) -> {ok, [event()]} | {error, term()}. read_by_event_types(StoreId, EventTypes, BatchSize) when is_list(EventTypes) -> case reckon_db_index_config:is_indexed(StoreId, event_type) of true -> limit_index_result( reckon_db_index:lookup_event_types(StoreId, EventTypes), BatchSize); false -> warn_unindexed(StoreId, event_type), scan_by_event_types(StoreId, EventTypes, BatchSize) end. %% @private Whole-store scan fallback for stores that did not declare the %% `event_type' index. Pushes the type match into Khepri (DB-side %% #if_data_matches) so non-matching events aren't loaded. -spec scan_by_event_types(atom(), [binary()], pos_integer()) -> {ok, [event()]} | {error, term()}. scan_by_event_types(StoreId, EventTypes, BatchSize) -> %% The leaf condition is applied at both depths by query_event_results %% (regular 4-level events + the 2-level DCB log). TypeConditions = [ #if_data_matches{pattern = #event{event_type = ET, _ = '_'}} || ET <- EventTypes ], LeafMatch = case TypeConditions of [] -> has_data_leaf(); [SingleCondition] -> #if_all{conditions = [ ?KHEPRI_WILDCARD_STAR, #if_has_data{has_data = true}, SingleCondition ]}; _ -> #if_all{conditions = [ ?KHEPRI_WILDCARD_STAR, #if_has_data{has_data = true}, #if_any{conditions = TypeConditions} ]} end, case query_event_results(StoreId, LeafMatch) of {ok, Results} -> Events = [convert_result_to_event(PathKey, Value) || {PathKey, Value} <- Results], ValidEvents = [E || E <- Events, E =/= undefined], SortedEvents = sort_by_epoch(ValidEvents), LimitedEvents = lists:sublist(SortedEvents, BatchSize), {ok, LimitedEvents}; {error, _} = Error -> Error end. %% @doc Read all events matching tags from all streams. %% %% Tags provide a mechanism for cross-stream querying without affecting %% stream-based concurrency control. This is useful for the process-centric %% model where you want to find all events related to specific participants. %% %% == Match Modes == %% %% `any' (default): Returns events containing ANY of the specified tags (union). %% Example: `read_by_tags(Store, [<<"student:456">>, <<"student:789">>], any, 100)' %% Returns events for either student. %% %% `all': Returns events containing ALL of the specified tags (intersection). %% Example: `read_by_tags(Store, [<<"student:456">>, <<"course:CS101">>], all, 100)' %% Returns only events tagged with both student 456 AND course CS101. %% %% == Parameters == %% %% StoreId - The store identifier %% Tags - List of tag binaries to match %% Match - `any' | `all' (matching strategy) %% BatchSize - Maximum number of events to return %% %% == Returns == %% %% Events sorted by epoch_us (global ordering). -spec read_by_tags(atom(), [binary()], any | all, pos_integer()) -> {ok, [event()]} | {error, term()}. read_by_tags(StoreId, Tags, Match, BatchSize) when is_list(Tags), is_atom(Match) -> case reckon_db_index_config:is_indexed(StoreId, tags) of true -> limit_index_result( reckon_db_index:lookup_tags(StoreId, Tags, Match), BatchSize); false -> warn_unindexed(StoreId, tags), scan_by_tags(StoreId, Tags, Match, BatchSize) end. %% @private Whole-store tag scan fallback for stores that did not declare %% the `tags' index. Khepri pattern matching doesn't support list %% membership, so events with data are fetched and filtered client-side. %% This is exactly the cross-cutting scan the secondary index replaces. -spec scan_by_tags(atom(), [binary()], any | all, pos_integer()) -> {ok, [event()]} | {error, term()}. scan_by_tags(StoreId, Tags, Match, BatchSize) -> case query_event_results(StoreId, has_data_leaf()) of {ok, Results} -> AllEvents = [convert_result_to_event(PathKey, Value) || {PathKey, Value} <- Results], ValidEvents = [E || E <- AllEvents, E =/= undefined], FilteredEvents = filter_events_by_tags(ValidEvents, Tags, Match), SortedEvents = sort_by_epoch(FilteredEvents), LimitedEvents = lists:sublist(SortedEvents, BatchSize), {ok, LimitedEvents}; {error, _} = Error -> Error end. %% @doc Read all events whose metadata key = value. %% %% This is the sanctioned primitive applications build causation / %% correlation / saga read models on (e.g. `read_by_metadata(Store, %% <<"causation_id">>, EventId)`). The store returns events matching a %% metadata key=value pair — bounded and indexed when `{meta, Key}' is %% declared — and does NOT interpret what the key means. Lineage %% traversal, graphs, and read models are the application's job. %% %% Indexed (O(matches)) when the store declared `{meta, Key}'; otherwise %% a one-time `logger:warning' is emitted and the query falls back to a %% whole-store scan (O(total events)). -spec read_by_metadata(atom(), binary(), binary()) -> {ok, [event()]} | {error, term()}. read_by_metadata(StoreId, Key, Value) when is_binary(Key), is_binary(Value) -> case reckon_db_index_config:is_indexed(StoreId, {meta, Key}) of true -> reckon_db_index:lookup_meta(StoreId, Key, Value); false -> warn_unindexed(StoreId, {meta, Key}), scan_by_metadata(StoreId, Key, Value) end. %% @private Whole-store scan fallback for an un-indexed metadata key. -spec scan_by_metadata(atom(), binary(), binary()) -> {ok, [event()]} | {error, term()}. scan_by_metadata(StoreId, Key, Value) -> case query_event_results(StoreId, has_data_leaf()) of {ok, Results} -> Events = [convert_result_to_event(P, V) || {P, V} <- Results], Matching = [E || #event{metadata = M} = E <- Events, is_map(M), maps:get(Key, M, undefined) =:= Value], {ok, sort_by_epoch(Matching)}; {error, _} = Error -> Error end. %% @private Apply the BatchSize limit to an index lookup result. -spec limit_index_result({ok, [event()]} | {error, term()}, pos_integer()) -> {ok, [event()]} | {error, term()}. limit_index_result({ok, Events}, BatchSize) -> {ok, lists:sublist(Events, BatchSize)}; limit_index_result({error, _} = Error, _BatchSize) -> Error. %% @private Warn ONCE per (store, index-kind) that a cross-cutting query %% is hitting an un-indexed scan, so operators can declare the index. No %% silent truncation — the query still runs, just O(store). The %% persistent_term flag keeps it to a single line per kind. -spec warn_unindexed(atom(), index_decl()) -> ok. warn_unindexed(StoreId, Kind) -> Flag = {reckon_db, idx_unindexed_warned, StoreId, Kind}, case persistent_term:get(Flag, false) of true -> ok; false -> persistent_term:put(Flag, true), logger:warning( "reckon_db: cross-cutting query on un-indexed ~p (store ~p) — " "O(store) scan; declare the index in store_config to make it " "O(matches)", [Kind, StoreId]), ok end. %% @private Filter events by tags according to match mode %% Empty search tags always returns empty (no criteria = no results) -spec filter_events_by_tags([event()], [binary()], any | all) -> [event()]. filter_events_by_tags(_Events, [], _Match) -> []; filter_events_by_tags(Events, Tags, any) -> %% Return events that have ANY of the tags (union) TagSet = sets:from_list(Tags), lists:filter( fun(#event{tags = EventTags}) when is_list(EventTags), EventTags =/= [] -> EventTagSet = sets:from_list(EventTags), sets:size(sets:intersection(TagSet, EventTagSet)) > 0; (_) -> false end, Events ); filter_events_by_tags(Events, Tags, all) -> %% Return events that have ALL of the tags (intersection) TagSet = sets:from_list(Tags), lists:filter( fun(#event{tags = EventTags}) when is_list(EventTags), EventTags =/= [] -> EventTagSet = sets:from_list(EventTags), sets:is_subset(TagSet, EventTagSet); (_) -> false end, Events ). %% @private Convert a Khepri result to an event, deriving stream_id from %% the path. Handles both the 4-level regular layout %% ([streams, Type, Id, Version]) and the 2-level DCB log %% ([streams, _dcb, SeqKey]); `stream_id_from_path/1' reconstructs the %% opaque id for both. -spec convert_result_to_event([atom() | binary()], term()) -> event() | undefined. convert_result_to_event([streams | _] = Path, Event) when is_record(Event, event) -> Event#event{stream_id = reckon_db_stream_path:stream_id_from_path(Path)}; convert_result_to_event([streams | _] = Path, EventMap) when is_map(EventMap) -> Event = map_to_event(EventMap), Event#event{stream_id = reckon_db_stream_path:stream_id_from_path(Path)}; convert_result_to_event(_, _) -> undefined. %% @doc Get current version of a stream %% %% Returns: %% -1 - if stream doesn't exist or is empty %% N >= 0 - representing the version of the latest event -spec get_version(atom(), binary()) -> integer(). get_version(StoreId, StreamId) -> case khepri:count(StoreId, reckon_db_stream_path:versions_pattern(StreamId)) of {ok, 0} -> ?NO_STREAM; {ok, Count} -> Count - 1; {error, _} -> ?NO_STREAM end. %% @doc Check if a stream exists -spec exists(atom(), binary()) -> boolean(). exists(StoreId, StreamId) -> Path = reckon_db_stream_path:stream_path(StreamId), case khepri:exists(StoreId, Path) of true -> true; false -> false; {error, _} -> false end. %% @doc Check if a store contains at least one event. %% Cannot rely on stream existence alone — streams can survive %% after all their events are deleted (truncation, GDPR erasure). %% Checks for actual event data by reading 1 event globally. -spec has_events(atom()) -> boolean(). has_events(StoreId) -> case read_all_global(StoreId, 0, 1) of {ok, [_ | _]} -> true; _ -> false end. %% @doc List all streams in the store -spec list_streams(atom()) -> {ok, [binary()]} | {error, term()}. list_streams(StoreId) -> %% Aggregate-id nodes live at depth 3: [streams, Type, Id]. One such %% node per stream (its children are the version leaves). Reconstruct %% the opaque id and drop the reserved DCB pseudo-stream — it is not a %% user stream. Path = ?STREAMS_PATH ++ [?KHEPRI_WILDCARD_STAR, ?KHEPRI_WILDCARD_STAR], case khepri:get_many(StoreId, Path) of {ok, Results} when is_map(Results) -> StreamIds = lists:usort([ reckon_db_stream_path:stream_id_from_path(P) || P <- maps:keys(Results), not is_dcb_path(P) ]), {ok, StreamIds}; {ok, _} -> {ok, []}; {error, _} = Error -> Error end. %% @private True for the DCB pseudo-stream's 2-level nodes %% ([streams, _dcb, SeqKey]) — excluded from user-facing stream listings. -spec is_dcb_path([atom() | binary()]) -> boolean(). is_dcb_path([streams, ?DCB_STREAM | _]) -> true; is_dcb_path(_) -> false. %% @doc Delete a stream and all its events -spec delete(atom(), binary()) -> ok | {error, term()}. delete(StoreId, StreamId) -> Path = reckon_db_stream_path:stream_path(StreamId), case khepri:delete(StoreId, Path) of ok -> ok; {error, _} = Error -> Error end. %%==================================================================== %% Internal functions %%==================================================================== %% @private -spec do_append(atom(), binary(), integer(), [new_event()]) -> {ok, non_neg_integer()} | {error, term()}. do_append(StoreId, StreamId, ExpectedVersion, Events) -> CurrentVersion = get_version(StoreId, StreamId), %% Check expected version case check_expected_version(ExpectedVersion, CurrentVersion) of ok -> append_events_to_stream(StoreId, StreamId, CurrentVersion, Events); {error, _} = Error -> Error end. %% @private -spec check_expected_version(integer(), integer()) -> ok | {error, term()}. check_expected_version(?ANY_VERSION, _CurrentVersion) -> ok; check_expected_version(?NO_STREAM, CurrentVersion) when CurrentVersion =:= ?NO_STREAM -> ok; check_expected_version(?NO_STREAM, CurrentVersion) -> {error, {wrong_expected_version, ?NO_STREAM, CurrentVersion}}; check_expected_version(ExpectedVersion, CurrentVersion) when ExpectedVersion =:= CurrentVersion -> ok; check_expected_version(ExpectedVersion, CurrentVersion) -> {error, {wrong_expected_version, ExpectedVersion, CurrentVersion}}. %% @private -spec append_events_to_stream(atom(), binary(), integer(), [new_event()]) -> {ok, non_neg_integer()} | {error, term()}. append_events_to_stream(StoreId, StreamId, CurrentVersion, Events) -> Now = erlang:system_time(millisecond), EpochUs = erlang:system_time(microsecond), %% Resolve the tamper-resistance context for this batch. For %% stores with integrity disabled (the default and only mode %% pre-2.1) this is a constant `disabled` pass-through. For %% integrity-enabled stores we load the HMAC key, ensure the %% per-stream watermark exists, and resolve the initial chain %% tip — all once, before the per-event loop. IntegrityCtx = setup_integrity(StoreId, StreamId, CurrentVersion), InitialTip = resolve_initial_tip(StoreId, StreamId, CurrentVersion + 1, IntegrityCtx), %% Build the event records AND their secondary-index entries OUTSIDE %% the transaction. Integrity stamping (HMAC + chain) reads the HMAC %% key from persistent_term and runs crypto — neither is available %% inside a Khepri (Ra) transaction fun, which must be deterministic %% and side-effect-free. This mirrors the DCB append: stamp outside, %% write inside. Declared = reckon_db_index_config:declared(StoreId), {Writes, IndexEntries, FinalVersion} = build_event_writes( StreamId, CurrentVersion, Events, Now, EpochUs, IntegrityCtx, InitialTip, Declared), %% Write the whole batch — every event record and every index entry — %% in ONE transaction. Either all land or none do, so the index is %% never left partially populated (a dropped index write would make %% queries silently incomplete — worse than slow). One Ra command per %% batch. %% %% NOTE: the optimistic-version check stays OUTSIDE the transaction %% (in do_append, as before) — making it transactional to close the %% per-stream concurrent-same-version TOCTOU is a separate, pre-existing %% concern and is intentionally not changed here. %% %% A failed transaction — most importantly {error, noproc} while the %% store's Ra server is mid-(re)start — MUST surface as a retriable %% {error, _} return, NOT a hard `ok =` badmatch (the boot-time %% crash-loop lesson the old per-put loop encoded). khepri:transaction %% returns {error, _} for a not-ready store; reckon-gater retries. case write_batch(StoreId, Writes, IndexEntries) of ok -> {ok, FinalVersion}; {error, _} = Error -> Error end. %% @private Build the {Path, Record} event writes and the flat list of %% {Path, EventRef} index entries for a batch, applying integrity stamping %% per event. Returns the writes, index entries, and the final version. -spec build_event_writes( binary(), integer(), [new_event()], integer(), integer(), integrity_ctx(), binary() | undefined, [index_decl()] ) -> {[{khepri_path:native_path(), event()}], [{khepri_path:native_path(), reckon_db_index:event_ref()}], non_neg_integer()}. build_event_writes(StreamId, CurrentVersion, Events, Now, EpochUs, IntegrityCtx, InitialTip, Declared) -> {WritesRev, IndexRev, FinalVersion, _FinalTip} = lists:foldl( fun(Event, {WAcc, IAcc, AccVersion, AccTip}) -> NewVersion = AccVersion + 1, RecordedEvent0 = create_event_record( Event, StreamId, NewVersion, Now, EpochUs), {RecordedEvent, NextTip} = apply_integrity_if_enabled( RecordedEvent0, AccTip, IntegrityCtx), PaddedVersion = pad_version(NewVersion, ?VERSION_PADDING), Path = reckon_db_stream_path:event_path(StreamId, PaddedVersion), Entries = reckon_db_index:entries(RecordedEvent, Declared), {[{Path, RecordedEvent} | WAcc], Entries ++ IAcc, NewVersion, NextTip} end, {[], [], CurrentVersion, InitialTip}, Events), {lists:reverse(WritesRev), IndexRev, FinalVersion}. %% @private Write all event records + index entries in one transaction. -spec write_batch( atom(), [{khepri_path:native_path(), event()}], [{khepri_path:native_path(), reckon_db_index:event_ref()}] ) -> ok | {error, term()}. write_batch(StoreId, Writes, IndexEntries) -> Fun = fun() -> lists:foreach(fun({P, Record}) -> ok = khepri_tx:put(P, Record) end, Writes), lists:foreach(fun({P, Ref}) -> ok = khepri_tx:put(P, Ref) end, IndexEntries), ok end, case khepri:transaction(StoreId, Fun) of {ok, ok} -> ok; ok -> ok; %% Khepri 0.17.x catch-all wraps process_command errors as {ok, Err} %% when Ra is not yet ready (e.g. store startup race). {ok, {error, E}} -> {error, E}; {error, _} = Error -> Error end. %% @private Tamper-resistance context for a single append batch. %% %% Either `disabled` (no integrity work) or %% `{enabled, Key, ChainStart}` carrying the HMAC key and the %% chain-start watermark for this stream. -type integrity_ctx() :: disabled | {enabled, Key :: binary(), ChainStart :: non_neg_integer()}. -spec setup_integrity(atom(), binary(), integer()) -> integrity_ctx(). setup_integrity(StoreId, StreamId, CurrentVersion) -> case reckon_db_integrity_key:is_enabled(StoreId) of false -> disabled; true -> NextVersion = CurrentVersion + 1, {ok, ChainStart} = reckon_db_chain_watermark:set_if_absent( StoreId, StreamId, NextVersion), Key = reckon_db_integrity_key:get(StoreId), {enabled, Key, ChainStart} end. %% @private Resolve the chain-tip value that the FIRST event in this %% batch must reference as its `prev_event_hash`. %% %% Disabled context: tip is `undefined` (unused). %% Enabled, first integrity event in stream (NextVersion =:= ChainStart): %% tip is the genesis 32-zero-byte value. %% Enabled, later batch (NextVersion > ChainStart): %% tip is computed from the predecessor event on disk. -spec resolve_initial_tip( atom(), binary(), non_neg_integer(), integrity_ctx() ) -> binary() | undefined. resolve_initial_tip(_StoreId, _StreamId, _NextVersion, disabled) -> undefined; resolve_initial_tip(_StoreId, _StreamId, NextVersion, {enabled, _Key, ChainStart}) when NextVersion =:= ChainStart -> reckon_gater_integrity:genesis_prev_hash(); resolve_initial_tip(StoreId, StreamId, NextVersion, {enabled, _Key, ChainStart}) when NextVersion > ChainStart -> PrevVersion = NextVersion - 1, PaddedVersion = pad_version(PrevVersion, ?VERSION_PADDING), Path = reckon_db_stream_path:event_path(StreamId, PaddedVersion), case khepri:get(StoreId, Path) of {ok, #event{prev_event_hash = PrevPrevHash} = PrevEvent} when is_binary(PrevPrevHash) -> reckon_gater_integrity:compute_chain_hash(PrevEvent, PrevPrevHash); Other -> %% Invariant violation: stream's watermark says the %% predecessor should be integrity-bearing, but we cannot %% find a usable predecessor. Surface fast — replay %% won't be able to verify anyway. erlang:error({integrity_setup_failed, #{stream_id => StreamId, looking_for_version => PrevVersion, chain_start => ChainStart, got => Other}}) end. %% @private Compute and attach the integrity fields for one event. %% %% Disabled context: pass-through. %% Enabled: set prev_event_hash to the running tip, compute MAC, then %% compute the next tip (= chain hash of the just-built event) for the %% next iteration of the fold. -spec apply_integrity_if_enabled( event(), Tip :: binary() | undefined, integrity_ctx() ) -> {event(), NextTip :: binary() | undefined}. apply_integrity_if_enabled(Event, _Tip, disabled) -> {Event, undefined}; apply_integrity_if_enabled(#event{} = Event, Tip, {enabled, Key, _ChainStart}) when is_binary(Tip) -> Event1 = Event#event{prev_event_hash = Tip}, Mac = reckon_gater_integrity:compute_event_mac(Event1, Key), Event2 = Event1#event{mac = Mac}, NextTip = reckon_gater_integrity:compute_chain_hash(Event2, Tip), {Event2, NextTip}. %% @private -spec create_event_record(new_event(), binary(), non_neg_integer(), integer(), integer()) -> event(). create_event_record(Event, StreamId, Version, Timestamp, EpochUs) -> EventId = maps:get(event_id, Event, generate_event_id()), EventType = maps:get(event_type, Event), Data = maps:get(data, Event), Metadata = maps:get(metadata, Event, #{}), Tags = maps:get(tags, Event, undefined), DataContentType = maps:get(data_content_type, Event, ?CONTENT_TYPE_JSON), MetadataContentType = maps:get(metadata_content_type, Event, ?CONTENT_TYPE_JSON), #event{ event_id = EventId, event_type = EventType, stream_id = StreamId, version = Version, data = Data, metadata = Metadata, tags = Tags, timestamp = Timestamp, epoch_us = EpochUs, data_content_type = DataContentType, metadata_content_type = MetadataContentType }. %% @private -spec generate_event_id() -> binary(). generate_event_id() -> reckon_gater_uuid:to_string(reckon_gater_uuid:v7()). %% @private -spec pad_version(non_neg_integer(), pos_integer()) -> binary(). pad_version(Version, Length) -> VersionStr = integer_to_list(Version), Padding = Length - length(VersionStr), PaddedStr = lists:duplicate(Padding, $0) ++ VersionStr, list_to_binary(PaddedStr). %% @private Backward-compatible: no verification (for internal callers %% that have not yet adopted Opts). -spec do_read(atom(), binary(), non_neg_integer(), pos_integer(), direction()) -> {ok, [event()]} | {error, term()}. do_read(StoreId, StreamId, StartVersion, Count, Direction) -> do_read_with_verify( StoreId, StreamId, StartVersion, Count, Direction, #{verify => skip_all}). %% @private Read with verification applied per the resolved options. -spec do_read_with_verify( atom(), binary(), non_neg_integer(), pos_integer(), direction(), read_opts() ) -> {ok, [event()]} | {error, term()}. do_read_with_verify(StoreId, StreamId, StartVersion, Count, Direction, Opts) -> case exists(StoreId, StreamId) of false -> {error, {stream_not_found, StreamId}}; true -> case read_events(StoreId, StreamId, StartVersion, Count, Direction) of {ok, Events} -> case maybe_verify_events( StoreId, StreamId, StartVersion, Direction, Events, Opts) of {ok, _} = Ok -> Ok; {integrity_violation, _} = Violation -> %% Wrap at the public API boundary so %% callers see {error, _}. Internal %% helpers and the gater module return %% the bare tuple. {error, Violation} end; Other -> Other end end. %% @private Apply the configured verification mode to a fresh result. %% %% Backward reads verify the same chain as forward reads by walking %% the result in forward order (smallest version first) and then %% reversing the returned list to preserve the caller's requested %% ordering. The chain semantics are direction-independent: every %% event's `prev_event_hash` must equal the chain hash of its %% predecessor, regardless of the order the caller chose to receive %% them in. -spec maybe_verify_events( atom(), binary(), non_neg_integer(), direction(), [event()], read_opts() ) -> {ok, [event()]} | {error, term()}. maybe_verify_events(StoreId, StreamId, StartVersion, Direction, Events, Opts) -> Mode = maps:get(verify, Opts, skip_legacy), case verify_required(StoreId, Direction, Mode) of false -> {ok, Events}; true -> verify_in_direction( StoreId, StreamId, StartVersion, Direction, Events, Mode) end. %% @private Verification runs on integrity-enabled stores regardless %% of read direction. The verification surface is the same in both %% directions — the only difference is the result-ordering of the %% returned events. -spec verify_required(atom(), direction(), verify_mode()) -> boolean(). verify_required(_StoreId, _Direction, skip_all) -> false; verify_required(StoreId, _Direction, _Mode) -> reckon_db_integrity_key:is_enabled(StoreId). %% @private Dispatch on direction: forward verifies in place; backward %% reverses to forward order, verifies, and reverses the result back. -spec verify_in_direction( atom(), binary(), non_neg_integer(), direction(), [event()], verify_mode() ) -> {ok, [event()]} | {error, term()}. verify_in_direction(StoreId, StreamId, StartVersion, forward, Events, Mode) -> verify_events_forward(StoreId, StreamId, StartVersion, Events, Mode); verify_in_direction(StoreId, StreamId, StartVersion, backward, Events, Mode) -> %% Backward reads return events highest-version-first. To verify %% the chain we re-order them lowest-version-first, walk the %% forward verifier, then reverse the verified list before %% returning so the caller gets the ordering they asked for. %% The forward verifier's `StartVersion` is the LOWEST version in %% the batch, which for a backward read is the LAST event. ForwardEvents = lists:reverse(Events), ForwardStart = case ForwardEvents of [] -> StartVersion; [#event{version = V} | _] -> V end, case verify_events_forward( StoreId, StreamId, ForwardStart, ForwardEvents, Mode) of {ok, Verified} -> {ok, lists:reverse(Verified)}; {integrity_violation, _} = Violation -> Violation end. %% @private Walk events in forward order, verifying each against the %% running chain tip. Short-circuits on the first integrity_violation. -spec verify_events_forward( atom(), binary(), non_neg_integer(), [event()], verify_mode() ) -> {ok, [event()]} | {error, term()}. verify_events_forward(StoreId, StreamId, StartVersion, Events, Mode) -> {ok, ChainStart} = reckon_db_chain_watermark:lookup(StoreId, StreamId), Key = reckon_db_integrity_key:get(StoreId), InitialTip = resolve_read_initial_tip( StoreId, StreamId, StartVersion, ChainStart), verify_events_loop(Events, InitialTip, ChainStart, Key, Mode, StreamId, []). %% @private verify_events_loop([], _Tip, _ChainStart, _Key, _Mode, _StreamId, Acc) -> {ok, lists:reverse(Acc)}; verify_events_loop([Event | Rest], Tip, ChainStart, Key, Mode, StreamId, Acc) -> case is_legacy_event(Event, ChainStart) of true -> handle_legacy(Event, Rest, Tip, ChainStart, Key, Mode, StreamId, Acc); false -> handle_integrity(Event, Rest, Tip, ChainStart, Key, Mode, StreamId, Acc) end. %% @private An event is legacy if it predates the watermark (or the %% watermark is absent, meaning the stream never had integrity events). is_legacy_event(_Event, undefined) -> true; is_legacy_event(#event{version = V}, ChainStart) when is_integer(ChainStart) -> V < ChainStart. handle_legacy(Event, Rest, Tip, ChainStart, Key, strict, StreamId, _Acc) -> %% Strict mode refuses to return legacy events at all. _ = Tip, _ = ChainStart, _ = Key, _ = Rest, {integrity_violation, #{ layer => storage, stream_id => StreamId, version => Event#event.version, kind => missing_integrity, context => #{detail => legacy_event_under_strict_mode} }}; handle_legacy(Event, Rest, Tip, ChainStart, Key, Mode, StreamId, Acc) -> %% skip_legacy: return the legacy event untouched. Emit telemetry %% so operators can monitor remediation progress. telemetry:execute( [reckon, db, read, legacy_event_returned], #{system_time => erlang:system_time(millisecond)}, #{store_id_hint => StreamId, version => Event#event.version} ), verify_events_loop(Rest, Tip, ChainStart, Key, Mode, StreamId, [Event | Acc]). handle_integrity(Event, Rest, undefined, ChainStart, Key, Mode, StreamId, Acc) when is_integer(ChainStart), is_record(Event, event), Event#event.version =:= ChainStart -> %% Mid-read transition from legacy region to integrity region: the %% first integrity-bearing event in a stream was written with %% prev_event_hash = genesis. Seed the running tip accordingly so %% the verifier has a concrete value to compare against. Genesis = reckon_gater_integrity:genesis_prev_hash(), handle_integrity(Event, Rest, Genesis, ChainStart, Key, Mode, StreamId, Acc); handle_integrity(Event, Rest, Tip, ChainStart, Key, Mode, StreamId, Acc) when is_binary(Tip) -> case reckon_gater_integrity:verify_event(Event, Tip, Key) of ok -> NextTip = reckon_gater_integrity:compute_chain_hash(Event, Tip), verify_events_loop( Rest, NextTip, ChainStart, Key, Mode, StreamId, [Event | Acc]); {integrity_violation, _} = Violation -> Violation end. %% @private Resolve the chain tip that the FIRST event in the returned %% batch should reference as its `prev_event_hash`. %% %% - undefined watermark or StartVersion < watermark: legacy-only; %% the running tip is irrelevant (verification won't be called). %% - StartVersion == watermark: tip = genesis (this is the first %% integrity-bearing event in the stream). %% - StartVersion > watermark: tip = chain_hash of the event at %% (StartVersion - 1), which must itself be integrity-bearing. -spec resolve_read_initial_tip( atom(), binary(), non_neg_integer(), non_neg_integer() | undefined ) -> binary() | undefined. resolve_read_initial_tip(_StoreId, _StreamId, _StartVersion, undefined) -> undefined; resolve_read_initial_tip(_StoreId, _StreamId, StartVersion, ChainStart) when StartVersion < ChainStart -> undefined; resolve_read_initial_tip(_StoreId, _StreamId, StartVersion, ChainStart) when StartVersion =:= ChainStart -> reckon_gater_integrity:genesis_prev_hash(); resolve_read_initial_tip(StoreId, StreamId, StartVersion, _ChainStart) when StartVersion > 0 -> PrevVersion = StartVersion - 1, PaddedVersion = pad_version(PrevVersion, ?VERSION_PADDING), Path = reckon_db_stream_path:event_path(StreamId, PaddedVersion), case khepri:get(StoreId, Path) of {ok, #event{prev_event_hash = PrevPrevHash} = PrevEvent} when is_binary(PrevPrevHash) -> reckon_gater_integrity:compute_chain_hash(PrevEvent, PrevPrevHash); _ -> %% No usable predecessor; legacy regime in practice. undefined end. %% @private -spec read_events(atom(), binary(), non_neg_integer(), pos_integer(), direction()) -> {ok, [event()]}. read_events(StoreId, StreamId, StartVersion, Count, Direction) -> Versions = calculate_versions(StartVersion, Count, Direction), Events = lists:filtermap( fun(Version) -> PaddedVersion = pad_version(Version, ?VERSION_PADDING), Path = reckon_db_stream_path:event_path(StreamId, PaddedVersion), case khepri:get(StoreId, Path) of {ok, Event} when is_record(Event, event) -> {true, Event}; {ok, EventMap} when is_map(EventMap) -> %% Convert map to record if needed {true, map_to_event(EventMap)}; _ -> false end end, Versions ), {ok, Events}. %% @private -spec calculate_versions(non_neg_integer(), pos_integer(), direction()) -> [non_neg_integer()]. calculate_versions(StartVersion, Count, forward) -> lists:seq(StartVersion, StartVersion + Count - 1); calculate_versions(StartVersion, Count, backward) -> EndVersion = max(0, StartVersion - Count + 1), lists:reverse(lists:seq(EndVersion, StartVersion)). %% @private -spec map_to_event(map()) -> event(). map_to_event(Map) -> #event{ event_id = maps:get(event_id, Map, undefined), event_type = maps:get(event_type, Map, undefined), stream_id = maps:get(stream_id, Map, undefined), version = maps:get(version, Map, 0), data = maps:get(data, Map, #{}), metadata = maps:get(metadata, Map, #{}), tags = maps:get(tags, Map, undefined), timestamp = maps:get(timestamp, Map, 0), epoch_us = maps:get(epoch_us, Map, 0), data_content_type = maps:get(data_content_type, Map, ?CONTENT_TYPE_JSON), metadata_content_type = maps:get(metadata_content_type, Map, ?CONTENT_TYPE_JSON) }. %% @private Safe version of lists:nthtail that returns [] when Offset >= length. -spec safe_nthtail(non_neg_integer(), list()) -> list(). safe_nthtail(0, List) -> List; safe_nthtail(_, []) -> []; safe_nthtail(N, [_ | Rest]) -> safe_nthtail(N - 1, Rest).