Merge pull request #35 from ccamel/copilot/rename-callbacks-for-consistency

refactor: rename event store callbacks to align with ES domain terminology

Author: ccamel

README.md

@@ -120,13 +120,13 @@ The event store is a core component in this experiment, designed as a customizab
 % Shuts down the event store.
 -callback stop() -> {ok} | {error, term()}.
 
-% Persists a list of events for a given stream.
--callback persist_events(StreamId, Events) -> ok | {error, term()}
+% Appends a list of events for a given stream.
+-callback append(StreamId, Events) -> ok | {error, term()}
     when StreamId :: stream_id(),
          Events :: [event()].
 
-% Retrieves events from a stream and folds them using a provided function
--callback retrieve_and_fold_events(StreamId, Options, FoldFun, InitialAcc) -> {ok, Acc} | {error, term()}
+% Folds events from a stream using a provided function
+-callback fold(StreamId, Options, FoldFun, InitialAcc) -> {ok, Acc} | {error, term()}
     when StreamId :: stream_id(),
          Options :: fold_events_opts(),
          FoldFun :: fold_events_fun(),
@@ -144,11 +144,11 @@ The event store supports snapshotting to optimize aggregate rehydration. Instead
 **Snapshot Callbacks:**
 
 ```erlang
-% Save a snapshot of aggregate state
--callback save_snapshot(Snapshot) -> ok when Snapshot :: snapshot().
+% Store a snapshot of aggregate state
+-callback store(Snapshot) -> ok when Snapshot :: snapshot().
 
-% Retrieve the latest snapshot for a stream
--callback retrieve_latest_snapshot(StreamId) -> {ok, Snapshot} | {error, not_found}.
+% Load the latest snapshot for a stream
+-callback load_latest(StreamId) -> {ok, Snapshot} | {error, not_found}.
 ```
 
 The snapshot record contains all necessary fields (domain, stream_id, sequence, timestamp, state), making the API consistent with event persistence where events are passed as complete records.

apps/event_sourcing_contract/src/event_sourcing_event_store_behaviour.erl

@@ -8,8 +8,8 @@ be replayed for state reconstruction.
 
 Callbacks:
 - `start/0`, `stop/0` — manage the backend's lifecycle
-- `persist_events/2` — append events to a stream, ensuring monotonic sequence numbers
-- `retrieve_and_fold_events/4` — replay events in order and fold them with a user function
+- `append/2` — append events to a stream, ensuring monotonic sequence numbers
+- `fold/4` — replay events in order and fold them with a user function
 
 Implementations must guarantee:
 - **ordering**: events for a given stream are stored and replayed in strictly increasing
@@ -46,7 +46,9 @@ Returns `ok` on success. May throw an exception if cleanup fails
 -doc """
 Append events to an event stream.
 
-This callback appends events to the specified streams.
+This callback appends events to the specified stream, ensuring monotonic ordering
+by sequence number. Each event is durably persisted and becomes part of the immutable
+event log for the stream.
 
 - StreamId is an atom identifying the event stream (e.g., order-123).
 - Events is the list of events to append to the stream. The events provided are unique and all
@@ -55,7 +57,7 @@ belong to the same stream.
 Returns `ok` on success. May throw an exception if persistence fails (e.g., badarg if the
 stream ID is incorrect, duplicate events if the sequence number is not unique).
 """.
--callback persist_events(StreamId, Events) -> ok when
+-callback append(StreamId, Events) -> ok when
     StreamId :: stream_id(),
     Events :: [event()].
 -doc """
@@ -71,11 +73,11 @@ rebuild application state by replaying events.
   - `{to, Sequence | infinity}`: End at this sequence (default: infinity).
   - `{limit, Limit}`: Maximum number of events to retrieve (default: infinity).
 - FoldFun is a function `fun((Event, AccIn) -> AccOut)` to process each event.
-- InitialAcc is The initial accumulator value (e.g., an empty state).
+- InitialAcc is the initial accumulator value (e.g., an empty state).
 
-Returns `{ok, Acc}` where `Acc` is the result of folding all events.
+Returns the final accumulator after folding all events.
 """.
--callback retrieve_and_fold_events(StreamId, Options, Fun, Acc0) -> Acc1 when
+-callback fold(StreamId, Options, Fun, Acc0) -> Acc1 when
     StreamId :: stream_id(),
     Options :: fold_events_opts(),
     Fun :: fun((Event :: event(), AccIn) -> AccOut),

apps/event_sourcing_contract/src/event_sourcing_snapshot_store_behaviour.erl

@@ -7,8 +7,8 @@ representations of the state after applying a set of events.
 
 Callbacks:
 - `start/0`, `stop/0` — manage backend initialization and shutdown
-- `save_snapshot/1` — persist a snapshot of the aggregate state
-- `retrieve_latest_snapshot/1` — fetch the most recent snapshot for a given stream
+- `store/1` — persist a snapshot of the aggregate state
+- `load_latest/1` — fetch the most recent snapshot for a given stream
 
 Design principles:
 - Snapshots are **optional optimizations**; events remain the source of truth.
@@ -23,37 +23,38 @@ systems (e.g., S3).
 -include("event_sourcing.hrl").
 
 -doc """
-Save a snapshot for a stream.
+Store a snapshot for a stream.
 
-This callback saves a snapshot of the aggregate state at a specific point in time.
-The snapshot represents the aggregate state after all events up to and including
-the given sequence number have been applied.
+This callback persists a snapshot of the aggregate state at a specific point in time,
+representing the state after applying all events up to and including the recorded
+sequence number. Snapshots are optimization aids; events remain the source of truth.
 
-The snapshot record already contains all necessary information including domain,
-stream_id, sequence, timestamp, and state. This is consistent with how events
-are handled - they are passed as complete records rather than decomposed fields.
+The snapshot record contains all necessary information: domain, stream_id, sequence,
+timestamp, and state. This design is consistent with event persistence, where complete
+records are passed rather than decomposed fields.
 
 - Snapshot is the complete snapshot record to persist.
 
 Returns `ok` on success, or `{warning, Reason}` if persistence fails. Returning a
 warning is preferred over throwing an exception, as snapshot failures should not
-crash aggregates (events are the source of truth).
+crash aggregates.
 """.
--callback save_snapshot(Snapshot) -> ok | {warning, Reason} when
+-callback store(Snapshot) -> ok | {warning, Reason} when
     Snapshot :: snapshot(),
     Reason :: term().
 
 -doc """
-Retrieve the latest snapshot for a stream.
+Load the latest snapshot for a stream.
 
 This callback retrieves the most recent snapshot for the given stream, if one exists.
-The snapshot can be used to restore aggregate state without replaying all events.
+The snapshot enables fast state reconstruction by avoiding full event replay from
+the beginning of the stream.
 
 - StreamId is the unique identifier for the stream.
 
 Returns `{ok, Snapshot}` if a snapshot exists, or `{error, not_found}` if no snapshot
 has been saved for this stream.
 """.
--callback retrieve_latest_snapshot(StreamId) -> {ok, Snapshot} | {error, not_found} when
+-callback load_latest(StreamId) -> {ok, Snapshot} | {error, not_found} when
     StreamId :: stream_id(),
     Snapshot :: snapshot().

apps/event_sourcing_core/src/event_sourcing_core_aggregate.erl

@@ -130,7 +130,7 @@ init({Aggregate, StoreContext, Id, Opts}) ->
 
     %% Try to load the latest snapshot
     {StateFromSnapshot, SequenceFromSnapshot} =
-        case event_sourcing_core_store:retrieve_latest_snapshot(StoreContext, Id) of
+        case event_sourcing_core_store:load_latest(StoreContext, Id) of
             {ok, Snapshot} ->
                 SnapshotState = event_sourcing_core_store:snapshot_state(Snapshot),
                 SnapshotSeq = event_sourcing_core_store:snapshot_sequence(Snapshot),
@@ -150,7 +150,7 @@ init({Aggregate, StoreContext, Id, Opts}) ->
             }
         end,
     {State1, Sequence1} =
-        event_sourcing_core_store:retrieve_and_fold_events(
+        event_sourcing_core_store:fold(
             StoreContext,
             Id,
             #{from => SequenceFromSnapshot + 1},
@@ -354,7 +354,7 @@ persist_events(PayloadEvents, {Aggregate, StoreContext, Id, Sequence0, SequenceN
         end,
         Events
     ),
-    event_sourcing_core_store:persist_events(StoreContext, Id, Events).
+    event_sourcing_core_store:append(StoreContext, Id, Events).
 
 -doc """
 Saves a snapshot if the snapshot interval is configured and the current
@@ -379,7 +379,7 @@ maybe_save_snapshot(
     Timestamp = NowFun(),
     logger:info("Saving snapshot for ~p at sequence ~p", [Id, Sequence]),
     Snapshot = event_sourcing_core_store:new_snapshot(Aggregate, Id, Sequence, Timestamp, AggState),
-    case event_sourcing_core_store:save_snapshot(StoreContext, Snapshot) of
+    case event_sourcing_core_store:store(StoreContext, Snapshot) of
         ok ->
             ok;
         {warning, Reason} ->

apps/event_sourcing_core/src/event_sourcing_core_store.erl

@@ -17,11 +17,11 @@ Both modules may be the same if one backend implements both roles.
 -export([
     start/1,
     stop/1,
-    persist_events/3,
-    retrieve_and_fold_events/5,
+    append/3,
+    fold/5,
     retrieve_events/3,
-    save_snapshot/2,
-    retrieve_latest_snapshot/2,
+    store/2,
+    load_latest/2,
     new_snapshot/5,
     snapshot_stream_id/1,
     snapshot_domain/1,
@@ -82,13 +82,17 @@ stop({EventModule, SnapshotModule}) ->
     end.
 
 -doc """
-Persists a list of events to the event store using the specified store module.
+Appends a list of events to the event store using the specified store module.
+
+This is the primary mechanism for persisting domain events. All events in the list
+must target the same stream and have unique identifiers. The store backend ensures
+atomic persistence and maintains sequence ordering.
 """.
--spec persist_events(StoreContext, StreamId, Events) -> ok when
+-spec append(StoreContext, StreamId, Events) -> ok when
     StoreContext :: store_context(),
     StreamId :: stream_id(),
     Events :: [event()].
-persist_events({EventModule, _}, StreamId, Events) when is_list(Events) ->
+append({EventModule, _}, StreamId, Events) when is_list(Events) ->
     %% Validate that all events target the same StreamId
     ok = lists:foreach(
         fun(Event) ->
@@ -111,12 +115,15 @@ persist_events({EventModule, _}, StreamId, Events) when is_list(Events) ->
             erlang:error(duplicate_event)
     end,
 
-    EventModule:persist_events(StreamId, Events).
+    EventModule:append(StreamId, Events).
 
 -doc """
-Retrieves and folds events from the event store using the specified persistence module.
+Folds events from the event store into an accumulator using the specified persistence module.
+
+This is the core operation for event replay and state reconstruction. The backend
+retrieves events matching the given criteria and applies the fold function in sequence order.
 """.
--spec retrieve_and_fold_events(StoreContext, StreamId, Options, Fun, Acc0) -> Acc1 when
+-spec fold(StoreContext, StreamId, Options, Fun, Acc0) -> Acc1 when
     StoreContext :: store_context(),
     StreamId :: stream_id(),
     Options :: fold_events_opts(),
@@ -125,19 +132,21 @@ Retrieves and folds events from the event store using the specified persistence
     Acc1 :: term(),
     AccIn :: term(),
     AccOut :: term().
-retrieve_and_fold_events({EventModule, _}, StreamId, Options, Fun, InitialResult) ->
-    EventModule:retrieve_and_fold_events(StreamId, Options, Fun, InitialResult).
+fold({EventModule, _}, StreamId, Options, Fun, InitialResult) ->
+    EventModule:fold(StreamId, Options, Fun, InitialResult).
 
 -doc """
 Retrieves events for a given stream using the specified store module and options.
+
+This is a convenience wrapper around fold/5 that collects all events into a list.
 """.
 -spec retrieve_events(StoreContext, StreamId, Options) -> Result when
     StoreContext :: store_context(),
     StreamId :: stream_id(),
     Options :: fold_events_opts(),
     Result :: [event()].
 retrieve_events(StoreContext, StreamId, Options) ->
-    retrieve_and_fold_events(
+    fold(
         StoreContext,
         StreamId,
         Options,
@@ -304,36 +313,39 @@ new_snapshot(Domain, StreamId, Sequence, Timestamp, State) ->
     }.
 
 -doc """
-Saves a snapshot using the specified store module.
+Stores a snapshot using the specified store module.
 
-This function delegates snapshot saving to the store module implementation.
-The snapshot captures the aggregate state at a specific point in time, allowing
-for faster aggregate rehydration by avoiding full event replay.
+This function delegates snapshot storage to the backend implementation. The snapshot
+captures aggregate state at a specific sequence number, enabling faster rehydration
+by avoiding full event replay from the stream's beginning.
 
 The snapshot record contains all necessary fields (domain, stream_id, sequence,
-timestamp, state), making the API consistent with event persistence where events
-are passed as complete records.
+timestamp, state), consistent with event persistence where complete records are
+passed rather than individual fields.
 
-Returns `ok` on success, or `{warning, Reason}` if persistence fails.
+Returns `ok` on success, or `{warning, Reason}` if persistence fails. Warnings are
+preferred over exceptions since snapshots are optimizations, not requirements.
 """.
--spec save_snapshot(StoreContext, Snapshot) -> ok | {warning, Reason} when
+-spec store(StoreContext, Snapshot) -> ok | {warning, Reason} when
     StoreContext :: store_context(),
     Snapshot :: snapshot(),
     Reason :: term().
-save_snapshot({_, SnapshotModule}, Snapshot) ->
-    SnapshotModule:save_snapshot(Snapshot).
+store({_, SnapshotModule}, Snapshot) ->
+    SnapshotModule:store(Snapshot).
 
 -doc """
-Retrieves the latest snapshot for a stream using the specified store module.
+Loads the latest snapshot for a stream using the specified store module.
 
-Returns `{ok, Snapshot}` if found, `{error, not_found}` otherwise.
+Returns `{ok, Snapshot}` if found, `{error, not_found}` otherwise. This enables
+fast aggregate rehydration by restoring state from the snapshot and replaying only
+events that occurred after the snapshot's sequence number.
 """.
--spec retrieve_latest_snapshot(StoreContext, StreamId) -> {ok, Snapshot} | {error, not_found} when
+-spec load_latest(StoreContext, StreamId) -> {ok, Snapshot} | {error, not_found} when
     StoreContext :: store_context(),
     StreamId :: stream_id(),
     Snapshot :: snapshot().
-retrieve_latest_snapshot({_, SnapshotModule}, StreamId) ->
-    SnapshotModule:retrieve_latest_snapshot(StreamId).
+load_latest({_, SnapshotModule}, StreamId) ->
+    SnapshotModule:load_latest(StreamId).
 
 -doc """
 Returns the stream identifier of the snapshot.

apps/event_sourcing_core/test/event_sourcing_core_aggregate_tests.erl

@@ -116,7 +116,7 @@ aggregate_snapshot_creation() ->
     ?assertState(Pid, Id, #{balance := 125}, 3),
 
     %% Snapshot should be saved at sequence 3 (3 % 3 == 0)
-    {ok, Snapshot} = event_sourcing_core_store:retrieve_latest_snapshot(
+    {ok, Snapshot} = event_sourcing_core_store:load_latest(
         ?ETS_STORE_CONTEXT,
         Id
     ),
@@ -130,7 +130,7 @@ aggregate_snapshot_creation() ->
     ?assertState(Pid, Id, #{balance := 250}, 6),
 
     %% Snapshot should now be at sequence 6 (6 % 3 == 0)
-    {ok, Snapshot2} = event_sourcing_core_store:retrieve_latest_snapshot(
+    {ok, Snapshot2} = event_sourcing_core_store:load_latest(
         ?ETS_STORE_CONTEXT,
         Id
     ),
@@ -156,7 +156,7 @@ aggregate_snapshot_rehydration() ->
     ?assertState(Pid1, Id, #{balance := 300}, 2),
 
     %% Snapshot should exist at sequence 2
-    {ok, _Snapshot} = event_sourcing_core_store:retrieve_latest_snapshot(
+    {ok, _Snapshot} = event_sourcing_core_store:load_latest(
         ?ETS_STORE_CONTEXT,
         Id
     ),

apps/event_sourcing_core/test/event_sourcing_core_store_snapshot_stub.erl

@@ -2,7 +2,7 @@
 
 -include_lib("event_sourcing_contract/include/event_sourcing.hrl").
 
--export([start/0, stop/0, save_snapshot/1, retrieve_latest_snapshot/1]).
+-export([start/0, stop/0, store/1, load_latest/1]).
 
 -define(TABLE, ?MODULE).
 
@@ -26,14 +26,14 @@ stop() ->
             ok
     end.
 
--spec save_snapshot(snapshot()) -> ok.
-save_snapshot(Snapshot) ->
+-spec store(snapshot()) -> ok.
+store(Snapshot) ->
     StreamId = event_sourcing_core_store:snapshot_stream_id(Snapshot),
     ets:insert(?TABLE, {StreamId, Snapshot}),
     ok.
 
--spec retrieve_latest_snapshot(stream_id()) -> {ok, snapshot()} | {error, not_found}.
-retrieve_latest_snapshot(StreamId) ->
+-spec load_latest(stream_id()) -> {ok, snapshot()} | {error, not_found}.
+load_latest(StreamId) ->
     case ets:lookup(?TABLE, StreamId) of
         [{_, Snapshot}] ->
             {ok, Snapshot};