evoq_projection behaviour (evoq v1.30.1)
View SourceProjection behavior for evoq.
Projections transform events into read model updates. They subscribe to event types and maintain query-optimized views.
Design Principles
- Projections do all calculations (events -> read model) - Read models are simple key-value lookups (no joins) - Projections are idempotent (can be replayed safely) - Checkpoints track progress for resume after restart
Callbacks
Required: - interested_in() -> [binary()] Event types this projection handles
- init(Config) -> {ok, State, ReadModel} Initialize projection with read model
- project(Event, Metadata, State, ReadModel) -> {ok, NewState, NewReadModel} | {skip, NewState, NewReadModel} | {error, Reason} Transform event into read model updates. skip says the projection does not want this event; the checkpoint still advances and the event is not retried.
Optional: - on_error(Error, Event, FailureContext, State) -> error_action() Handle projection errors: retry, {retry, DelayMs}, skip, stop or {dead_letter, Reason}, exactly as for an event handler. Without it the shared default applies (backoff, then dead letter).
- max_retries() -> pos_integer() - backoff_ms(Attempt) -> pos_integer() Tune the default retry policy.
Example
-module(order_summary_projection).
-behaviour(evoq_projection).
interested_in() -> [<<"OrderPlaced">>, <<"OrderShipped">>].
init(_Config) ->
{ok, RM} = evoq_read_model:new(evoq_read_model_ets, #{}),
{ok, #{}, RM}.
project(#{event_type := <<"OrderPlaced">>, data := Data}, Meta, State, RM) ->
OrderId = maps:get(order_id, Data),
Summary = #{status => placed, items => maps:get(items, Data, [])},
{ok, RM2} = evoq_read_model:put(OrderId, Summary, RM),
{ok, State, RM2}.Summary
Functions
Get the current checkpoint position: the highest position projected so far (the store's global position for events from the store subscription or a rebuild), or -1 when nothing has been projected yet.
Get event types this projection handles.
Get the read model instance.
Notify projection of an event.
Rebuild the projection from scratch. Clears the read model and replays all events.
Rebuild with options.
Start a projection.
Start a projection with options. Options: - store_id: Event store to replay from (overrides app env) - checkpoint_store: Module for persistent checkpoint storage
Callbacks
-callback backoff_ms(AttemptNumber :: pos_integer()) -> pos_integer().
-callback init(Config :: map()) -> {ok, State :: term(), ReadModel :: evoq_read_model:read_model()} | {error, Reason :: term()}.
-callback interested_in() -> [EventType :: binary()].
-callback max_retries() -> pos_integer().
-callback on_error(Error :: term(), Event :: map(), FailureContext :: #evoq_failure_context{handler_module :: atom(), event :: map(), error :: term(), attempt_number :: pos_integer(), first_failure_at :: integer(), last_failure_at :: integer(), first_failure_mono :: integer(), last_failure_mono :: integer(), stacktrace :: list()}, State :: term()) -> evoq_error_handler:error_action().
-callback project(Event :: map(), Metadata :: map(), State :: term(), ReadModel :: evoq_read_model:read_model()) -> {ok, NewState :: term(), NewReadModel :: evoq_read_model:read_model()} | {skip, State :: term(), ReadModel :: evoq_read_model:read_model()} | {error, Reason :: term()}.
Functions
Get the current checkpoint position: the highest position projected so far (the store's global position for events from the store subscription or a rebuild), or -1 when nothing has been projected yet.
Get event types this projection handles.
-spec get_read_model(pid()) -> evoq_read_model:read_model().
Get the read model instance.
Notify projection of an event.
Rebuild the projection from scratch. Clears the read model and replays all events.
Rebuild with options.
Start a projection.
Start a projection with options. Options: - store_id: Event store to replay from (overrides app env) - checkpoint_store: Module for persistent checkpoint storage