Current section
Files
Jump to
Current section
Files
src/eventsourcing.erl
-module(eventsourcing).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/eventsourcing.gleam").
-export([new/5, with_snapshots/2, execute_with_metadata/4, execute/3, load_aggregate/2, add_query/2, load_events/2, get_latest_snapshot/2]).
-export_type([aggregate/4, snapshot/1, snapshot_config/0, event_envelop/1, event_sourcing_error/1, event_sourcing/6, event_store/6]).
-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.
-type aggregate(MLY, MLZ, MMA, MMB) :: {aggregate, binary(), MLY, integer()} |
{gleam_phantom, MLZ, MMA, MMB}.
-type snapshot(MMC) :: {snapshot, binary(), MMC, integer(), integer()}.
-type snapshot_config() :: {snapshot_config, integer()}.
-type event_envelop(MMD) :: {memory_store_event_envelop,
binary(),
integer(),
MMD,
list({binary(), binary()})} |
{serialized_event_envelop,
binary(),
integer(),
MMD,
list({binary(), binary()}),
binary(),
binary(),
binary()}.
-type event_sourcing_error(MME) :: {domain_error, MME} |
{event_store_error, binary()} |
entity_not_found |
transaction_failed |
transaction_rolled_back.
-opaque event_sourcing(MMF, MMG, MMH, MMI, MMJ, MMK) :: {event_sourcing,
event_store(MMF, MMG, MMH, MMI, MMJ, MMK),
list(fun((binary(), list(event_envelop(MMI))) -> nil)),
fun((MMG, MMH) -> {ok, list(MMI)} | {error, MMJ}),
fun((MMG, MMI) -> MMG),
MMG,
gleam@option:option(snapshot_config())}.
-type event_store(MML, MMM, MMN, MMO, MMP, MMQ) :: {event_store,
fun((fun((MMQ) -> {ok, nil} | {error, event_sourcing_error(MMP)})) -> {ok,
nil} |
{error, event_sourcing_error(MMP)}),
fun((fun((MMQ) -> {ok, aggregate(MMM, MMN, MMO, MMP)} |
{error, event_sourcing_error(MMP)})) -> {ok,
aggregate(MMM, MMN, MMO, MMP)} |
{error, event_sourcing_error(MMP)}),
fun((fun((MMQ) -> {ok, list(event_envelop(MMO))} |
{error, event_sourcing_error(MMP)})) -> {ok,
list(event_envelop(MMO))} |
{error, event_sourcing_error(MMP)}),
fun((fun((MMQ) -> {ok, gleam@option:option(snapshot(MMM))} |
{error, event_sourcing_error(MMP)})) -> {ok,
gleam@option:option(snapshot(MMM))} |
{error, event_sourcing_error(MMP)}),
fun((MMQ, aggregate(MMM, MMN, MMO, MMP), list(MMO), list({binary(),
binary()})) -> {ok, {list(event_envelop(MMO)), integer()}} |
{error, event_sourcing_error(MMP)}),
fun((MML, MMQ, binary(), integer()) -> {ok, list(event_envelop(MMO))} |
{error, event_sourcing_error(MMP)}),
fun((MMQ, binary()) -> {ok, gleam@option:option(snapshot(MMM))} |
{error, event_sourcing_error(MMP)}),
fun((MMQ, snapshot(MMM)) -> {ok, nil} |
{error, event_sourcing_error(MMP)}),
MML}.
-file("src/eventsourcing.gleam", 195).
?DOC(
" Creates a new EventSourcing instance with the provided configuration.\n"
" \n"
" ## Arguments\n"
" - `event_store`: The storage implementation to use\n"
" - `queries`: List of query handlers to process events\n"
" - `handle`: Function to handle commands\n"
" - `apply`: Function to apply events\n"
" - `empty_state`: Initial state for new aggregates\n"
" \n"
" ## Returns\n"
" A new EventSourcing instance without snapshot support\n"
).
-spec new(
event_store(MND, MNE, MNF, MNG, MNH, MNI),
list(fun((binary(), list(event_envelop(MNG))) -> nil)),
fun((MNE, MNF) -> {ok, list(MNG)} | {error, MNH}),
fun((MNE, MNG) -> MNE),
MNE
) -> event_sourcing(MND, MNE, MNF, MNG, MNH, MNI).
new(Event_store, Queries, Handle, Apply, Empty_state) ->
{event_sourcing, Event_store, Queries, Handle, Apply, Empty_state, none}.
-file("src/eventsourcing.gleam", 227).
?DOC(
" \n"
" ## Arguments\n"
" - `event_sourcing`: The EventSourcing instance to modify\n"
" - `config`: Snapshot configuration specifying creation frequency\n"
" \n"
" ## Returns\n"
" A new EventSourcing instance with snapshot support enabled\n"
).
-spec with_snapshots(
event_sourcing(MNY, MNZ, MOA, MOB, MOC, MOD),
snapshot_config()
) -> event_sourcing(MNY, MNZ, MOA, MOB, MOC, MOD).
with_snapshots(Event_sourcing, Config) ->
{event_sourcing,
erlang:element(2, Event_sourcing),
erlang:element(3, Event_sourcing),
erlang:element(4, Event_sourcing),
erlang:element(5, Event_sourcing),
erlang:element(6, Event_sourcing),
{some, Config}}.
-file("src/eventsourcing.gleam", 351).
-spec load_aggregate_or_create_new(
event_sourcing(any(), MPW, MPX, MPY, MPZ, MQA),
MQA,
binary()
) -> {ok, aggregate(MPW, MPX, MPY, MPZ)} | {error, event_sourcing_error(MPZ)}.
load_aggregate_or_create_new(Eventsourcing, Tx, Aggregate_id) ->
begin
Result = case erlang:element(7, Eventsourcing) of
none ->
{ok, none};
{some, _} ->
(erlang:element(8, erlang:element(2, Eventsourcing)))(
Tx,
Aggregate_id
)
end,
case Result of
{ok, X} ->
{Starting_state, Starting_sequence} = case X of
none ->
{erlang:element(6, Eventsourcing), 0};
{some, Snapshot} ->
{erlang:element(3, Snapshot),
erlang:element(4, Snapshot)}
end,
begin
Result@1 = (erlang:element(
7,
erlang:element(2, Eventsourcing)
))(
erlang:element(10, erlang:element(2, Eventsourcing)),
Tx,
Aggregate_id,
Starting_sequence
),
case Result@1 of
{ok, X@1} ->
{ok,
begin
{Instance, Sequence@1} = begin
_pipe = X@1,
gleam@list:fold(
_pipe,
{Starting_state, Starting_sequence},
fun(
Aggregate_and_sequence,
Event_envelop
) ->
{Aggregate, Sequence} = Aggregate_and_sequence,
{(erlang:element(
5,
Eventsourcing
))(
Aggregate,
erlang:element(
4,
Event_envelop
)
),
Sequence + 1}
end
)
end,
{aggregate,
Aggregate_id,
Instance,
Sequence@1}
end};
{error, E} ->
{error, E}
end
end;
{error, E@1} ->
{error, E@1}
end
end.
-file("src/eventsourcing.gleam", 282).
?DOC(
" Executes a command with additional metadata.\n"
" \n"
" ## Arguments\n"
" - `event_sourcing`: The EventSourcing instance\n"
" - `aggregate_id`: ID of the aggregate to execute command against\n"
" - `command`: The command to execute\n"
" - `metadata`: Additional metadata to store with generated events\n"
" \n"
" ## Returns\n"
" Ok(Nil) if successful, or an error if command handling fails\n"
).
-spec execute_with_metadata(
event_sourcing(any(), any(), MPH, any(), MPJ, any()),
binary(),
MPH,
list({binary(), binary()})
) -> {ok, nil} | {error, event_sourcing_error(MPJ)}.
execute_with_metadata(Event_sourcing, Aggregate_id, Command, Metadata) ->
(erlang:element(2, erlang:element(2, Event_sourcing)))(
fun(Tx) ->
gleam@result:'try'(
load_aggregate_or_create_new(Event_sourcing, Tx, Aggregate_id),
fun(Aggregate) ->
gleam@result:'try'(
begin
_pipe = (erlang:element(4, Event_sourcing))(
erlang:element(3, Aggregate),
Command
),
gleam@result:map_error(
_pipe,
fun(Error) -> {domain_error, Error} end
)
end,
fun(Events) ->
Post_command_aggregate = {aggregate,
erlang:element(2, Aggregate),
begin
_pipe@1 = Events,
gleam@list:fold(
_pipe@1,
erlang:element(3, Aggregate),
fun(Entity, Event) ->
(erlang:element(5, Event_sourcing))(
Entity,
Event
)
end
)
end,
erlang:element(4, Aggregate)},
gleam@result:'try'(
(erlang:element(
6,
erlang:element(2, Event_sourcing)
))(Tx, Post_command_aggregate, Events, Metadata),
fun(_use0) ->
{Commited_events, Sequence} = _use0,
_pipe@2 = case erlang:element(
7,
Event_sourcing
) of
{some, Config} when ((Sequence rem erlang:element(
2,
Config
)) =:= 0) andalso (erlang:element(
2,
Config
) =/= 0) ->
Snapshot = {snapshot,
erlang:element(2, Aggregate),
erlang:element(
3,
Post_command_aggregate
),
Sequence,
erlang:element(
1,
gleam@time@timestamp:to_unix_seconds_and_nanoseconds(
gleam@time@timestamp:system_time(
)
)
)},
(erlang:element(
9,
erlang:element(
2,
Event_sourcing
)
))(Tx, Snapshot);
_ ->
{ok, nil}
end,
gleam@result:'try'(
_pipe@2,
fun(_) ->
_pipe@3 = erlang:element(
3,
Event_sourcing
),
gleam@list:map(
_pipe@3,
fun(Query) ->
Query(
erlang:element(
2,
Aggregate
),
Commited_events
)
end
),
{ok, nil}
end
)
end
)
end
)
end
)
end
).
-file("src/eventsourcing.gleam", 257).
?DOC(
" Executes a command against an aggregate.\n"
" \n"
" ## Arguments\n"
" - `event_sourcing`: The EventSourcing instance\n"
" - `aggregate_id`: ID of the aggregate to execute command against\n"
" - `command`: The command to execute\n"
" \n"
" ## Returns\n"
" Ok(Nil) if successful, or an error if command handling fails\n"
).
-spec execute(
event_sourcing(any(), any(), MOS, any(), MOU, any()),
binary(),
MOS
) -> {ok, nil} | {error, event_sourcing_error(MOU)}.
execute(Event_sourcing, Aggregate_id, Command) ->
execute_with_metadata(Event_sourcing, Aggregate_id, Command, []).
-file("src/eventsourcing.gleam", 400).
?DOC(
" Loads the current state of an aggregate.\n"
" \n"
" ## Arguments\n"
" - `event_sourcing`: The EventSourcing instance\n"
" - `aggregate_id`: ID of the aggregate to load\n"
" \n"
" ## Returns\n"
" The current state of the aggregate, or an error if loading fails\n"
).
-spec load_aggregate(event_sourcing(any(), MQP, MQQ, MQR, MQS, any()), binary()) -> {ok,
aggregate(MQP, MQQ, MQR, MQS)} |
{error, event_sourcing_error(MQS)}.
load_aggregate(Event_sourcing, Aggregate_id) ->
(erlang:element(3, erlang:element(2, Event_sourcing)))(
fun(Tx) ->
_pipe = load_aggregate_or_create_new(
Event_sourcing,
Tx,
Aggregate_id
),
gleam@result:'try'(
_pipe,
fun(Aggregate) ->
case erlang:element(3, Aggregate) =:= erlang:element(
6,
Event_sourcing
) of
true ->
{error, entity_not_found};
false ->
{ok, Aggregate}
end
end
)
end
).
-file("src/eventsourcing.gleam", 425).
?DOC(
" Add a query to the EventSourcing instance.\n"
"\n"
" Queries are functions that run when events are committed.\n"
" They can be used for things like updating read models or sending notifications.\n"
).
-spec add_query(
event_sourcing(MRH, MRI, MRJ, MRK, MRL, MRM),
fun((binary(), list(event_envelop(MRK))) -> nil)
) -> event_sourcing(MRH, MRI, MRJ, MRK, MRL, MRM).
add_query(Eventsourcing, Query) ->
{event_sourcing,
erlang:element(2, Eventsourcing),
[Query | erlang:element(3, Eventsourcing)],
erlang:element(4, Eventsourcing),
erlang:element(5, Eventsourcing),
erlang:element(6, Eventsourcing),
erlang:element(7, Eventsourcing)}.
-file("src/eventsourcing.gleam", 459).
?DOC(
" Loads all events for an aggregate from a specified sequence number.\n"
" \n"
" This function retrieves all events for an aggregate starting from a given sequence number,\n"
" allowing for partial event stream loading and event replay from a specific point in time.\n"
" \n"
" ## Arguments\n"
" - `event_sourcing`: The EventSourcing instance\n"
" - `aggregate_id`: ID of the aggregate whose events should be loaded\n"
" - `start_from`: The sequence number to start loading events from\n"
" \n"
" ## Returns\n"
" A Result containing:\n"
" - Ok(List(EventEnvelop(event))): List of events if successful\n"
" - Error(EventSourcingError): If loading fails\n"
" \n"
" ## Example\n"
" ```gleam\n"
" let assert Ok(events) = load_events(event_sourcing, \"account-123\", 5)\n"
" // events will contain all events for account-123 starting from sequence 5\n"
" ```\n"
).
-spec load_events(
event_sourcing(any(), any(), any(), MRY, MRZ, any()),
binary()
) -> {ok, list(event_envelop(MRY))} | {error, event_sourcing_error(MRZ)}.
load_events(Event_sourcing, Aggregate_id) ->
(erlang:element(4, erlang:element(2, Event_sourcing)))(
fun(Tx) ->
(erlang:element(7, erlang:element(2, Event_sourcing)))(
erlang:element(10, erlang:element(2, Event_sourcing)),
Tx,
Aggregate_id,
0
)
end
).
-file("src/eventsourcing.gleam", 503).
?DOC(
" Retrieves the most recent snapshot for an aggregate if it exists.\n"
" \n"
" This function attempts to load the latest snapshot for an aggregate, which can be\n"
" used as a starting point for rebuilding aggregate state without replaying all events\n"
" from the beginning.\n"
" \n"
" ## Arguments\n"
" - `event_sourcing`: The EventSourcing instance\n"
" - `aggregate_id`: ID of the aggregate to get the snapshot for\n"
" \n"
" ## Returns\n"
" A Result containing:\n"
" - Ok(Some(Snapshot)): The latest snapshot if one exists\n"
" - Ok(None): If no snapshot exists for the aggregate\n"
" - Error(EventSourcingError): If snapshot retrieval fails\n"
" \n"
" ## Example\n"
" ```gleam\n"
" let assert Ok(maybe_snapshot) = get_latest_snapshot(event_sourcing, \"account-123\")\n"
" case maybe_snapshot {\n"
" Some(snapshot) -> // Use snapshot as starting point\n"
" None -> // No snapshot exists, start from initial state\n"
" }\n"
" ```\n"
).
-spec get_latest_snapshot(
event_sourcing(any(), MSN, any(), any(), MSQ, any()),
binary()
) -> {ok, gleam@option:option(snapshot(MSN))} |
{error, event_sourcing_error(MSQ)}.
get_latest_snapshot(Event_sourcing, Aggregate_id) ->
(erlang:element(5, erlang:element(2, Event_sourcing)))(
fun(Tx) -> case erlang:element(7, Event_sourcing) of
none ->
{ok, none};
{some, _} ->
(erlang:element(8, erlang:element(2, Event_sourcing)))(
Tx,
Aggregate_id
)
end end
).