Skip to content

Commit 61df63b

Browse files
ccamelccamel
authored
Merge pull request #59 from ccamel/feat/projection-runtime
✨ Feat/projection runtime
2 parents f73590c + 1c5e5db commit 61df63b

8 files changed

Lines changed: 582 additions & 0 deletions

‎apps/es_kernel/src/es_contract_projection_checkpoint_store.erl‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
-module(es_contract_projection_checkpoint_store).
2+
3+
-moduledoc """
4+
Behaviour for projection checkpoint storage.
5+
6+
Projection checkpoints track the last global event position processed by a
7+
projection runner. They are runtime progress metadata, not projection state.
8+
""".
9+
10+
-doc """
11+
Load the last processed global position for a projection.
12+
""".
13+
-callback load_checkpoint(ProjectionName) ->
14+
{ok, Position} | {error, not_found} | {error, Reason}
15+
when
16+
ProjectionName :: atom(),
17+
Position :: es_contract_event_store:position(),
18+
Reason :: term().
19+
20+
-doc """
21+
Store the last processed global position for a projection.
22+
""".
23+
-callback store_checkpoint(ProjectionName, Position) ->
24+
ok | {error, Reason}
25+
when
26+
ProjectionName :: atom(),
27+
Position :: es_contract_event_store:position(),
28+
Reason :: term().

‎apps/es_kernel/src/es_kernel.app.src‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,9 +14,13 @@
1414
es_contract_command,
1515
es_contract_event_store,
1616
es_contract_event,
17+
es_contract_projection,
18+
es_contract_projection_checkpoint_store,
1719
es_contract_range,
1820
es_contract_snapshot_store,
1921
es_contract_snapshot,
22+
es_projection_checkpoint_ets,
23+
es_kernel_projection,
2024
es_kernel_store,
2125
es_kernel_mgr_aggregate,
2226
es_kernel_aggregate,

‎apps/es_kernel/src/es_kernel_projection.erl‎

Lines changed: 297 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,297 @@
1+
-module(es_kernel_projection).
2+
3+
-moduledoc """
4+
Projection runtime built on the global event log.
5+
6+
The runner consumes events with `es_kernel_store:fold_all/4`, applies the
7+
projection callback module, and stores progress after each processed position.
8+
9+
The polling runner is fail-fast: any store, checkpoint, or projection handling
10+
error stops the process. Run it under a supervisor with an appropriate restart
11+
strategy when automatic recovery is desired.
12+
""".
13+
14+
-behaviour(gen_server).
15+
16+
-export([
17+
run_once/3,
18+
start_link/3,
19+
stop/1
20+
]).
21+
22+
-export([
23+
init/1,
24+
handle_call/3,
25+
handle_cast/2,
26+
handle_info/2,
27+
terminate/2,
28+
code_change/3
29+
]).
30+
31+
-define(DEFAULT_CHECKPOINT_STORE, es_projection_checkpoint_ets).
32+
-define(DEFAULT_START_POSITION, 0).
33+
-define(DEFAULT_POLL_INTERVAL, 200).
34+
35+
-record(state, {
36+
store_context :: es_kernel_store:store_context(),
37+
projection_module :: module(),
38+
projection_name :: atom(),
39+
checkpoint_store :: module(),
40+
projection_state :: es_contract_projection:projection_state(),
41+
next_position :: es_contract_event_store:position(),
42+
last_position :: es_contract_event_store:position() | undefined,
43+
poll_interval :: pos_integer()
44+
}).
45+
46+
-opaque state() :: #state{}.
47+
-export_type([options/0, state/0]).
48+
49+
-type options() :: #{
50+
checkpoint_store => module(),
51+
start_position => es_contract_event_store:position(),
52+
poll_interval => pos_integer(),
53+
name => atom() | undefined
54+
}.
55+
56+
-doc """
57+
Run a projection catch-up once, then return the resulting projection state.
58+
""".
59+
-spec run_once(StoreContext, ProjectionModule, Options) ->
60+
{ok, ProjectionState, LastPosition} | {error, Reason}
61+
when
62+
StoreContext :: es_kernel_store:store_context(),
63+
ProjectionModule :: module(),
64+
Options :: options(),
65+
ProjectionState :: es_contract_projection:projection_state(),
66+
LastPosition :: es_contract_event_store:position() | undefined,
67+
Reason :: term().
68+
run_once(StoreContext, ProjectionModule, Options) ->
69+
case init_runtime(StoreContext, ProjectionModule, Options) of
70+
{ok, Runtime} ->
71+
catch_up(Runtime);
72+
{error, Reason} ->
73+
{error, Reason}
74+
end.
75+
76+
-doc """
77+
Start a polling projection runner.
78+
""".
79+
-spec start_link(StoreContext, ProjectionModule, Options) -> gen_server:start_ret() when
80+
StoreContext :: es_kernel_store:store_context(),
81+
ProjectionModule :: module(),
82+
Options :: options().
83+
start_link(StoreContext, ProjectionModule, Options) ->
84+
Args = {StoreContext, ProjectionModule, Options},
85+
case maps:get(name, Options, undefined) of
86+
undefined ->
87+
gen_server:start_link(?MODULE, Args, []);
88+
Name when is_atom(Name) ->
89+
gen_server:start_link({local, Name}, ?MODULE, Args, [])
90+
end.
91+
92+
-doc """
93+
Stop a polling projection runner.
94+
""".
95+
-spec stop(gen_server:server_ref()) -> ok.
96+
stop(ServerRef) ->
97+
gen_server:stop(ServerRef).
98+
99+
-spec init({StoreContext, ProjectionModule, Options}) ->
100+
{ok, state()} | {stop, Reason}
101+
when
102+
StoreContext :: es_kernel_store:store_context(),
103+
ProjectionModule :: module(),
104+
Options :: options(),
105+
Reason :: term().
106+
init({StoreContext, ProjectionModule, Options}) ->
107+
case init_runtime(StoreContext, ProjectionModule, Options) of
108+
{ok, State} ->
109+
self() ! tick,
110+
{ok, State};
111+
{error, Reason} ->
112+
{stop, Reason}
113+
end.
114+
115+
-spec handle_call(term(), {pid(), term()}, state()) -> {reply, {error, unknown_call}, state()}.
116+
handle_call(_Request, _From, State) ->
117+
{reply, {error, unknown_call}, State}.
118+
119+
-spec handle_cast(term(), state()) -> {noreply, state()}.
120+
handle_cast(_Request, State) ->
121+
{noreply, State}.
122+
123+
-spec handle_info(term(), state()) -> {noreply, state()} | {stop, term(), state()}.
124+
handle_info(tick, State) ->
125+
case catch_up(State) of
126+
{ok, ProjectionState, LastPosition} ->
127+
NextPosition = next_position(LastPosition, State#state.next_position),
128+
erlang:send_after(State#state.poll_interval, self(), tick),
129+
{noreply, State#state{
130+
projection_state = ProjectionState,
131+
last_position = LastPosition,
132+
next_position = NextPosition
133+
}};
134+
{error, Reason} ->
135+
{stop, Reason, State}
136+
end;
137+
handle_info(_Info, State) ->
138+
{noreply, State}.
139+
140+
-spec terminate(term(), state()) -> ok.
141+
terminate(_Reason, _State) ->
142+
ok.
143+
144+
-spec code_change(term(), state(), term()) -> {ok, state()}.
145+
code_change(_OldVsn, State, _Extra) ->
146+
{ok, State}.
147+
148+
-spec init_runtime(es_kernel_store:store_context(), module(), options()) ->
149+
{ok, state()} | {error, term()}.
150+
init_runtime(StoreContext, ProjectionModule, Options) ->
151+
CheckpointStore = maps:get(checkpoint_store, Options, ?DEFAULT_CHECKPOINT_STORE),
152+
StartPosition = maps:get(start_position, Options, ?DEFAULT_START_POSITION),
153+
PollInterval = maps:get(poll_interval, Options, ?DEFAULT_POLL_INTERVAL),
154+
ProjectionName = ProjectionModule:name(),
155+
ProjectionState = ProjectionModule:init(),
156+
case ensure_checkpoint_store_started(CheckpointStore) of
157+
ok ->
158+
case load_next_position(CheckpointStore, ProjectionName, StartPosition) of
159+
{ok, NextPosition, LastPosition} ->
160+
{ok, #state{
161+
store_context = StoreContext,
162+
projection_module = ProjectionModule,
163+
projection_name = ProjectionName,
164+
checkpoint_store = CheckpointStore,
165+
projection_state = ProjectionState,
166+
next_position = NextPosition,
167+
last_position = LastPosition,
168+
poll_interval = PollInterval
169+
}};
170+
{error, Reason} ->
171+
{error, Reason}
172+
end;
173+
{error, Reason} ->
174+
{error, Reason}
175+
end.
176+
177+
-spec ensure_checkpoint_store_started(module()) -> ok | {error, term()}.
178+
ensure_checkpoint_store_started(CheckpointStore) ->
179+
case code:ensure_loaded(CheckpointStore) of
180+
{module, CheckpointStore} ->
181+
case erlang:function_exported(CheckpointStore, start, 0) of
182+
true ->
183+
CheckpointStore:start();
184+
false ->
185+
ok
186+
end;
187+
{error, Reason} ->
188+
{error, Reason}
189+
end.
190+
191+
-spec load_next_position(module(), atom(), es_contract_event_store:position()) ->
192+
{ok, NextPosition, LastPosition} | {error, term()}
193+
when
194+
NextPosition :: es_contract_event_store:position(),
195+
LastPosition :: es_contract_event_store:position() | undefined.
196+
load_next_position(CheckpointStore, ProjectionName, StartPosition) ->
197+
case CheckpointStore:load_checkpoint(ProjectionName) of
198+
{ok, LastPosition} ->
199+
{ok, max(StartPosition, LastPosition + 1), LastPosition};
200+
{error, not_found} ->
201+
{ok, StartPosition, undefined};
202+
{error, Reason} ->
203+
{error, Reason}
204+
end.
205+
206+
-spec catch_up(state()) -> {ok, ProjectionState, LastPosition} | {error, Reason} when
207+
ProjectionState :: es_contract_projection:projection_state(),
208+
LastPosition :: es_contract_event_store:position() | undefined,
209+
Reason :: term().
210+
catch_up(#state{next_position = NextPosition} = State) ->
211+
Range = es_contract_range:new(NextPosition, infinity),
212+
Acc0 = {ok, State#state.projection_state, State#state.last_position},
213+
FoldFun = fun(Event, Position, Acc) -> process_event(Event, Position, Acc, State) end,
214+
case es_kernel_store:fold_all(State#state.store_context, FoldFun, Acc0, Range) of
215+
{ok, {ok, ProjectionState, LastPosition}} ->
216+
{ok, ProjectionState, LastPosition};
217+
{ok, {error, Reason}} ->
218+
{error, Reason};
219+
{error, Reason} ->
220+
{error, Reason}
221+
end.
222+
223+
-spec process_event(
224+
es_contract_event:t(),
225+
es_contract_event_store:position(),
226+
{ok, ProjectionState, LastPosition} | {error, Reason},
227+
state()
228+
) ->
229+
{ok, NewProjectionState, NewLastPosition} | {error, Reason}
230+
when
231+
ProjectionState :: es_contract_projection:projection_state(),
232+
NewProjectionState :: es_contract_projection:projection_state(),
233+
LastPosition :: es_contract_event_store:position() | undefined,
234+
NewLastPosition :: es_contract_event_store:position(),
235+
Reason :: term().
236+
process_event(_Event, _Position, {error, Reason}, _State) ->
237+
{error, Reason};
238+
process_event(Event, Position, {ok, ProjectionState, _LastPosition}, State) ->
239+
case should_process(State#state.projection_module, Event) of
240+
true ->
241+
handle_projection_event(Event, Position, ProjectionState, State);
242+
false ->
243+
checkpoint(Position, ProjectionState, State)
244+
end.
245+
246+
-spec should_process(module(), es_contract_event:t()) -> boolean().
247+
should_process(ProjectionModule, Event) ->
248+
case erlang:function_exported(ProjectionModule, event_filter, 1) of
249+
true ->
250+
ProjectionModule:event_filter(Event);
251+
false ->
252+
true
253+
end.
254+
255+
-spec handle_projection_event(
256+
es_contract_event:t(),
257+
es_contract_event_store:position(),
258+
es_contract_projection:projection_state(),
259+
state()
260+
) ->
261+
{ok, es_contract_projection:projection_state(), es_contract_event_store:position()}
262+
| {error, term()}.
263+
handle_projection_event(Event, Position, ProjectionState, State) ->
264+
ProjectionModule = State#state.projection_module,
265+
case ProjectionModule:handle_event(Event, ProjectionState) of
266+
{ok, NewProjectionState} ->
267+
checkpoint(Position, NewProjectionState, State);
268+
{error, Reason} ->
269+
{error, {handle_event_failed, Position, Reason}}
270+
end.
271+
272+
-spec checkpoint(
273+
es_contract_event_store:position(),
274+
es_contract_projection:projection_state(),
275+
state()
276+
) ->
277+
{ok, es_contract_projection:projection_state(), es_contract_event_store:position()}
278+
| {error, term()}.
279+
checkpoint(Position, ProjectionState, State) ->
280+
CheckpointStore = State#state.checkpoint_store,
281+
ProjectionName = State#state.projection_name,
282+
case CheckpointStore:store_checkpoint(ProjectionName, Position) of
283+
ok ->
284+
{ok, ProjectionState, Position};
285+
{error, Reason} ->
286+
{error, {checkpoint_failed, Position, Reason}}
287+
end.
288+
289+
-spec next_position(
290+
es_contract_event_store:position() | undefined,
291+
es_contract_event_store:position()
292+
) ->
293+
es_contract_event_store:position().
294+
next_position(undefined, CurrentNextPosition) ->
295+
CurrentNextPosition;
296+
next_position(LastPosition, _CurrentNextPosition) ->
297+
LastPosition + 1.

0 commit comments

Comments
 (0)