evoq_aggregate behaviour (evoq v1.30.1)
View SourceAggregate behavior and GenServer implementation.
Aggregates are the consistency boundary in event sourcing. Each aggregate: - Has a unique stream ID - Declares a state module via state_module/0 - Processes commands via execute/2 callback - Applies events via apply/2 callback (typically delegates to state module) - Supports snapshots for fast recovery - Has configurable lifespan (TTL, hibernate, passivate)
Callbacks
Required: - init(AggregateId) -> {ok, State} - execute(State, Command) -> {ok, [Event]} | {error, Reason} - apply(State, Event) -> NewState runs after execute/2 and on every later load, in stream version order, possibly starting from a from_snapshot/1 state; pure: State and Event only
Optional: - state_module() -> module() the evoq_state module the aggregate delegates to; evoq never calls it - snapshot(State) -> SnapshotData - from_snapshot(SnapshotData) -> State
Summary
Functions
Execute a command against an aggregate.
Execute a command and return the post-event aggregate state.
Get the current state of an aggregate (for debugging).
Get the current version of an aggregate.
Start an aggregate process (uses env store_id).
Start an aggregate process with explicit store_id.
Callbacks
Functions
-spec execute_command(pid(), #evoq_command{command_id :: binary() | undefined, command_type :: atom() | undefined, aggregate_type :: atom() | undefined, aggregate_id :: binary() | undefined, payload :: map(), metadata :: map(), causation_id :: binary() | undefined, correlation_id :: binary() | undefined, idempotency_key :: binary() | undefined}) -> {ok, integer(), [map()]} | {error, term()}.
Execute a command against an aggregate.
The command's ids are ensured here (evoq_command:ensure_id/1), exactly as the command router does for a dispatched command: a caller that talks to an aggregate directly gets the same minted command_id and correlation_id instead of events stamped with undefined. Both paths are idempotent, so a command that already went through the router is unchanged.
AN AGGREGATE THAT CANNOT ANSWER IS AN ERROR HERE, NOT AN EXIT FOR ITS CALLER. The store load runs after init (see handle_continue/2), so a store that cannot be read takes this process down while a caller is waiting on it. Without the catch below that caller exits with this process, and the dispatcher's own {error, _} clauses never see the failure.
-spec execute_command_with_state(pid(), #evoq_command{command_id :: binary() | undefined, command_type :: atom() | undefined, aggregate_type :: atom() | undefined, aggregate_id :: binary() | undefined, payload :: map(), metadata :: map(), causation_id :: binary() | undefined, correlation_id :: binary() | undefined, idempotency_key :: binary() | undefined}) -> {ok, integer(), [map()], term()} | {error, term()}.
Execute a command and return the post-event aggregate state.
Like execute_command/2 but includes the aggregate state after applying all new events. Enables session-level consistency where the caller receives immediate truth about the resulting state.
Get the current state of an aggregate (for debugging).
A READ ANSWERS LIKE A COMMAND HERE, for the same reason execute_command/2 does: since the load moved off init/1 the registry hands out this pid BEFORE the stream is loaded, so a caller waiting here is waiting on that load. Two shapes of it used to end the caller instead of answering it: a store read that fails kills the process mid-call, and a cold stream whose replay outruns gen_server's 5 s default times the call out. Neither was reachable while the load ran inside init/1, before the pid existed. called/2 bounds neither wait but turns both into the value this call always claimed to return.
Get the current version of an aggregate.
Same contract as get_state/1. A caller that wants only the version of a stream it is willing to lose should still match {error, _}.
Start an aggregate process (uses env store_id).
Start an aggregate process with explicit store_id.