eventsourcing

-module(eventsourcing@memory_store).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).

-export([load_events/2, load_aggregate/2, new/3]).
-export_type([memory_store/4, message/1]).

-opaque memory_store(PHP, PHQ, PHR, PHS) :: {memory_store,
        gleam@erlang@process:subject(message(PHR)),
        eventsourcing:aggregate(PHP, PHQ, PHR, PHS)}.

-type message(PHT) :: {set, binary(), list(eventsourcing:event_envelop(PHT))} |
    {get,
        binary(),
        gleam@erlang@process:subject({ok,
                list(eventsourcing:event_envelop(PHT))} |
            {error, nil})}.

-spec handle_message(
    message(PIK),
    gleam@dict:dict(binary(), list(eventsourcing:event_envelop(PIK)))
) -> gleam@otp@actor:next(any(), gleam@dict:dict(binary(), list(eventsourcing:event_envelop(PIK)))).
handle_message(Message, State) ->
    case Message of
        {set, Key, Value} ->
            _pipe = State,
            _pipe@1 = gleam@dict:insert(_pipe, Key, Value),
            gleam@otp@actor:continue(_pipe@1);

        {get, Key@1, Response} ->
            Value@1 = begin
                _pipe@2 = State,
                gleam@dict:get(_pipe@2, Key@1)
            end,
            gleam@otp@actor:send(Response, Value@1),
            gleam@otp@actor:continue(State)
    end.

-spec load_commited_events(memory_store(any(), any(), PIQ, any()), binary()) -> list(eventsourcing:event_envelop(PIQ)).
load_commited_events(Memory_store, Aggregate_id) ->
    _pipe = gleam@otp@actor:call(
        erlang:element(2, Memory_store),
        fun(_capture) -> {get, Aggregate_id, _capture} end,
        10000
    ),
    gleam@result:unwrap(_pipe, []).

-spec load_events(memory_store(any(), any(), PIZ, any()), binary()) -> list(eventsourcing:event_envelop(PIZ)).
load_events(Memory_store, Aggregate_id) ->
    _pipe = load_commited_events(Memory_store, Aggregate_id),
    (fun(Events) ->
        gleam@io:println(
            <<<<<<<<"loading: "/utf8,
                            (begin
                                _pipe@1 = Events,
                                _pipe@2 = erlang:length(_pipe@1),
                                gleam@int:to_string(_pipe@2)
                            end)/binary>>/binary,
                        " events for Aggregate ID '"/utf8>>/binary,
                    Aggregate_id/binary>>/binary,
                "'"/utf8>>
        ),
        Events
    end)(_pipe).

-spec load_aggregate(memory_store(PJG, PJH, PJI, PJJ), binary()) -> eventsourcing:aggregate_context(PJG, PJH, PJI, PJJ).
load_aggregate(Memory_store, Aggregate_id) ->
    Commited_events = load_events(Memory_store, Aggregate_id),
    {Aggregate@1, Sequence} = gleam@list:fold(
        begin
            _pipe = Commited_events,
            lists:reverse(_pipe)
        end,
        {erlang:element(3, Memory_store), 0},
        fun(Aggregate_and_sequence, Event_envelop) ->
            {Aggregate, _} = Aggregate_and_sequence,
            {erlang:setelement(
                    2,
                    Aggregate,
                    (erlang:element(4, Aggregate))(
                        erlang:element(2, Aggregate),
                        erlang:element(4, Event_envelop)
                    )
                ),
                erlang:element(3, Event_envelop)}
        end
    ),
    {aggregate_context, Aggregate_id, Aggregate@1, Sequence}.

-spec wrap_events(binary(), integer(), list(PKG)) -> list(eventsourcing:event_envelop(PKG)).
wrap_events(Aggregate_id, Current_sequence, Events) ->
    _pipe = gleam@list:map_fold(
        Events,
        Current_sequence,
        fun(Sequence, Event) ->
            Next_sequence = Sequence + 1,
            {Next_sequence,
                {memory_store_event_envelop, Aggregate_id, Sequence + 1, Event}}
        end
    ),
    gleam@pair:second(_pipe).

-spec commit(
    memory_store(PJS, PJT, PJU, PJV),
    eventsourcing:aggregate_context(PJS, PJT, PJU, PJV),
    list(PJU)
) -> list(eventsourcing:event_envelop(PJU)).
commit(Memory_store, Context, Events) ->
    {aggregate_context, Aggregate_id, _, Sequence} = Context,
    Wrapped_events = wrap_events(Aggregate_id, Sequence, Events),
    Past_events = load_commited_events(Memory_store, Aggregate_id),
    Events@1 = lists:append(Wrapped_events, Past_events),
    gleam@io:println(
        <<<<<<<<"storing: "/utf8,
                        (begin
                            _pipe = Wrapped_events,
                            _pipe@1 = erlang:length(_pipe),
                            gleam@int:to_string(_pipe@1)
                        end)/binary>>/binary,
                    " events for Aggregate ID '"/utf8>>/binary,
                Aggregate_id/binary>>/binary,
            "'"/utf8>>
    ),
    gleam@otp@actor:send(
        erlang:element(2, Memory_store),
        {set, Aggregate_id, Events@1}
    ),
    Wrapped_events.

-spec new(
    PHZ,
    fun((PHZ, PIA) -> {ok, list(PIB)} | {error, PIC}),
    fun((PHZ, PIB) -> PHZ)
) -> eventsourcing:event_store(memory_store(PHZ, PIA, PIB, PIC), PHZ, PIA, PIB, PIC).
new(Empty_entity, Handle, Apply) ->
    _assert_subject = begin
        _pipe = gleam@otp@actor:start(gleam@dict:new(), fun handle_message/2),
        gleam@result:'try'(
            _pipe,
            fun(Subject) ->
                {ok,
                    {memory_store,
                        Subject,
                        {aggregate, Empty_entity, Handle, Apply}}}
            end
        )
    end,
    {ok, Actor} = case _assert_subject of
        {ok, _} -> _assert_subject;
        _assert_fail ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Assertion pattern match failed"/utf8>>,
                        value => _assert_fail,
                        module => <<"eventsourcing/memory_store"/utf8>>,
                        function => <<"new"/utf8>>,
                        line => 31})
    end,
    {event_store, Actor, fun load_aggregate/2, fun commit/3}.