Skip to content

Commit 7d3e0c1

Browse files
ccamelccamel
authored
Merge pull request #38 from ccamel/refactor/fold-interval-parameter
Refactor fold callback to use interval as first-class parameter
2 parents 3f914a8 + 012d846 commit 7d3e0c1

12 files changed

Lines changed: 351 additions & 108 deletions

‎README.md‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -126,11 +126,14 @@ The event store is a core component in this experiment, designed as a customizab
126126
Events :: [event()].
127127

128128
% Folds events from a stream using a provided function
129-
-callback fold(StreamId, FoldFun, InitialAcc, Options) -> {ok, Acc} | {error, term()}
129+
-callback fold(StreamId, FoldFun, InitialAcc, Interval) -> Acc1
130130
when StreamId :: stream_id(),
131-
FoldFun :: fold_events_fun(),
132-
InitialAcc :: Acc,
133-
Options :: fold_events_opts().
131+
FoldFun :: fun((Event :: event(), AccIn) -> AccOut),
132+
InitialAcc :: term(),
133+
Interval :: event_sourcing_interval:interval(),
134+
Acc1 :: term(),
135+
AccIn :: term(),
136+
AccOut :: term().
134137
```
135138

136139
#### Snapshot Support
@@ -298,7 +301,7 @@ The manager is responsible for:
298301
The aggregate manager maintains a mapping of stream IDs to aggregate process PIDs. When a command is received:
299302

300303
1. The `Router` module extracts the target aggregate type and stream ID from the command.
301-
2. If the aggregate type matches the manager’s configured `Aggregate` module:
304+
2. If the aggregate type matches the manager's configured `Aggregate` module:
302305
- The manager checks its internal `pids` map for an existing process for the stream ID.
303306
- If none exists, it spawns a new `event_sourcing_core_aggregate` process using the provided Aggregate, Store, and stream ID, then monitors it.
304307
- The command is forwarded to the aggregate process via `event_sourcing_core_aggregate:dispatch/2`.

‎apps/event_sourcing_contract/include/event_sourcing.hrl‎

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -77,11 +77,3 @@ Fields:
7777
- `state`: The serialized aggregate state at the snapshot point.
7878
""".
7979
-type snapshot() :: #snapshot{}.
80-
81-
%%% Options for folding events
82-
-type fold_events_opts() ::
83-
#{
84-
from => non_neg_integer(),
85-
to => non_neg_integer() | infinity,
86-
limit => pos_integer() | infinity
87-
}.

‎apps/event_sourcing_contract/rebar.config‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
{erl_opts, [debug_info]}.
2+
3+
{deps, []}.
4+
5+
{eunit_opts, [verbose]}.

‎apps/event_sourcing_contract/src/event_sourcing_event_store_behaviour.erl‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -63,25 +63,25 @@ stream ID is incorrect, duplicate events if the sequence number is not unique).
6363
-doc """
6464
Retrieves events from a stream and folds them into an accumulator.
6565

66-
This callback fetches events for the given `StreamId`, applies the `FoldFun` to each
67-
event in sequence order, and returns the final accumulator. It’s typically used to
68-
rebuild application state by replaying events.
66+
This callback fetches events for the given `StreamId` within the specified `Interval`,
67+
applies the `FoldFun` to each event in sequence order, and returns the final accumulator.
68+
It's typically used to rebuild application state by replaying events.
6969

7070
- StreamId is an atom identifying the event stream (e.g., order-123).
7171
- FoldFun is a function `fun((Event, AccIn) -> AccOut)` to process each event.
7272
- InitialAcc is the initial accumulator value (e.g., an empty state).
73-
- Options is A list of filters:
74-
- `{from, Sequence}`: Start at this sequence (default: 0).
75-
- `{to, Sequence | infinity}`: End at this sequence (default: infinity).
76-
- `{limit, Limit}`: Maximum number of events to retrieve (default: infinity).
73+
- Interval is a sequence interval defining which events to retrieve:
74+
- Use `event_sourcing_interval:new(0, infinity)` to replay all events.
75+
- Use `event_sourcing_interval:new(N, infinity)` to replay from checkpoint N.
76+
- Use `event_sourcing_interval:new(M, N)` to replay a specific bounded range.
7777

78-
Returns the final accumulator after folding all events.
78+
Returns the final accumulator after folding all events in the interval.
7979
""".
80-
-callback fold(StreamId, Fun, Acc0, Options) -> Acc1 when
80+
-callback fold(StreamId, Fun, Acc0, Interval) -> Acc1 when
8181
StreamId :: stream_id(),
8282
Fun :: fun((Event :: event(), AccIn) -> AccOut),
8383
Acc0 :: term(),
84-
Options :: fold_events_opts(),
84+
Interval :: event_sourcing_interval:interval(),
8585
Acc1 :: term(),
8686
AccIn :: term(),
8787
AccOut :: term().

‎apps/event_sourcing_contract/src/event_sourcing_interval.erl‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
-module(event_sourcing_interval).
2+
-moduledoc """
3+
Algebraic data type for representing sequence intervals with lower and upper bounds.
4+
Supports both bounded intervals `[From, To)` and unbounded intervals `[From, +∞)`.
5+
6+
Intervals are half-open: they include the lower bound but exclude the upper bound.
7+
""".
8+
9+
-export([
10+
new/2,
11+
is_empty/1,
12+
advance/2,
13+
lower_bound/1,
14+
upper_bound/1
15+
]).
16+
17+
-export_type([interval/0]).
18+
19+
-type interval() :: {non_neg_integer(), non_neg_integer() | infinity}.
20+
21+
-doc """
22+
Create a new interval with the given lower and upper bounds.
23+
Upper bound can be `infinity` for unbounded intervals.
24+
25+
The interval is half-open: [From, To) includes From but excludes To.
26+
""".
27+
-spec new(non_neg_integer(), non_neg_integer() | infinity) -> interval().
28+
new(From, To) when
29+
From >= 0,
30+
(is_integer(To) andalso To >= From) orelse To =:= infinity
31+
->
32+
{From, To}.
33+
34+
-doc """
35+
Check if an interval is empty.
36+
An interval [From, To) is empty when From >= To.
37+
Unbounded intervals (with upper bound = infinity) are never empty.
38+
""".
39+
-spec is_empty(interval()) -> boolean().
40+
is_empty({From, infinity}) when From >= 0 ->
41+
false;
42+
is_empty({From, To}) when is_integer(From), is_integer(To) ->
43+
From >= To.
44+
45+
-doc """
46+
Advance the interval by moving the lower bound forward by the given amount.
47+
Returns the new interval.
48+
""".
49+
-spec advance(interval(), non_neg_integer()) -> interval().
50+
advance({From, To}, N) when N >= 0 ->
51+
{From + N, To}.
52+
53+
-doc """
54+
Get the lower bound of the interval.
55+
""".
56+
-spec lower_bound(interval()) -> non_neg_integer().
57+
lower_bound({From, _To}) ->
58+
From.
59+
60+
-doc """
61+
Get the upper bound of the interval.
62+
""".
63+
-spec upper_bound(interval()) -> non_neg_integer() | infinity.
64+
upper_bound({_From, To}) ->
65+
To.

‎apps/event_sourcing_contract/test/event_sourcing_interval_tests.erl‎

Lines changed: 152 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,152 @@
1+
-module(event_sourcing_interval_tests).
2+
3+
-include_lib("eunit/include/eunit.hrl").
4+
5+
%%% Test suite
6+
7+
suite_test_() ->
8+
{foreach, fun setup/0, fun teardown/1, [
9+
{"new_interval_creation", fun new_interval_creation/0},
10+
{"new_interval_validation", fun new_interval_validation/0},
11+
{"is_empty_check", fun is_empty_check/0},
12+
{"advance_interval", fun advance_interval/0},
13+
{"lower_bound_access", fun lower_bound_access/0},
14+
{"upper_bound_access", fun upper_bound_access/0},
15+
{"interval_properties", fun interval_properties/0},
16+
{"edge_cases", fun edge_cases/0}
17+
]}.
18+
19+
setup() ->
20+
ok.
21+
22+
teardown(_) ->
23+
ok.
24+
25+
%%% Test cases
26+
27+
new_interval_creation() ->
28+
% Test basic interval creation [0, 10)
29+
Interval1 = event_sourcing_interval:new(0, 10),
30+
?assertEqual({0, 10}, Interval1),
31+
32+
% Test unbounded interval [5, +∞)
33+
Interval2 = event_sourcing_interval:new(5, infinity),
34+
?assertEqual({5, infinity}, Interval2),
35+
36+
% Test non-empty single-element interval [7, 8)
37+
Interval3 = event_sourcing_interval:new(7, 8),
38+
?assertEqual({7, 8}, Interval3).
39+
40+
new_interval_validation() ->
41+
% Test valid cases
42+
?assertEqual({1, 5}, event_sourcing_interval:new(1, 5)),
43+
?assertEqual({10, infinity}, event_sourcing_interval:new(10, infinity)),
44+
?assertEqual({5, 5}, event_sourcing_interval:new(5, 5)),
45+
46+
% Test invalid cases - should throw function_clause error due to guard failure
47+
?assertError(function_clause, event_sourcing_interval:new(-1, 5)),
48+
% From > To
49+
?assertError(function_clause, event_sourcing_interval:new(5, 3)),
50+
?assertError(function_clause, event_sourcing_interval:new(5, -1)).
51+
52+
is_empty_check() ->
53+
% Test non-empty intervals (half-open [From, To))
54+
55+
% [0, 10) has 10 elements
56+
?assertEqual(false, event_sourcing_interval:is_empty({0, 10})),
57+
% [5, 6) has 1 element
58+
?assertEqual(false, event_sourcing_interval:is_empty({5, 6})),
59+
?assertEqual(false, event_sourcing_interval:is_empty({1, infinity})),
60+
61+
% Test empty intervals (From >= To)
62+
63+
% [5, 5) is empty
64+
?assertEqual(true, event_sourcing_interval:is_empty({5, 5})),
65+
% [5, 3) is empty
66+
?assertEqual(true, event_sourcing_interval:is_empty({5, 3})),
67+
% [10, 9) is empty
68+
?assertEqual(true, event_sourcing_interval:is_empty({10, 9})),
69+
70+
% Unbounded intervals are never empty
71+
?assertEqual(false, event_sourcing_interval:is_empty({0, infinity})),
72+
?assertEqual(false, event_sourcing_interval:is_empty({100, infinity})).
73+
74+
advance_interval() ->
75+
% Test advancing bounded intervals
76+
Interval1 = event_sourcing_interval:new(0, 10),
77+
Advanced1 = event_sourcing_interval:advance(Interval1, 3),
78+
?assertEqual({3, 10}, Advanced1),
79+
80+
% Test advancing unbounded intervals
81+
Interval2 = event_sourcing_interval:new(5, infinity),
82+
Advanced2 = event_sourcing_interval:advance(Interval2, 7),
83+
?assertEqual({12, infinity}, Advanced2),
84+
85+
% Test advancing by zero
86+
Interval3 = event_sourcing_interval:new(2, 8),
87+
Advanced3 = event_sourcing_interval:advance(Interval3, 0),
88+
?assertEqual({2, 8}, Advanced3),
89+
90+
% Test advancing to boundary (creates empty interval)
91+
Interval4 = event_sourcing_interval:new(3, 7),
92+
Advanced4 = event_sourcing_interval:advance(Interval4, 4),
93+
?assertEqual({7, 7}, Advanced4),
94+
?assertEqual(true, event_sourcing_interval:is_empty(Advanced4)),
95+
96+
% Test advancing beyond boundary (creates empty interval)
97+
Interval5 = event_sourcing_interval:new(3, 7),
98+
Advanced5 = event_sourcing_interval:advance(Interval5, 5),
99+
?assertEqual({8, 7}, Advanced5),
100+
?assertEqual(true, event_sourcing_interval:is_empty(Advanced5)).
101+
102+
lower_bound_access() ->
103+
% Test lower bound extraction
104+
?assertEqual(0, event_sourcing_interval:lower_bound({0, 10})),
105+
?assertEqual(5, event_sourcing_interval:lower_bound({5, infinity})),
106+
?assertEqual(7, event_sourcing_interval:lower_bound({7, 8})),
107+
?assertEqual(100, event_sourcing_interval:lower_bound({100, 200})).
108+
109+
upper_bound_access() ->
110+
% Test upper bound extraction
111+
?assertEqual(10, event_sourcing_interval:upper_bound({0, 10})),
112+
?assertEqual(infinity, event_sourcing_interval:upper_bound({5, infinity})),
113+
?assertEqual(8, event_sourcing_interval:upper_bound({7, 8})),
114+
?assertEqual(200, event_sourcing_interval:upper_bound({100, 200})).
115+
116+
interval_properties() ->
117+
% Test that intervals maintain invariants for half-open intervals [From, To)
118+
Interval1 = event_sourcing_interval:new(0, 10),
119+
?assert(
120+
event_sourcing_interval:lower_bound(Interval1) <
121+
event_sourcing_interval:upper_bound(Interval1) orelse
122+
event_sourcing_interval:upper_bound(Interval1) =:= infinity
123+
),
124+
125+
Interval2 = event_sourcing_interval:new(5, infinity),
126+
?assert(event_sourcing_interval:lower_bound(Interval2) >= 0),
127+
?assert(event_sourcing_interval:upper_bound(Interval2) =:= infinity),
128+
129+
% Test that advancing preserves the upper bound
130+
Original = event_sourcing_interval:new(2, 15),
131+
Advanced = event_sourcing_interval:advance(Original, 3),
132+
?assertEqual(
133+
event_sourcing_interval:upper_bound(Original), event_sourcing_interval:upper_bound(Advanced)
134+
).
135+
136+
edge_cases() ->
137+
% Test single-element interval [0, 1)
138+
SingleElement = event_sourcing_interval:new(0, 1),
139+
?assertEqual(false, event_sourcing_interval:is_empty(SingleElement)),
140+
?assertEqual(0, event_sourcing_interval:lower_bound(SingleElement)),
141+
?assertEqual(1, event_sourcing_interval:upper_bound(SingleElement)),
142+
143+
% Test large numbers
144+
LargeInterval = event_sourcing_interval:new(1000000, 2000000),
145+
?assertEqual(false, event_sourcing_interval:is_empty(LargeInterval)),
146+
AdvancedLarge = event_sourcing_interval:advance(LargeInterval, 500000),
147+
?assertEqual({1500000, 2000000}, AdvancedLarge),
148+
149+
% Test advancing unbounded interval by large amount
150+
Unbounded = event_sourcing_interval:new(0, infinity),
151+
AdvancedUnbounded = event_sourcing_interval:advance(Unbounded, 1000000),
152+
?assertEqual({1000000, infinity}, AdvancedUnbounded).

‎apps/event_sourcing_core/src/event_sourcing_core_aggregate.erl‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -155,7 +155,7 @@ init({Aggregate, StoreContext, Id, Opts}) ->
155155
Id,
156156
FoldFun,
157157
{StateFromSnapshot, SequenceFromSnapshot},
158-
#{from => SequenceFromSnapshot + 1}
158+
event_sourcing_interval:new(SequenceFromSnapshot + 1, infinity)
159159
),
160160
Timeout = maps:get(timeout, Opts, ?INACTIVITY_TIMEOUT),
161161
SnapshotInterval = maps:get(snapshot_interval, Opts, 0),

‎apps/event_sourcing_core/src/event_sourcing_core_store.erl‎

Lines changed: 10 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,6 @@ Both modules may be the same if one backend implements both roles.
5454
snapshot/0,
5555
snapshot_id/0,
5656
snapshot_data/0,
57-
fold_events_opts/0,
5857
store_context/0
5958
]).
6059

@@ -121,37 +120,37 @@ append({EventModule, _}, StreamId, Events) when is_list(Events) ->
121120
Folds events from the event store into an accumulator using the specified persistence module.
122121

123122
This is the core operation for event replay and state reconstruction. The backend
124-
retrieves events matching the given criteria and applies the fold function in sequence order.
123+
retrieves events within the specified interval and applies the fold function in sequence order.
125124
""".
126-
-spec fold(StoreContext, StreamId, Fun, Acc0, Options) -> Acc1 when
125+
-spec fold(StoreContext, StreamId, Fun, Acc0, Interval) -> Acc1 when
127126
StoreContext :: store_context(),
128127
StreamId :: stream_id(),
129128
Fun :: fun((Event :: event(), AccIn) -> AccOut),
130129
Acc0 :: term(),
131-
Options :: fold_events_opts(),
130+
Interval :: event_sourcing_interval:interval(),
132131
Acc1 :: term(),
133132
AccIn :: term(),
134133
AccOut :: term().
135-
fold({EventModule, _}, StreamId, Fun, InitialResult, Options) ->
136-
EventModule:fold(StreamId, Fun, InitialResult, Options).
134+
fold({EventModule, _}, StreamId, Fun, InitialResult, Interval) ->
135+
EventModule:fold(StreamId, Fun, InitialResult, Interval).
137136

138137
-doc """
139-
Retrieves events for a given stream using the specified store module and options.
138+
Retrieves events for a given stream using the specified store module and interval.
140139

141140
This is a convenience wrapper around fold/5 that collects all events into a list.
142141
""".
143-
-spec retrieve_events(StoreContext, StreamId, Options) -> Result when
142+
-spec retrieve_events(StoreContext, StreamId, Interval) -> Result when
144143
StoreContext :: store_context(),
145144
StreamId :: stream_id(),
146-
Options :: fold_events_opts(),
145+
Interval :: event_sourcing_interval:interval(),
147146
Result :: [event()].
148-
retrieve_events(StoreContext, StreamId, Options) ->
147+
retrieve_events(StoreContext, StreamId, Interval) ->
149148
fold(
150149
StoreContext,
151150
StreamId,
152151
fun(Event, Acc) -> Acc ++ [Event] end,
153152
[],
154-
Options
153+
Interval
155154
).
156155

157156
-doc """

‎apps/event_sourcing_core/test/event_sourcing_core_aggregate_tests.erl‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -202,7 +202,9 @@ aggregate_custom_now_fun() ->
202202
?assertEqual(ok, event_sourcing_core_aggregate:dispatch(Pid, {bank, deposit, Id, 42})),
203203

204204
%% Retrieve persisted events and assert the timestamp matches the injected Now
205-
Events = event_sourcing_core_store:retrieve_events(?ETS_STORE_CONTEXT, Id, #{}),
205+
Events = event_sourcing_core_store:retrieve_events(
206+
?ETS_STORE_CONTEXT, Id, event_sourcing_interval:new(0, infinity)
207+
),
206208
?assertEqual(1, length(Events)),
207209
[Event] = Events,
208210
?assertEqual(Now, event_sourcing_core_store:timestamp(Event)).

0 commit comments

Comments
 (0)