eventsourcing

-module(eventsourcing@memory_store).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/eventsourcing/memory_store.gleam").
-export([supervised/3]).
-export_type([memory_store/4, event_message/1, snapshot_message/1]).

-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.

-opaque memory_store(LQK, LQL, LQM, LQN) :: {memory_store,
        gleam@erlang@process:name(event_message(LQM)),
        gleam@erlang@process:name(snapshot_message(LQK))} |
    {gleam_phantom, LQL, LQN}.

-type event_message(LQO) :: {set_events,
        binary(),
        list(eventsourcing:event_envelop(LQO))} |
    {get_events,
        binary(),
        gleam@erlang@process:subject(gleam@option:option(list(eventsourcing:event_envelop(LQO))))}.

-type snapshot_message(LQP) :: {set_snapshot,
        binary(),
        eventsourcing:snapshot(LQP)} |
    {get_snapshot,
        binary(),
        gleam@erlang@process:subject(gleam@option:option(eventsourcing:snapshot(LQP)))}.

-file("src/eventsourcing/memory_store.gleam", 153).
-spec handle_events_message(
    gleam@dict:dict(binary(), list(eventsourcing:event_envelop(LSQ))),
    event_message(LSQ)
) -> gleam@otp@actor:next(gleam@dict:dict(binary(), list(eventsourcing:event_envelop(LSQ))), any()).
handle_events_message(State, Message) ->
    case Message of
        {set_events, Key, Value} ->
            _pipe = State,
            _pipe@1 = gleam@dict:insert(_pipe, Key, Value),
            gleam@otp@actor:continue(_pipe@1);

        {get_events, Key@1, Response} ->
            Value@1 = begin
                _pipe@2 = State,
                _pipe@3 = gleam_stdlib:map_get(_pipe@2, Key@1),
                gleam@option:from_result(_pipe@3)
            end,
            gleam@otp@actor:send(Response, Value@1),
            gleam@otp@actor:continue(State)
    end.

-file("src/eventsourcing/memory_store.gleam", 52).
-spec supervised_events_actor(
    gleam@erlang@process:subject(gleam@otp@actor:started(gleam@erlang@process:subject(event_message(LQZ)))),
    gleam@erlang@process:name(event_message(LQZ))
) -> gleam@otp@supervision:child_specification(gleam@erlang@process:subject(event_message(LQZ))).
supervised_events_actor(Events_actor_receiver, Name) ->
    gleam@otp@supervision:worker(
        fun() ->
            gleam@result:map(
                begin
                    _pipe = gleam@otp@actor:new(maps:new()),
                    _pipe@1 = gleam@otp@actor:on_message(
                        _pipe,
                        fun handle_events_message/2
                    ),
                    _pipe@2 = gleam@otp@actor:named(_pipe@1, Name),
                    gleam@otp@actor:start(_pipe@2)
                end,
                fun(Started) ->
                    gleam@erlang@process:send(Events_actor_receiver, Started),
                    Started
                end
            )
        end
    ).

-file("src/eventsourcing/memory_store.gleam", 203).
-spec wrap_events(binary(), integer(), list(LUB), list({binary(), binary()})) -> list(eventsourcing:event_envelop(LUB)).
wrap_events(Aggregate_id, Current_sequence, Events, Metadata) ->
    _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,
                    Metadata}}
        end
    ),
    gleam@pair:second(_pipe).

-file("src/eventsourcing/memory_store.gleam", 228).
-spec handle_snapshot_message(
    gleam@dict:dict(binary(), eventsourcing:snapshot(LUG)),
    snapshot_message(LUG)
) -> gleam@otp@actor:next(gleam@dict:dict(binary(), eventsourcing:snapshot(LUG)), any()).
handle_snapshot_message(State, Message) ->
    case Message of
        {set_snapshot, Key, Value} ->
            _pipe = State,
            _pipe@1 = gleam@dict:insert(_pipe, Key, Value),
            gleam@otp@actor:continue(_pipe@1);

        {get_snapshot, Key@1, Response} ->
            Value@1 = begin
                _pipe@2 = State,
                _pipe@3 = gleam_stdlib:map_get(_pipe@2, Key@1),
                gleam@option:from_result(_pipe@3)
            end,
            gleam@otp@actor:send(Response, Value@1),
            gleam@otp@actor:continue(State)
    end.

-file("src/eventsourcing/memory_store.gleam", 70).
-spec supervised_snapshot_actor(
    gleam@erlang@process:subject(gleam@otp@actor:started(gleam@erlang@process:subject(snapshot_message(LRJ)))),
    gleam@erlang@process:name(snapshot_message(LRJ))
) -> gleam@otp@supervision:child_specification(gleam@erlang@process:subject(snapshot_message(LRJ))).
supervised_snapshot_actor(Snapshot_actor_receiver, Name) ->
    gleam@otp@supervision:worker(
        fun() ->
            gleam@result:map(
                begin
                    _pipe = gleam@otp@actor:new(maps:new()),
                    _pipe@1 = gleam@otp@actor:on_message(
                        _pipe,
                        fun handle_snapshot_message/2
                    ),
                    _pipe@2 = gleam@otp@actor:named(_pipe@1, Name),
                    gleam@otp@actor:start(_pipe@2)
                end,
                fun(Started) ->
                    gleam@erlang@process:send(Snapshot_actor_receiver, Started),
                    Started
                end
            )
        end
    ).

-file("src/eventsourcing/memory_store.gleam", 244).
-spec save_snapshot(
    memory_store(LUK, any(), any(), LUN),
    eventsourcing:snapshot(LUK)
) -> {ok, nil} | {error, eventsourcing:event_sourcing_error(LUN)}.
save_snapshot(Memory_store, Snapshot) ->
    _pipe = gleam@otp@actor:send(
        gleam@erlang@process:named_subject(erlang:element(3, Memory_store)),
        {set_snapshot, erlang:element(2, Snapshot), Snapshot}
    ),
    {ok, _pipe}.

-file("src/eventsourcing/memory_store.gleam", 166).
-spec load_events(
    memory_store(any(), any(), LSW, LSX),
    any(),
    binary(),
    integer()
) -> {ok, list(eventsourcing:event_envelop(LSW))} |
    {error, eventsourcing:event_sourcing_error(LSX)}.
load_events(Memory_store, _, Aggregate_id, Start_from) ->
    Events = gleam@erlang@process:call(
        gleam@erlang@process:named_subject(erlang:element(2, Memory_store)),
        1000,
        fun(_capture) -> {get_events, Aggregate_id, _capture} end
    ),
    case Events of
        {some, Events@1} ->
            {ok,
                begin
                    _pipe = Events@1,
                    gleam@list:drop(_pipe, Start_from)
                end};

        none ->
            {ok, []}
    end.

-file("src/eventsourcing/memory_store.gleam", 184).
-spec commit_events(
    memory_store(LTI, LTJ, LTK, LTL),
    eventsourcing:aggregate(LTI, LTJ, LTK, LTL),
    list(LTK),
    list({binary(), binary()})
) -> {ok, {list(eventsourcing:event_envelop(LTK)), integer()}} |
    {error, eventsourcing:event_sourcing_error(LTL)}.
commit_events(Memory_store, Aggregate, Events, Metadata) ->
    {aggregate, Aggregate_id, _, Sequence} = Aggregate,
    Wrapped_events = wrap_events(Aggregate_id, Sequence, Events, Metadata),
    begin
        Result = load_events(Memory_store, nil, Aggregate_id, 0),
        case Result of
            {ok, X} ->
                {ok,
                    begin
                        All_events = lists:append(X, Wrapped_events),
                        gleam@otp@actor:send(
                            gleam@erlang@process:named_subject(
                                erlang:element(2, Memory_store)
                            ),
                            {set_events, Aggregate_id, All_events}
                        ),
                        Last_event@1 = case gleam@list:last(Wrapped_events) of
                            {ok, Last_event} -> Last_event;
                            _assert_fail ->
                                erlang:error(#{gleam_error => let_assert,
                                            message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
                                            file => <<?FILEPATH/utf8>>,
                                            module => <<"eventsourcing/memory_store"/utf8>>,
                                            function => <<"commit_events"/utf8>>,
                                            line => 198,
                                            value => _assert_fail,
                                            start => 6063,
                                            'end' => 6116,
                                            pattern_start => 6074,
                                            pattern_end => 6088})
                        end,
                        {Wrapped_events, erlang:element(3, Last_event@1)}
                    end};

            {error, E} ->
                {error, E}
        end
    end.

-file("src/eventsourcing/memory_store.gleam", 255).
-spec load_snapshot(memory_store(LUW, any(), any(), LUZ), binary()) -> {ok,
        gleam@option:option(eventsourcing:snapshot(LUW))} |
    {error, eventsourcing:event_sourcing_error(LUZ)}.
load_snapshot(Memory_store, Aggregate_id) ->
    {ok,
        gleam@erlang@process:call(
            gleam@erlang@process:named_subject(erlang:element(3, Memory_store)),
            1000,
            fun(_capture) -> {get_snapshot, Aggregate_id, _capture} end
        )}.

-file("src/eventsourcing/memory_store.gleam", 102).
?DOC(
    " Creates a supervised in-memory event store with separate actors for events and snapshots.\n"
    " Returns both the EventStore interface and a supervision specification that manages\n"
    " the underlying storage actors.\n"
    "\n"
    " ## Example\n"
    " ```gleam\n"
    " let events_name = process.new_name(\"events_actor\")\n"
    " let snapshot_name = process.new_name(\"snapshot_actor\")\n"
    " let #(eventstore, supervisor_spec) = memory_store.supervised(\n"
    "   events_name,\n"
    "   snapshot_name,\n"
    "   static_supervisor.OneForOne\n"
    " )\n"
    " ```\n"
).
-spec supervised(
    gleam@erlang@process:name(event_message(LRT)),
    gleam@erlang@process:name(snapshot_message(LRW)),
    gleam@otp@static_supervisor:strategy()
) -> {eventsourcing:event_store(memory_store(LRW, LRZ, LRT, LSA), LRW, LRZ, LRT, LSA, memory_store(LRW, LRZ, LRT, LSA)),
    gleam@otp@supervision:child_specification(gleam@otp@static_supervisor:supervisor())}.
supervised(Events_actor_name, Snapshot_actor_name, Strategy) ->
    Events_actor_receiver = gleam@erlang@process:new_subject(),
    Supervised_events_actor_spec = supervised_events_actor(
        Events_actor_receiver,
        Events_actor_name
    ),
    Snapshot_actor_receiver = gleam@erlang@process:new_subject(),
    Supervised_snapshot_actor_spec = supervised_snapshot_actor(
        Snapshot_actor_receiver,
        Snapshot_actor_name
    ),
    Memory_store = {memory_store, Events_actor_name, Snapshot_actor_name},
    Eventstore = {event_store,
        fun(F) -> F(Memory_store) end,
        fun(F@1) -> F@1(Memory_store) end,
        fun(F@2) -> F@2(Memory_store) end,
        fun(F@3) -> F@3(Memory_store) end,
        fun commit_events/4,
        fun load_events/4,
        fun load_snapshot/2,
        fun save_snapshot/2,
        Memory_store},
    {Eventstore,
        begin
            _pipe = gleam@otp@static_supervisor:new(Strategy),
            _pipe@1 = gleam@otp@static_supervisor:add(
                _pipe,
                Supervised_events_actor_spec
            ),
            _pipe@2 = gleam@otp@static_supervisor:add(
                _pipe@1,
                Supervised_snapshot_actor_spec
            ),
            gleam@otp@static_supervisor:supervised(_pipe@2)
        end}.