signal

-module(signal).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).

-export([configure/1, with_subscriber/2, with_persistance_layer/2, with_pool_size_limit/2, with_custom_logger/2, without_info_logging/1, without_debug_logging/1, in_memory_persistance_handler/2, format_telemetry_message/2, telemetry_log_level/1, console_logger/2, aggregate/2, create/2, handle_command/2, get_state/1, get_id/1, get_current_pool_size/1, start/1]).
-export_type([event/1, context_message/3, signal/3, store_message/1, subscriber/2, consumer_message/2, signal_config/4, aggregate_config/3, telemetry_message/0, telemetry_event/0, log_level/0, signal_service/3, aggregate_state/3, aggregate_message/3, aggregate_update_context/3, pool_message/3, bus_message/1]).

-type event(JUX) :: {event,
        integer(),
        binary(),
        binary(),
        binary(),
        binary(),
        JUX}.

-opaque context_message(JUY, JUZ, JVA) :: {get_logger,
        gleam@erlang@process:subject({ok,
                gleam@erlang@process:subject(telemetry_message())} |
            {error, binary()})} |
    {set_logger, gleam@erlang@process:subject(telemetry_message())} |
    {get_store,
        gleam@erlang@process:subject({ok,
                gleam@erlang@process:subject(store_message(JVA))} |
            {error, binary()})} |
    {set_store, gleam@erlang@process:subject(store_message(JVA))} |
    {get_bus,
        gleam@erlang@process:subject({ok,
                gleam@erlang@process:subject(bus_message(JVA))} |
            {error, binary()})} |
    {set_bus, gleam@erlang@process:subject(bus_message(JVA))} |
    {get_aggregate_pool,
        gleam@erlang@process:subject({ok,
                gleam@erlang@process:subject(pool_message(JUY, JUZ, JVA))} |
            {error, binary()})} |
    {set_aggregate_pool,
        gleam@erlang@process:subject(pool_message(JUY, JUZ, JVA))} |
    shutdown_signal.

-opaque signal(JVB, JVC, JVD) :: {signal,
        gleam@erlang@process:subject(context_message(JVB, JVC, JVD))}.

-type store_message(JVE) :: {get_stored_events,
        gleam@erlang@process:subject({ok, list(event(JVE))} | {error, binary()}),
        binary()} |
    {is_identity_available,
        gleam@erlang@process:subject({ok, boolean()} | {error, binary()}),
        binary()} |
    {store_event, event(JVE)} |
    shutdown_persistance_layer.

-type subscriber(JVF, JVG) :: {consumer,
        gleam@erlang@process:subject(consumer_message(JVF, JVG))}.

-type consumer_message(JVH, JVI) :: {consume, event(JVI)} |
    {get_consumer_state, gleam@erlang@process:subject(JVH)} |
    shutdown_consumer.

-opaque signal_config(JVJ, JVK, JVL, JVM) :: {signal_config,
        aggregate_config(JVJ, JVL, JVM),
        gleam@option:option(gleam@erlang@process:subject(store_message(JVM))),
        list(subscriber(JVK, JVM)),
        integer(),
        gleam@option:option(gleam@erlang@process:subject(telemetry_message())),
        boolean(),
        boolean()}.

-type aggregate_config(JVN, JVO, JVP) :: {aggregate_config,
        JVN,
        fun((JVO, JVN) -> {ok, list(JVP)} | {error, binary()}),
        fun((JVN, event(JVP)) -> JVN)}.

-type telemetry_message() :: {report, telemetry_event(), binary()} |
    shutdown_telemetry.

-type telemetry_event() :: {pool_creating_aggregate, binary()} |
    {pool_created_aggregate, binary()} |
    {pool_cannot_create_aggregate_with_id, binary()} |
    {pool_hydrating_aggregate, binary()} |
    {pool_hydrated_aggregate, binary()} |
    {pool_aggregate_not_found, binary()} |
    {pool_rebalancing_started, integer()} |
    {pool_evicted_aggregate, binary()} |
    {pool_rebalancing_completed, integer()} |
    {aggregate_processing_command, binary(), binary()} |
    {aggregate_processed_command, binary(), binary()} |
    {aggregate_command_processing_failed, binary(), binary()} |
    {aggregate_events_produced, binary(), binary()} |
    {bus_triggering_subscribers, binary(), integer()} |
    {bus_subscribers_informed, binary(), integer()} |
    {store_pushed_event_to_write_ahead_log, binary(), integer()} |
    {store_write_ahead_log_size_warning, integer()} |
    {store_submitted_batch_for_persistance, integer()} |
    {store_persistance_completed, integer(), integer()}.

-type log_level() :: log_debug | log_info | log_warning | log_error.

-type signal_service(JVQ, JVR, JVS) :: {signal_service,
        gleam@option:option(gleam@erlang@process:subject(telemetry_message())),
        gleam@option:option(gleam@erlang@process:subject(pool_message(JVQ, JVR, JVS))),
        gleam@option:option(gleam@erlang@process:subject(bus_message(JVS))),
        gleam@option:option(gleam@erlang@process:subject(store_message(JVS)))}.

-type aggregate_state(JVT, JVU, JVV) :: {aggregate_state, integer(), JVT} |
    {gleam_phantom, JVU, JVV}.

-type aggregate_message(JVW, JVX, JVY) :: {state,
        gleam@erlang@process:subject(JVW)} |
    {identity, gleam@erlang@process:subject(binary())} |
    {handle_command,
        gleam@erlang@process:subject({ok, JVW} | {error, binary()}),
        JVX} |
    shutdown_aggregate |
    {gleam_phantom, JVY}.

-type aggregate_update_context(JVZ, JWA, JWB) :: {aggregate_update_context,
        gleam@erlang@process:subject(context_message(JVZ, JWA, JWB)),
        binary(),
        fun((JVZ, event(JWB)) -> JVZ),
        aggregate_state(JVZ, JWA, JWB)}.

-type pool_message(JWC, JWD, JWE) :: {create_aggregate,
        gleam@erlang@process:subject({ok,
                gleam@erlang@process:subject(aggregate_message(JWC, JWD, JWE))} |
            {error, binary()}),
        binary()} |
    {get_aggregate,
        gleam@erlang@process:subject({ok,
                gleam@erlang@process:subject(aggregate_message(JWC, JWD, JWE))} |
            {error, binary()}),
        binary()} |
    {pool_size, gleam@erlang@process:subject(integer())} |
    shutdown_pool.

-type bus_message(JWF) :: {push_event, event(JWF)} | shutdown_bus.

-spec configure(aggregate_config(JXG, JXH, JXI)) -> signal_config(JXG, any(), JXH, JXI).
configure(Agg) ->
    {signal_config, Agg, none, [], 100, none, true, true}.

-spec with_subscriber(signal_config(JXR, JXS, JXT, JXU), subscriber(JXS, JXU)) -> signal_config(JXR, JXS, JXT, JXU).
with_subscriber(Config, Sub) ->
    erlang:setelement(4, Config, [Sub | erlang:element(4, Config)]).

-spec with_persistance_layer(
    signal_config(JYF, JYG, JYH, JYI),
    gleam@erlang@process:subject(store_message(JYI))
) -> signal_config(JYF, JYG, JYH, JYI).
with_persistance_layer(Config, Persist) ->
    erlang:setelement(3, Config, {some, Persist}).

-spec with_pool_size_limit(signal_config(JYT, JYU, JYV, JYW), integer()) -> signal_config(JYT, JYU, JYV, JYW).
with_pool_size_limit(Config, Aggregates_in_memory) ->
    erlang:setelement(5, Config, Aggregates_in_memory).

-spec with_custom_logger(
    signal_config(JZC, JZD, JZE, JZF),
    gleam@erlang@process:subject(telemetry_message())
) -> signal_config(JZC, JZD, JZE, JZF).
with_custom_logger(Config, Logger) ->
    erlang:setelement(6, Config, {some, Logger}).

-spec without_info_logging(signal_config(JZP, JZQ, JZR, JZS)) -> signal_config(JZP, JZQ, JZR, JZS).
without_info_logging(Config) ->
    erlang:setelement(7, Config, false).

-spec without_debug_logging(signal_config(KAB, KAC, KAD, KAE)) -> signal_config(KAB, KAC, KAD, KAE).
without_debug_logging(Config) ->
    erlang:setelement(8, Config, false).

-spec send_and_continue(
    gleam@erlang@process:subject({ok, KDG} | {error, binary()}),
    gleam@option:option(KDG),
    KDL
) -> gleam@otp@actor:next(any(), KDL).
send_and_continue(S, Msg, State) ->
    case Msg of
        {some, M} ->
            gleam@erlang@process:send(S, {ok, M});

        none ->
            gleam@erlang@process:send(S, {error, <<"No logger set"/utf8>>})
    end,
    gleam@otp@actor:continue(State).

-spec context_handler(
    context_message(KCW, KCX, KCY),
    signal_service(KCW, KCX, KCY)
) -> gleam@otp@actor:next(any(), signal_service(KCW, KCX, KCY)).
context_handler(Msg, Cfg) ->
    case Msg of
        {get_logger, S} ->
            send_and_continue(S, erlang:element(2, Cfg), Cfg);

        {get_store, S@1} ->
            send_and_continue(S@1, erlang:element(5, Cfg), Cfg);

        {get_bus, S@2} ->
            send_and_continue(S@2, erlang:element(4, Cfg), Cfg);

        {get_aggregate_pool, S@3} ->
            send_and_continue(S@3, erlang:element(3, Cfg), Cfg);

        {set_logger, S@4} ->
            gleam@otp@actor:continue(erlang:setelement(2, Cfg, {some, S@4}));

        {set_store, S@5} ->
            gleam@otp@actor:continue(erlang:setelement(5, Cfg, {some, S@5}));

        {set_bus, S@6} ->
            gleam@otp@actor:continue(erlang:setelement(4, Cfg, {some, S@6}));

        {set_aggregate_pool, S@7} ->
            gleam@otp@actor:continue(erlang:setelement(3, Cfg, {some, S@7}));

        shutdown_signal ->
            {stop, normal}
    end.

-spec apply_event(
    event(KFG),
    fun((KFI, event(KFG)) -> KFI),
    aggregate_state(KFI, KFL, KFG)
) -> aggregate_state(KFI, KFL, KFG).
apply_event(Event, Handler, Agg) ->
    {aggregate_state,
        erlang:element(2, Agg) + 1,
        Handler(erlang:element(3, Agg), Event)}.

-spec aggregate_init(list(event(KDW)), aggregate_config(KDZ, any(), KDW)) -> fun(() -> gleam@otp@actor:init_result(aggregate_state(KDZ, any(), KDW), any())).
aggregate_init(Events, Cfg) ->
    fun() ->
        Aggregate = {aggregate_state, 0, erlang:element(2, Cfg)},
        {ready,
            begin
                _pipe = Events,
                gleam@list:fold(
                    _pipe,
                    Aggregate,
                    fun(Agg, E) ->
                        apply_event(E, erlang:element(4, Cfg), Agg)
                    end
                )
            end,
            gleam_erlang_ffi:new_selector()}
    end.

-spec in_memory_persistance_handler(store_message(KKK), list(event(KKK))) -> gleam@otp@actor:next(any(), list(event(KKK))).
in_memory_persistance_handler(Message, State) ->
    case Message of
        {get_stored_events, S, Aggregate_id} ->
            case gleam@list:any(
                State,
                fun(E) -> erlang:element(4, E) =:= Aggregate_id end
            ) of
                true ->
                    gleam@erlang@process:send(
                        S,
                        {ok,
                            gleam@list:filter(
                                State,
                                fun(E@1) ->
                                    erlang:element(4, E@1) =:= Aggregate_id
                                end
                            )}
                    );

                false ->
                    gleam@erlang@process:send(
                        S,
                        {error, <<"Aggregate not found"/utf8>>}
                    )
            end,
            gleam@otp@actor:continue(State);

        {is_identity_available, S@1, Aggregate_id@1} ->
            case gleam@list:all(
                State,
                fun(E@2) -> erlang:element(4, E@2) /= Aggregate_id@1 end
            ) of
                true ->
                    gleam@erlang@process:send(S@1, {ok, true});

                false ->
                    gleam@erlang@process:send(S@1, {ok, false})
            end,
            gleam@otp@actor:continue(State);

        {store_event, Event} ->
            gleam@otp@actor:continue(
                begin
                    _pipe = [Event | lists:reverse(State)],
                    lists:reverse(_pipe)
                end
            );

        shutdown_persistance_layer ->
            {stop, normal}
    end.

-spec telemetry_to_string_list(telemetry_event()) -> list(binary()).
telemetry_to_string_list(Ev) ->
    case Ev of
        {pool_creating_aggregate, Id} ->
            [Id];

        {pool_created_aggregate, Id@1} ->
            [Id@1];

        {pool_cannot_create_aggregate_with_id, Id@2} ->
            [Id@2];

        {pool_hydrating_aggregate, Id@3} ->
            [Id@3];

        {pool_hydrated_aggregate, Id@4} ->
            [Id@4];

        {pool_aggregate_not_found, Id@5} ->
            [Id@5];

        {pool_rebalancing_started, Size} ->
            [gleam@int:to_string(Size)];

        {pool_evicted_aggregate, Id@6} ->
            [Id@6];

        {pool_rebalancing_completed, Size@1} ->
            [gleam@int:to_string(Size@1)];

        {aggregate_processing_command, A, B} ->
            [A, B];

        {aggregate_processed_command, A@1, B@1} ->
            [A@1, B@1];

        {aggregate_command_processing_failed, A@2, B@2} ->
            [A@2, B@2];

        {aggregate_events_produced, A@3, B@3} ->
            [A@3, B@3];

        {bus_triggering_subscribers, A@4, B@4} ->
            [A@4, gleam@int:to_string(B@4)];

        {bus_subscribers_informed, A@5, B@5} ->
            [A@5, gleam@int:to_string(B@5)];

        {store_pushed_event_to_write_ahead_log, A@6, B@6} ->
            [A@6, gleam@int:to_string(B@6)];

        {store_write_ahead_log_size_warning, Pool_size} ->
            [gleam@int:to_string(Pool_size)];

        {store_submitted_batch_for_persistance, Batch_size} ->
            [gleam@int:to_string(Batch_size)];

        {store_persistance_completed, A@7, B@7} ->
            [gleam@int:to_string(A@7), gleam@int:to_string(B@7)]
    end.

-spec format_telemetry_message(telemetry_event(), binary()) -> binary().
format_telemetry_message(Data, Template) ->
    _pipe = gleam@list:interleave(
        [gleam@string:split(Template, <<"|"/utf8>>),
            telemetry_to_string_list(Data)]
    ),
    gleam@string:concat(_pipe).

-spec telemetry_log_level(telemetry_event()) -> log_level().
telemetry_log_level(Ev) ->
    case Ev of
        {pool_creating_aggregate, _} ->
            log_debug;

        {pool_created_aggregate, _} ->
            log_debug;

        {pool_cannot_create_aggregate_with_id, _} ->
            log_error;

        {pool_hydrating_aggregate, _} ->
            log_debug;

        {pool_hydrated_aggregate, _} ->
            log_debug;

        {pool_aggregate_not_found, _} ->
            log_error;

        {pool_rebalancing_started, _} ->
            log_debug;

        {pool_evicted_aggregate, _} ->
            log_debug;

        {pool_rebalancing_completed, _} ->
            log_debug;

        {aggregate_processing_command, _, _} ->
            log_info;

        {aggregate_processed_command, _, _} ->
            log_info;

        {aggregate_command_processing_failed, _, _} ->
            log_error;

        {aggregate_events_produced, _, _} ->
            log_debug;

        {bus_triggering_subscribers, _, _} ->
            log_debug;

        {bus_subscribers_informed, _, _} ->
            log_debug;

        {store_pushed_event_to_write_ahead_log, _, _} ->
            log_debug;

        {store_write_ahead_log_size_warning, _} ->
            log_warning;

        {store_submitted_batch_for_persistance, _} ->
            log_debug;

        {store_persistance_completed, _, _} ->
            log_debug
    end.

-spec type_name(any()) -> binary().
type_name(V) ->
    Str = begin
        _pipe = V,
        _pipe@1 = gleam@string:inspect(_pipe),
        gleam@string:split_once(_pipe@1, <<"("/utf8>>)
    end,
    case Str of
        {ok, {Type_string, _}} ->
            Type_string;

        {error, _} ->
            <<"Unknown"/utf8>>
    end.

-spec hydrate_event(KEZ, aggregate_update_context(any(), any(), KEZ)) -> event(KEZ).
hydrate_event(Event, Ctx) ->
    Current_date = begin
        _pipe = birl:utc_now(),
        birl:to_iso8601(_pipe)
    end,
    {event,
        erlang:element(2, erlang:element(5, Ctx)) + 1,
        type_name(erlang:element(3, erlang:element(5, Ctx))),
        erlang:element(3, Ctx),
        type_name(Event),
        Current_date,
        Event}.

-spec console_logger(boolean(), boolean()) -> fun((telemetry_message(), nil) -> gleam@otp@actor:next(any(), nil)).
console_logger(Log_info, Log_debug) ->
    fun(Message, _) -> case Message of
            {report, Event, Template} ->
                Log_level = telemetry_log_level(Event),
                Prefix = <<<<"SIGNAL --- "/utf8, (type_name(Event))/binary>>/binary,
                    " --- "/utf8>>,
                case Log_level of
                    log_error ->
                        gleam@io:print_error(
                            <<Prefix/binary,
                                (format_telemetry_message(Event, Template))/binary>>
                        );

                    log_warning ->
                        gleam@io:println(
                            <<Prefix/binary,
                                (format_telemetry_message(Event, Template))/binary>>
                        );

                    log_info when Log_info ->
                        gleam@io:println(
                            <<Prefix/binary,
                                (format_telemetry_message(Event, Template))/binary>>
                        );

                    log_debug when Log_debug ->
                        gleam@io:println(
                            <<Prefix/binary,
                                (format_telemetry_message(Event, Template))/binary>>
                        );

                    _ ->
                        nil
                end,
                gleam@otp@actor:continue(nil);

            shutdown_telemetry ->
                {stop, normal}
        end end.

-spec aggregate(
    gleam@erlang@process:subject(context_message(KAW, KAX, KAY)),
    binary()
) -> {ok, gleam@erlang@process:subject(aggregate_message(KAW, KAX, KAY))} |
    {error, binary()}.
aggregate(Signal, Id) ->
    gleam@result:'try'(
        gleam@erlang@process:call(
            Signal,
            fun(_capture) -> {get_aggregate_pool, _capture} end,
            100
        ),
        fun(Pool) ->
            gleam@erlang@process:call(
                Pool,
                fun(_capture@1) -> {get_aggregate, _capture@1, Id} end,
                100
            )
        end
    ).

-spec create(
    gleam@erlang@process:subject(context_message(KBI, KBJ, KBK)),
    binary()
) -> {ok, gleam@erlang@process:subject(aggregate_message(KBI, KBJ, KBK))} |
    {error, binary()}.
create(Signal, Id) ->
    gleam@result:'try'(
        gleam@erlang@process:call(
            Signal,
            fun(_capture) -> {get_aggregate_pool, _capture} end,
            100
        ),
        fun(Pool) ->
            gleam@erlang@process:call(
                Pool,
                fun(_capture@1) -> {create_aggregate, _capture@1, Id} end,
                100
            )
        end
    ).

-spec handle_command(
    gleam@erlang@process:subject(aggregate_message(KBU, KBV, any())),
    KBV
) -> {ok, KBU} | {error, binary()}.
handle_command(Agg, Command) ->
    gleam@erlang@process:call(
        Agg,
        fun(_capture) -> {handle_command, _capture, Command} end,
        100
    ).

-spec get_state(
    gleam@erlang@process:subject(aggregate_message(KCC, any(), any()))
) -> KCC.
get_state(Agg) ->
    gleam@erlang@process:call(Agg, fun(_capture) -> {state, _capture} end, 100).

-spec get_id(
    gleam@erlang@process:subject(aggregate_message(any(), any(), any()))
) -> binary().
get_id(Agg) ->
    gleam@erlang@process:call(
        Agg,
        fun(_capture) -> {identity, _capture} end,
        100
    ).

-spec get_current_pool_size(
    gleam@erlang@process:subject(context_message(any(), any(), any()))
) -> {ok, integer()} | {error, binary()}.
get_current_pool_size(Signal) ->
    gleam@result:'try'(
        gleam@erlang@process:call(
            Signal,
            fun(_capture) -> {get_aggregate_pool, _capture} end,
            100
        ),
        fun(Pool) ->
            {ok,
                gleam@erlang@process:call(
                    Pool,
                    fun(_capture@1) -> {pool_size, _capture@1} end,
                    100
                )}
        end
    ).

-spec send_event_to_bus(event(KFS), aggregate_update_context(any(), any(), KFS)) -> event(KFS).
send_event_to_bus(Event, Ctx) ->
    _assert_subject = gleam@erlang@process:call(
        erlang:element(2, Ctx),
        fun(_capture) -> {get_bus, _capture} end,
        100
    ),
    {ok, Bus} = 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 => <<"signal"/utf8>>,
                        function => <<"send_event_to_bus"/utf8>>,
                        line => 681})
    end,
    gleam@erlang@process:send(Bus, {push_event, Event}),
    Event.

-spec update_aggregate(aggregate_update_context(KES, KET, KEU), KEU) -> aggregate_state(KES, KET, KEU).
update_aggregate(Ctx, Event) ->
    _pipe = Event,
    _pipe@1 = hydrate_event(_pipe, Ctx),
    _pipe@2 = send_event_to_bus(_pipe@1, Ctx),
    apply_event(_pipe@2, erlang:element(4, Ctx), erlang:element(5, Ctx)).

-spec store_has_aggregate(
    gleam@erlang@process:subject(context_message(any(), any(), any())),
    binary()
) -> boolean().
store_has_aggregate(Signal, Key) ->
    Response = (gleam@result:'try'(
        gleam@erlang@process:call(
            Signal,
            fun(_capture) -> {get_store, _capture} end,
            100
        ),
        fun(Store) ->
            gleam@erlang@process:call(
                Store,
                fun(_capture@1) -> {is_identity_available, _capture@1, Key} end,
                100
            )
        end
    )),
    case Response of
        {error, _} ->
            true;

        {ok, false} ->
            true;

        {ok, true} ->
            false
    end.

-spec notify_store(
    gleam@erlang@process:subject(context_message(any(), any(), KKD)),
    event(KKD)
) -> nil.
notify_store(Signal, Event) ->
    _assert_subject = gleam@erlang@process:call(
        Signal,
        fun(_capture) -> {get_store, _capture} end,
        100
    ),
    {ok, Store} = 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 => <<"signal"/utf8>>,
                        function => <<"notify_store"/utf8>>,
                        line => 952})
    end,
    gleam@erlang@process:send(Store, {store_event, Event}).

-spec log_telemetry(
    gleam@erlang@process:subject(context_message(any(), any(), any())),
    telemetry_event()
) -> nil.
log_telemetry(Signal, Event) ->
    case gleam@erlang@process:call(
        Signal,
        fun(_capture) -> {get_logger, _capture} end,
        100
    ) of
        {ok, Logger} ->
            gleam@erlang@process:send(Logger, {report, Event, case Event of
                        {pool_creating_aggregate, _} ->
                            <<"Creating aggregate |"/utf8>>;

                        {pool_created_aggregate, _} ->
                            <<"Pool created aggregate |"/utf8>>;

                        {pool_cannot_create_aggregate_with_id, _} ->
                            <<"Cannot create aggregate |, that id is already taken!"/utf8>>;

                        {pool_hydrating_aggregate, _} ->
                            <<"Hydrating aggregate |"/utf8>>;

                        {pool_hydrated_aggregate, _} ->
                            <<"Hydrated aggregate |"/utf8>>;

                        {pool_aggregate_not_found, _} ->
                            <<"Aggregate | not found"/utf8>>;

                        {pool_rebalancing_started, _} ->
                            <<"Rebalancing pool to, from size |"/utf8>>;

                        {pool_evicted_aggregate, _} ->
                            <<"Evicted aggregate |"/utf8>>;

                        {pool_rebalancing_completed, _} ->
                            <<"Rebalancing completed, new size |"/utf8>>;

                        {aggregate_processing_command, _, _} ->
                            <<"Processing command | with aggregate |"/utf8>>;

                        {aggregate_processed_command, _, _} ->
                            <<"Command | processed by aggregate |"/utf8>>;

                        {aggregate_command_processing_failed, _, _} ->
                            <<"Command | processing failed by aggregate |"/utf8>>;

                        {aggregate_events_produced, _, _} ->
                            <<"Events | produced by aggregate: |"/utf8>>;

                        {bus_triggering_subscribers, _, _} ->
                            <<"Sending event | to | subscribers"/utf8>>;

                        {bus_subscribers_informed, _, _} ->
                            <<"Sent event | to | subscribers"/utf8>>;

                        {store_pushed_event_to_write_ahead_log, _, _} ->
                            <<"Pushed event | to write ahead log, wal size |"/utf8>>;

                        {store_write_ahead_log_size_warning, _} ->
                            <<"Wal is overflowing! | events are waiting to be persisted!"/utf8>>;

                        {store_submitted_batch_for_persistance, _} ->
                            <<"Submitted | events for persistance"/utf8>>;

                        {store_persistance_completed, _, _} ->
                            <<"Persisted | events, | events have not yet been reported persisted."/utf8>>
                    end});

        _ ->
            nil
    end.

-spec aggregate_handler(
    gleam@erlang@process:subject(context_message(KEF, KEG, KEH)),
    binary(),
    fun((KEG, KEF) -> {ok, list(KEH)} | {error, binary()}),
    fun((KEF, event(KEH)) -> KEF)
) -> fun((aggregate_message(KEF, KEG, KEH), aggregate_state(KEF, KEG, KEH)) -> gleam@otp@actor:next(any(), aggregate_state(KEF, KEG, KEH))).
aggregate_handler(Signal, Id, Command_handler, Event_handler) ->
    fun(Operation, Agg) -> case Operation of
            {state, Client} ->
                gleam@erlang@process:send(Client, erlang:element(3, Agg)),
                gleam@otp@actor:continue(Agg);

            {identity, Client@1} ->
                gleam@erlang@process:send(Client@1, Id),
                gleam@otp@actor:continue(Agg);

            {handle_command, Client@2, Command} ->
                log_telemetry(
                    Signal,
                    {aggregate_processing_command, type_name(Command), Id}
                ),
                case Command_handler(Command, erlang:element(3, Agg)) of
                    {ok, Events} ->
                        New_state = gleam@list:fold(
                            Events,
                            Agg,
                            fun(Agg@1, E) ->
                                update_aggregate(
                                    {aggregate_update_context,
                                        Signal,
                                        Id,
                                        Event_handler,
                                        Agg@1},
                                    E
                                )
                            end
                        ),
                        log_telemetry(
                            Signal,
                            {aggregate_processed_command,
                                type_name(Command),
                                Id}
                        ),
                        case Events of
                            [] ->
                                gleam@list:new();

                            E@1 ->
                                gleam@list:map(
                                    E@1,
                                    fun(Ev) ->
                                        log_telemetry(
                                            Signal,
                                            {aggregate_events_produced,
                                                type_name(Ev),
                                                Id}
                                        )
                                    end
                                )
                        end,
                        gleam@erlang@process:send(
                            Client@2,
                            {ok, erlang:element(3, New_state)}
                        ),
                        gleam@otp@actor:continue(New_state);

                    {error, Msg} ->
                        log_telemetry(
                            Signal,
                            {aggregate_command_processing_failed,
                                type_name(Command),
                                Id}
                        ),
                        gleam@erlang@process:send(Client@2, {error, Msg}),
                        gleam@otp@actor:continue(Agg)
                end;

            shutdown_aggregate ->
                {stop, normal}
        end end.

-spec evict_aggregates_workflow(
    gleam@erlang@process:subject(context_message(KGL, KGM, KGN)),
    integer(),
    gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(KGL, KGM, KGN)),
        integer()})
) -> gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(KGL, KGM, KGN)),
    integer()}).
evict_aggregates_workflow(Signal, Max_size, State) ->
    case maps:size(State) >= Max_size of
        true ->
            log_telemetry(Signal, {pool_rebalancing_started, Max_size}),
            Eviction_list = begin
                _pipe = maps:to_list(State),
                _pipe@1 = gleam@list:filter(
                    _pipe,
                    fun(Agg) ->
                        {_, {_, Position}} = Agg,
                        Position >= Max_size
                    end
                ),
                gleam@list:map(
                    _pipe@1,
                    fun(Agg@1) ->
                        {Id, {Agg@2, _}} = Agg@1,
                        gleam@erlang@process:send(Agg@2, shutdown_aggregate),
                        log_telemetry(Signal, {pool_evicted_aggregate, Id}),
                        Id
                    end
                )
            end,
            log_telemetry(
                Signal,
                {pool_rebalancing_completed, erlang:length(Eviction_list)}
            ),
            gleam@dict:drop(State, Eviction_list);

        false ->
            State
    end.

-spec gather_aggregate_from_store(
    gleam@erlang@process:subject(context_message(KIA, KIB, KIC)),
    fun((list(event(KIC))) -> {ok,
            gleam@erlang@process:subject(aggregate_message(KIA, KIB, KIC))} |
        {error, binary()}),
    binary()
) -> {ok, gleam@erlang@process:subject(aggregate_message(KIA, KIB, KIC))} |
    {error, binary()}.
gather_aggregate_from_store(Signal, Aggregate_initializer, Id) ->
    log_telemetry(Signal, {pool_hydrating_aggregate, Id}),
    Result = (gleam@result:'try'(
        gleam@erlang@process:call(
            Signal,
            fun(_capture) -> {get_store, _capture} end,
            100
        ),
        fun(Store) ->
            gleam@erlang@process:call(
                Store,
                fun(_capture@1) -> {get_stored_events, _capture@1, Id} end,
                100
            )
        end
    )),
    case Result of
        {error, Msg} ->
            {error, Msg};

        {ok, Events} ->
            log_telemetry(Signal, {pool_hydrated_aggregate, Id}),
            Aggregate_initializer(Events)
    end.

-spec gather_aggregate(
    gleam@erlang@process:subject(context_message(KHG, KHH, KHI)),
    fun((list(event(KHI))) -> {ok,
            gleam@erlang@process:subject(aggregate_message(KHG, KHH, KHI))} |
        {error, binary()}),
    gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(KHG, KHH, KHI)),
        integer()}),
    binary()
) -> {ok, gleam@erlang@process:subject(aggregate_message(KHG, KHH, KHI))} |
    {error, binary()}.
gather_aggregate(Signal, Aggregate_initializer, Dict, Id) ->
    case gleam@dict:get(Dict, Id) of
        {ok, {Value, _}} ->
            {ok, Value};

        {error, _} ->
            gather_aggregate_from_store(Signal, Aggregate_initializer, Id)
    end.

-spec start_aggregate(
    gleam@erlang@process:subject(context_message(KIP, KIQ, KIR)),
    aggregate_config(KIP, KIQ, KIR),
    binary(),
    list(event(KIR))
) -> {ok, gleam@erlang@process:subject(aggregate_message(KIP, KIQ, KIR))} |
    {error, binary()}.
start_aggregate(Signal, Config, Id, Events) ->
    case gleam@otp@actor:start_spec(
        {spec,
            aggregate_init(Events, Config),
            5,
            aggregate_handler(
                Signal,
                Id,
                erlang:element(3, Config),
                erlang:element(4, Config)
            )}
    ) of
        {ok, Actor} ->
            {ok, Actor};

        _ ->
            {error, <<"Could not initialize the actor: "/utf8, Id/binary>>}
    end.

-spec pool_handler(
    gleam@erlang@process:subject(context_message(KGA, KGB, KGC)),
    aggregate_config(KGA, KGB, KGC),
    integer()
) -> fun((pool_message(KGA, KGB, KGC), gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(KGA, KGB, KGC)),
    integer()})) -> gleam@otp@actor:next(any(), gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(KGA, KGB, KGC)),
    integer()}))).
pool_handler(Signal, Config, Max_size) ->
    fun(Operation, State) -> case Operation of
            {create_aggregate, Client, With_id} ->
                log_telemetry(Signal, {pool_creating_aggregate, With_id}),
                Exists_in_store = store_has_aggregate(Signal, With_id),
                Exists_in_pool = gleam@dict:has_key(State, With_id),
                case {Exists_in_store, Exists_in_pool} of
                    {false, false} ->
                        case start_aggregate(Signal, Config, With_id, []) of
                            {ok, Agg} ->
                                gleam@erlang@process:send(Client, {ok, Agg}),
                                log_telemetry(
                                    Signal,
                                    {pool_created_aggregate, With_id}
                                ),
                                gleam@otp@actor:continue(
                                    gleam@dict:insert(
                                        evict_aggregates_workflow(
                                            Signal,
                                            Max_size,
                                            State
                                        ),
                                        With_id,
                                        {Agg, maps:size(State) + 1}
                                    )
                                );

                            Error ->
                                gleam@erlang@process:send(Client, Error),
                                gleam@otp@actor:continue(State)
                        end;

                    {_, _} ->
                        log_telemetry(
                            Signal,
                            {pool_cannot_create_aggregate_with_id, With_id}
                        ),
                        gleam@erlang@process:send(
                            Client,
                            {error,
                                <<<<"Aggregate ID must be unique, "/utf8,
                                        With_id/binary>>/binary,
                                    " is not!"/utf8>>}
                        ),
                        gleam@otp@actor:continue(State)
                end;

            {get_aggregate, Client@1, Id} ->
                case gather_aggregate(
                    Signal,
                    fun(_capture) ->
                        start_aggregate(Signal, Config, Id, _capture)
                    end,
                    State,
                    Id
                ) of
                    {ok, Aggregate} ->
                        gleam@dict:insert(
                            State,
                            Id,
                            {Aggregate, maps:size(State) + 1}
                        ),
                        gleam@erlang@process:send(Client@1, {ok, Aggregate});

                    {error, Msg} ->
                        log_telemetry(Signal, {pool_aggregate_not_found, Id}),
                        gleam@erlang@process:send(Client@1, {error, Msg})
                end,
                gleam@otp@actor:continue(State);

            {pool_size, S} ->
                gleam@erlang@process:send(
                    S,
                    begin
                        _pipe = maps:to_list(State),
                        erlang:length(_pipe)
                    end
                ),
                gleam@otp@actor:continue(State);

            shutdown_pool ->
                {stop, normal}
        end end.

-spec notify_subscribers(
    gleam@erlang@process:subject(context_message(any(), any(), KJQ)),
    event(KJQ),
    list(subscriber(any(), KJQ))
) -> nil.
notify_subscribers(Signal, Event, Consumers) ->
    log_telemetry(
        Signal,
        {bus_triggering_subscribers,
            erlang:element(5, Event),
            erlang:length(Consumers)}
    ),
    case Consumers of
        [] ->
            nil;

        [{consumer, S}] ->
            gleam@erlang@process:send(S, {consume, Event});

        [{consumer, S@1} | Rest] ->
            gleam@erlang@process:send(S@1, {consume, Event}),
            notify_subscribers(Signal, Event, Rest)
    end,
    log_telemetry(
        Signal,
        {bus_subscribers_informed,
            erlang:element(5, Event),
            erlang:length(Consumers)}
    ).

-spec bus_handler(
    gleam@erlang@process:subject(context_message(any(), any(), KJE)),
    list(subscriber(any(), KJE))
) -> fun((bus_message(KJE), nil) -> gleam@otp@actor:next(any(), nil)).
bus_handler(Signal, Subscribers) ->
    fun(Message, _) -> case Message of
            {push_event, Event} ->
                notify_store(Signal, Event),
                notify_subscribers(Signal, Event, Subscribers),
                gleam@otp@actor:continue(nil);

            shutdown_bus ->
                {stop, normal}
        end end.

-spec signal_init(signal_config(KDN, any(), KDP, KDQ)) -> {ok,
        gleam@erlang@process:subject(context_message(KDN, KDP, KDQ))} |
    {error, gleam@otp@actor:start_error()}.
signal_init(Config) ->
    gleam@result:'try'(
        gleam@otp@actor:start(
            {signal_service, none, none, none, none},
            fun context_handler/2
        ),
        fun(Signal) -> gleam@result:'try'(case erlang:element(6, Config) of
                    {some, Logger} ->
                        {ok, Logger};

                    none ->
                        gleam@otp@actor:start(
                            nil,
                            console_logger(
                                erlang:element(7, Config),
                                erlang:element(8, Config)
                            )
                        )
                end, fun(Logger@1) ->
                    gleam@otp@actor:send(Signal, {set_logger, Logger@1}),
                    Store@2 = (case erlang:element(3, Config) of
                        {some, Store} ->
                            Store;

                        none ->
                            _assert_subject = gleam@otp@actor:start(
                                [],
                                fun in_memory_persistance_handler/2
                            ),
                            {ok, Store@1} = 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 => <<"signal"/utf8>>,
                                                function => <<"signal_init"/utf8>>,
                                                line => 511})
                            end,
                            Store@1
                    end),
                    gleam@otp@actor:send(Signal, {set_store, Store@2}),
                    gleam@result:'try'(
                        gleam@otp@actor:start(
                            nil,
                            bus_handler(Signal, erlang:element(4, Config))
                        ),
                        fun(Bus) ->
                            gleam@otp@actor:send(Signal, {set_bus, Bus}),
                            gleam@result:'try'(
                                gleam@otp@actor:start(
                                    gleam@dict:new(),
                                    pool_handler(
                                        Signal,
                                        erlang:element(2, Config),
                                        erlang:element(5, Config)
                                    )
                                ),
                                fun(Pool) ->
                                    gleam@otp@actor:send(
                                        Signal,
                                        {set_aggregate_pool, Pool}
                                    ),
                                    {ok, Signal}
                                end
                            )
                        end
                    )
                end) end
    ).

-spec start(signal_config(KAN, any(), KAP, KAQ)) -> {ok,
        gleam@erlang@process:subject(context_message(KAN, KAP, KAQ))} |
    {error, gleam@otp@actor:start_error()}.
start(Config) ->
    signal_init(Config).