refactor(store): clarify meaning of store context

Author: ccamel

apps/event_sourcing_core/src/event_sourcing_core_aggregate.erl

@@ -1,6 +1,6 @@
 -module(event_sourcing_core_aggregate).
 
--include_lib("event_sourcing_core.hrl").
+-include_lib("event_sourcing_core/include/event_sourcing_core.hrl").
 -include_lib("kernel/include/logger.hrl").
 
 -behaviour(gen_server).
@@ -34,15 +34,15 @@
 Starts an aggregate process with a given timeout.
 
 - Aggregate is the aggregate module implementing the behavior.
-- Store is a `{EventStore, SnapshotStore}` tuple.
+- StoreContext is a `{EventStore, SnapshotStore}` tuple.
 - Id is the unique identifier for the aggregate instance.
 - Timeout is the inactivity timeout in milliseconds before passivation.
 
 Function returns `{ok, Pid}` if successful, `{error, Reason}` otherwise.
 """.
--spec start_link(Aggregate, Store, Id, Opts) -> gen_server:start_ret() when
+-spec start_link(Aggregate, StoreContext, Id, Opts) -> gen_server:start_ret() when
     Aggregate :: module(),
-    Store :: event_sourcing_core_store:store(),
+    StoreContext :: event_sourcing_core_store:store_context(),
     Id :: stream_id(),
     Opts ::
         #{
@@ -52,22 +52,24 @@ Function returns `{ok, Pid}` if successful, `{error, Reason}` otherwise.
             now_fun => fun(() -> timestamp()),
             snapshot_interval => non_neg_integer()
         }.
-start_link(Aggregate, Store, Id, Opts) ->
-    gen_server:start_link(?MODULE, {Aggregate, Store, Id, Opts}, []).
+start_link(Aggregate, StoreContext, Id, Opts) ->
+    gen_server:start_link(?MODULE, {Aggregate, StoreContext, Id, Opts}, []).
 
 -doc """
 Starts a new aggregate process.
 
 - Aggregate is the aggregate module to start.
-- Store must be provided as `{EventStore, SnapshotStore}`.
+- StoreContext must be provided as `{EventStore, SnapshotStore}`.
 - Id is the unique identifier for the aggregate instance.
 """.
 -spec start_link(
-    Aggregate :: module(), Store :: event_sourcing_core_store:store(), Id :: stream_id()
+    Aggregate :: module(),
+    StoreContext :: event_sourcing_core_store:store_context(),
+    Id :: stream_id()
 ) ->
     gen_server:start_ret().
-start_link(Aggregate, Store, Id) ->
-    start_link(Aggregate, Store, Id, #{}).
+start_link(Aggregate, StoreContext, Id) ->
+    start_link(Aggregate, StoreContext, Id, #{}).
 
 -spec dispatch(Pid, Command) -> {ok, Result} | {error, Reason} when
     Pid :: pid(),
@@ -79,7 +81,7 @@ dispatch(Pid, Command) ->
 
 -record(state, {
     aggregate :: module(),
-    store :: event_sourcing_core_store:store(),
+    store :: event_sourcing_core_store:store_context(),
     id :: stream_id(),
     state :: aggregate_state(),
     sequence = ?SEQUENCE_ZERO :: non_neg_integer(),
@@ -100,7 +102,7 @@ Retrieves all events for the aggregate from the persistence layer and applies th
 sequentially to rehydrate the aggregate's state.
 
 - Aggregate is the aggregate module implementing the domain logic.
-- Store is the persistence module (store) implementing event retrieval.
+- StoreContext is a `{EventStore, SnapshotStore}` tuple used for event and snapshot persistence.
 - Id is the unique identifier for the aggregate.
 - Opts is a map of options including:
   - `timeout`: Inactivity timeout in milliseconds.
@@ -112,7 +114,7 @@ sequentially to rehydrate the aggregate's state.
 Function returns {ok, state()} on success, and returns {stop, Reason} on failure.
 """.
 -spec init(
-    {module(), event_sourcing_core_store:store(), stream_id(), #{
+    {module(), event_sourcing_core_store:store_context(), stream_id(), #{
         timeout => timeout(),
         sequence_zero => fun(() -> sequence()),
         sequence_next => fun((sequence()) -> sequence()),
@@ -121,14 +123,14 @@ Function returns {ok, state()} on success, and returns {stop, Reason} on failure
     }}
 ) ->
     {ok, state()}.
-init({Aggregate, Store, Id, Opts}) ->
+init({Aggregate, StoreContext, Id, Opts}) ->
     State0 = Aggregate:init(),
     SequenceZero = maps:get(sequence_zero, Opts, fun() -> ?SEQUENCE_ZERO end),
     SequenceNext = maps:get(sequence_next, Opts, fun(Sequence) -> Sequence + 1 end),
 
     %% Try to load the latest snapshot
     {StateFromSnapshot, SequenceFromSnapshot} =
-        case event_sourcing_core_store:retrieve_latest_snapshot(Store, Id) of
+        case event_sourcing_core_store:retrieve_latest_snapshot(StoreContext, Id) of
             {ok, Snapshot} ->
                 SnapshotState = event_sourcing_core_store:snapshot_state(Snapshot),
                 SnapshotSeq = event_sourcing_core_store:snapshot_sequence(Snapshot),
@@ -149,7 +151,7 @@ init({Aggregate, Store, Id, Opts}) ->
         end,
     {State1, Sequence1} =
         event_sourcing_core_store:retrieve_and_fold_events(
-            Store,
+            StoreContext,
             Id,
             #{from => SequenceFromSnapshot + 1},
             FoldFun,
@@ -160,7 +162,7 @@ init({Aggregate, Store, Id, Opts}) ->
     TimerRef = install_passivation(Timeout, undefined),
     {ok, #state{
         aggregate = Aggregate,
-        store = Store,
+        store = StoreContext,
         id = Id,
         state = State1,
         sequence = Sequence1,
@@ -269,7 +271,7 @@ Function returns the new state and sequence of the aggregate after the command i
 process_command(
     #state{
         aggregate = Aggregate,
-        store = Store,
+        store = StoreContext,
         id = Id,
         state = State0,
         sequence = Sequence0,
@@ -284,7 +286,7 @@ process_command(
             {ok, {State0, Sequence0}};
         {ok, PayloadEvents} when is_list(PayloadEvents) ->
             ok = persist_events(
-                PayloadEvents, {Aggregate, Store, Id, Sequence0, SequenceNext, NowFun}
+                PayloadEvents, {Aggregate, StoreContext, Id, Sequence0, SequenceNext, NowFun}
             ),
             {State1, Sequence1} =
                 apply_events(PayloadEvents, {Aggregate, State0, Sequence0, SequenceNext}),
@@ -318,14 +320,14 @@ apply_events(PayloadEvents, {Aggregate, State0, Sequence0, SequenceNext}) ->
     PayloadEvents :: [event_payload()],
     {
         Aggregate :: module(),
-        Store :: event_sourcing_core_store:store(),
+        StoreContext :: event_sourcing_core_store:store_context(),
         Id :: stream_id(),
         Sequence0 :: sequence(),
         SequenceNext :: fun((sequence()) -> sequence()),
         NowFun :: fun(() -> timestamp())
     }
 ) -> ok.
-persist_events(PayloadEvents, {Aggregate, Store, Id, Sequence0, SequenceNext, NowFun}) ->
+persist_events(PayloadEvents, {Aggregate, StoreContext, Id, Sequence0, SequenceNext, NowFun}) ->
     {Events, _} =
         lists:foldl(
             fun(PayloadEvent, {Events, SequenceN}) ->
@@ -352,7 +354,7 @@ persist_events(PayloadEvents, {Aggregate, Store, Id, Sequence0, SequenceNext, No
         end,
         Events
     ),
-    event_sourcing_core_store:persist_events(Store, Id, Events).
+    event_sourcing_core_store:persist_events(StoreContext, Id, Events).
 
 -doc """
 Saves a snapshot if the snapshot interval is configured and the current
@@ -367,7 +369,7 @@ maybe_save_snapshot(
     #state{
         snapshot_interval = Interval,
         sequence = Sequence,
-        store = Store,
+        store = StoreContext,
         aggregate = Aggregate,
         id = Id,
         state = AggState,
@@ -377,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(Store, Snapshot) of
+    case event_sourcing_core_store:save_snapshot(StoreContext, Snapshot) of
         ok ->
             ok;
         {warning, Reason} ->

apps/event_sourcing_core/src/event_sourcing_core_event_store.erl

@@ -20,7 +20,7 @@ Implementations must guarantee:
 Typical implementations include in-memory stores (ETS) and relational databases.
 """.
 
--include_lib("event_sourcing_core.hrl").
+-include_lib("event_sourcing_core/include/event_sourcing_core.hrl").
 
 -doc """
 Starts the event store, performing any necessary initialization.

apps/event_sourcing_core/src/event_sourcing_core_mgr_aggregate.erl

@@ -10,7 +10,7 @@ monitoring them for crashes.
 
 -behaviour(gen_server).
 
--include_lib("event_sourcing_core.hrl").
+-include_lib("event_sourcing_core/include/event_sourcing_core.hrl").
 
 -export([
     init/1,
@@ -28,7 +28,7 @@ monitoring them for crashes.
 
 -record(state, {
     aggregate :: module(),
-    store :: event_sourcing_core_store:store(),
+    store :: event_sourcing_core_store:store_context(),
     router :: module(),
     opts ::
         #{
@@ -46,7 +46,7 @@ monitoring them for crashes.
 Starts the aggregate manager with custom options.
 
 - Aggregate is he module implementing the aggregate logic.
-- Store is a `{EventStore, SnapshotStore}` tuple.
+- StoreContext is a `{EventStore, SnapshotStore}` tuple.
 - Router is the module extracting routing info from commands.
 - Opts is the configuration options:
   - `timeout`: Timeout for operations (default: `infinity`).
@@ -56,9 +56,9 @@ Starts the aggregate manager with custom options.
 
 Function returns `{ok, Pid}` on success, or an error tuple if the server fails to start.
 """.
--spec start_link(Aggregate, Store, Router, Opts) -> gen_server:start_ret() when
+-spec start_link(Aggregate, StoreContext, Router, Opts) -> gen_server:start_ret() when
     Aggregate :: module(),
-    Store :: event_sourcing_core_store:store(),
+    StoreContext :: event_sourcing_core_store:store_context(),
     Router :: module(),
     Opts ::
         #{
@@ -67,25 +67,25 @@ Function returns `{ok, Pid}` on success, or an error tuple if the server fails t
             sequence_next => fun((sequence()) -> sequence()),
             now_fun => fun(() -> timestamp())
         }.
-start_link(Aggregate, Store, Router, Opts) ->
-    gen_server:start_link({local, ?MODULE}, ?MODULE, {Aggregate, Store, Router, Opts}, []).
+start_link(Aggregate, StoreContext, Router, Opts) ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, {Aggregate, StoreContext, Router, Opts}, []).
 
 -doc """
 Starts the aggregate manager with the given aggregate, store, and router modules,
 using default options.
 
 - Aggregate is the module implementing the aggregate logic.
-- Store follows the same `{EventStore, SnapshotStore}` convention.
+- StoreContext follows the same `{EventStore, SnapshotStore}` convention.
 - Router is the module extracting routing info from commands.
 
 Function returns `{ok, Pid}` on success, or an error tuple if the server fails to start.
 """.
--spec start_link(Aggregate, Store, Router) -> gen_server:start_ret() when
+-spec start_link(Aggregate, StoreContext, Router) -> gen_server:start_ret() when
     Aggregate :: module(),
-    Store :: event_sourcing_core_store:store(),
+    StoreContext :: event_sourcing_core_store:store_context(),
     Router :: module().
-start_link(Aggregate, Store, Router) ->
-    start_link(Aggregate, Store, Router, #{}).
+start_link(Aggregate, StoreContext, Router) ->
+    start_link(Aggregate, StoreContext, Router, #{}).
 
 -doc """
 Stops the aggregate manager.
@@ -124,9 +124,9 @@ Initializes the aggregate manager state.
 
 Function returns `{ok, State}` with an initialized state record.
 """.
--spec init({Aggregate, Store, Router, Opts}) -> {ok, State} when
+-spec init({Aggregate, StoreContext, Router, Opts}) -> {ok, State} when
     Aggregate :: module(),
-    Store :: event_sourcing_core_store:store(),
+    StoreContext :: event_sourcing_core_store:store_context(),
     Router :: module(),
     Opts ::
         #{
@@ -136,10 +136,10 @@ Function returns `{ok, State}` with an initialized state record.
             now_fun => fun(() -> timestamp())
         },
     State :: state().
-init({Aggregate, Store, Router, Opts}) ->
+init({Aggregate, StoreContext, Router, Opts}) ->
     {ok, #state{
         aggregate = Aggregate,
-        store = Store,
+        store = StoreContext,
         router = Router,
         opts = Opts,
         pids = #{}
@@ -217,15 +217,15 @@ ensure_and_dispatch(
     Id,
     Command,
     #state{
-        store = Store,
+        store = StoreContext,
         pids = Pids,
         opts = Opts
     } =
         State
 ) ->
     case maps:get(Id, Pids, undefined) of
         undefined ->
-            case start_aggregate(Aggregate, Store, Id, Opts) of
+            case start_aggregate(Aggregate, StoreContext, Id, Opts) of
                 {ok, Pid} ->
                     Result = forward(Pid, Command),
                     NewPids = maps:put(Id, Pid, Pids),
@@ -258,9 +258,9 @@ Monitors the new process and returns its pid.
 
 Function returns `{ok, Pid}` on success, or `{error, Reason}` on failure.
 """.
--spec start_aggregate(Aggregate, Store, Id, Opts) -> {ok, Result} | {error, Reason} when
+-spec start_aggregate(Aggregate, StoreContext, Id, Opts) -> {ok, Result} | {error, Reason} when
     Aggregate :: module(),
-    Store :: event_sourcing_core_store:store(),
+    StoreContext :: event_sourcing_core_store:store_context(),
     Id :: stream_id(),
     Opts ::
         #{
@@ -271,8 +271,8 @@ Function returns `{ok, Pid}` on success, or `{error, Reason}` on failure.
         },
     Result :: pid(),
     Reason :: term().
-start_aggregate(Aggregate, Store, Id, Opts) ->
-    case event_sourcing_core_aggregate:start_link(Aggregate, Store, Id, Opts) of
+start_aggregate(Aggregate, StoreContext, Id, Opts) ->
+    case event_sourcing_core_aggregate:start_link(Aggregate, StoreContext, Id, Opts) of
         {ok, Pid} ->
             erlang:monitor(process, Pid),
             {ok, Pid};

apps/event_sourcing_core/src/event_sourcing_core_mgr_behaviour.erl

@@ -9,7 +9,7 @@ for commands.
 Implementations of this behaviour are responsible for extracting routing information,
 enabling the aggregate manager to dispatch commands to the correct aggregate instance.
 """.
--include_lib("event_sourcing_core.hrl").
+-include_lib("event_sourcing_core/include/event_sourcing_core.hrl").
 
 -doc """
 Extracts the routing information from the command.

apps/event_sourcing_core/src/event_sourcing_core_snapshot_store.erl

@@ -20,7 +20,7 @@ Common implementations include key–value stores, databases, or object storage
 systems (e.g., S3).
 """.
 
--include_lib("event_sourcing_core.hrl").
+-include_lib("event_sourcing_core/include/event_sourcing_core.hrl").
 
 -doc """
 Save a snapshot for a stream.