-module(eventsourcing). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]). -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(QYC, QYD, QYE, QYF) :: {aggregate, binary(), QYC, integer()} | {gleam_phantom, QYD, QYE, QYF}. -type snapshot(QYG) :: {snapshot, binary(), QYG, integer(), integer()}. -type snapshot_config() :: {snapshot_config, integer()}. -type event_envelop(QYH) :: {memory_store_event_envelop, binary(), integer(), QYH, list({binary(), binary()})} | {serialized_event_envelop, binary(), integer(), QYH, list({binary(), binary()}), binary(), binary(), binary()}. -type event_sourcing_error(QYI) :: {domain_error, QYI} | {event_store_error, binary()} | entity_not_found | transaction_failed | transaction_rolled_back. -opaque event_sourcing(QYJ, QYK, QYL, QYM, QYN, QYO) :: {event_sourcing, event_store(QYJ, QYK, QYL, QYM, QYN, QYO), list(fun((binary(), list(event_envelop(QYM))) -> nil)), fun((QYK, QYL) -> {ok, list(QYM)} | {error, QYN}), fun((QYK, QYM) -> QYK), QYK, gleam@option:option(snapshot_config())}. -type event_store(QYP, QYQ, QYR, QYS, QYT, QYU) :: {event_store, fun((fun((QYU) -> {ok, nil} | {error, event_sourcing_error(QYT)})) -> {ok, nil} | {error, event_sourcing_error(QYT)}), fun((fun((QYU) -> {ok, aggregate(QYQ, QYR, QYS, QYT)} | {error, event_sourcing_error(QYT)})) -> {ok, aggregate(QYQ, QYR, QYS, QYT)} | {error, event_sourcing_error(QYT)}), fun((fun((QYU) -> {ok, list(event_envelop(QYS))} | {error, event_sourcing_error(QYT)})) -> {ok, list(event_envelop(QYS))} | {error, event_sourcing_error(QYT)}), fun((fun((QYU) -> {ok, gleam@option:option(snapshot(QYQ))} | {error, event_sourcing_error(QYT)})) -> {ok, gleam@option:option(snapshot(QYQ))} | {error, event_sourcing_error(QYT)}), fun((QYU, aggregate(QYQ, QYR, QYS, QYT), list(QYS), list({binary(), binary()})) -> {ok, {list(event_envelop(QYS)), integer()}} | {error, event_sourcing_error(QYT)}), fun((QYP, QYU, binary(), integer()) -> {ok, list(event_envelop(QYS))} | {error, event_sourcing_error(QYT)}), fun((QYU, binary()) -> {ok, gleam@option:option(snapshot(QYQ))} | {error, event_sourcing_error(QYT)}), fun((QYU, snapshot(QYQ)) -> {ok, nil} | {error, event_sourcing_error(QYT)}), QYP}. -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(QZH, QZI, QZJ, QZK, QZL, QZM), list(fun((binary(), list(event_envelop(QZK))) -> nil)), fun((QZI, QZJ) -> {ok, list(QZK)} | {error, QZL}), fun((QZI, QZK) -> QZI), QZI ) -> event_sourcing(QZH, QZI, QZJ, QZK, QZL, QZM). 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(RAC, RAD, RAE, RAF, RAG, RAH), snapshot_config() ) -> event_sourcing(RAC, RAD, RAE, RAF, RAG, RAH). with_snapshots(Event_sourcing, Config) -> _record = Event_sourcing, {event_sourcing, erlang:element(2, _record), erlang:element(3, _record), erlang:element(4, _record), erlang:element(5, _record), erlang:element(6, _record), {some, Config}}. -file("src/eventsourcing.gleam", 346). -spec load_aggregate_or_create_new( event_sourcing(any(), RCA, RCB, RCC, RCD, RCE), RCE, binary() ) -> {ok, aggregate(RCA, RCB, RCC, RCD)} | {error, event_sourcing_error(RCD)}. load_aggregate_or_create_new(Eventsourcing, Tx, Aggregate_id) -> gleam@result:'try'(case erlang:element(7, Eventsourcing) of none -> {ok, none}; {some, _} -> (erlang:element(8, erlang:element(2, Eventsourcing)))( Tx, Aggregate_id ) end, fun(Maybe_snapshot) -> {Starting_state, Starting_sequence} = case Maybe_snapshot of none -> {erlang:element(6, Eventsourcing), 0}; {some, Snapshot} -> {erlang:element(3, Snapshot), erlang:element(4, Snapshot)} end, gleam@result:map( (erlang:element(7, erlang:element(2, Eventsourcing)))( erlang:element(10, erlang:element(2, Eventsourcing)), Tx, Aggregate_id, Starting_sequence ), fun(Events) -> {Instance, Sequence@1} = begin _pipe = Events, 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 ) 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(), RBL, any(), RBN, any()), binary(), RBL, list({binary(), binary()}) ) -> {ok, nil} | {error, event_sourcing_error(RBN)}. 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 = begin _record = Aggregate, {aggregate, erlang:element(2, _record), 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, _record)} end, 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, birl:to_unix(birl:now())}, (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(), RAW, any(), RAY, any()), binary(), RAW ) -> {ok, nil} | {error, event_sourcing_error(RAY)}. execute(Event_sourcing, Aggregate_id, Command) -> execute_with_metadata(Event_sourcing, Aggregate_id, Command, []). -file("src/eventsourcing.gleam", 395). ?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(), RCT, RCU, RCV, RCW, any()), binary()) -> {ok, aggregate(RCT, RCU, RCV, RCW)} | {error, event_sourcing_error(RCW)}. 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", 420). ?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(RDL, RDM, RDN, RDO, RDP, RDQ), fun((binary(), list(event_envelop(RDO))) -> nil) ) -> event_sourcing(RDL, RDM, RDN, RDO, RDP, RDQ). add_query(Eventsourcing, Query) -> _record = Eventsourcing, {event_sourcing, erlang:element(2, _record), [Query | erlang:element(3, Eventsourcing)], erlang:element(4, _record), erlang:element(5, _record), erlang:element(6, _record), erlang:element(7, _record)}. -file("src/eventsourcing.gleam", 454). ?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(), REC, RED, any()), binary() ) -> {ok, list(event_envelop(REC))} | {error, event_sourcing_error(RED)}. 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", 498). ?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(), RER, any(), any(), REU, any()), binary() ) -> {ok, gleam@option:option(snapshot(RER))} | {error, event_sourcing_error(REU)}. 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 ).