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([signal_message/3, event/1, persistance_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, store_message/1]).

-opaque signal_message(IGF, IGG, IGH) :: {get_pool,
        gleam@erlang@process:subject(gleam@erlang@process:subject(pool_message(IGF, IGG, IGH)))} |
    shutdown.

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

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

-type subscriber(IGK, IGL) :: {consumer,
        gleam@erlang@process:subject(consumer_message(IGK, IGL))} |
    {policy, gleam@otp@task:task(event(IGL))}.

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

-opaque signal_config(IGO, IGP, IGQ, IGR) :: {signal_config,
        aggregate_config(IGO, IGQ, IGR),
        gleam@option:option(gleam@erlang@process:subject(persistance_message(IGR))),
        list(subscriber(IGP, IGR)),
        integer(),
        gleam@option:option(gleam@erlang@process:subject(telemetry_message())),
        boolean(),
        boolean()}.

-type aggregate_config(IGS, IGT, IGU) :: {aggregate_config,
        IGS,
        fun((IGT, IGS) -> {ok, list(IGU)} | {error, binary()}),
        fun((IGS, event(IGU)) -> IGS)}.

-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(IGV, IGW, IGX) :: {signal_service,
        gleam@erlang@process:subject(pool_message(IGV, IGW, IGX)),
        gleam@erlang@process:subject(bus_message(IGX)),
        gleam@erlang@process:subject(store_message(IGX))}.

-type aggregate_state(IGY, IGZ, IHA) :: {aggregate_state, integer(), IGY} |
    {gleam_phantom, IGZ, IHA}.

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

-type aggregate_update_context(IHE, IHF, IHG) :: {aggregate_update_context,
        binary(),
        gleam@erlang@process:subject(bus_message(IHG)),
        fun((IHE, event(IHG)) -> IHE),
        aggregate_state(IHE, IHF, IHG)}.

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

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

-type store_message(IHL) :: {store_event, event(IHL)} |
    {get_events,
        gleam@erlang@process:subject({ok, list(event(IHL))} | {error, binary()}),
        binary()} |
    {id_exists,
        gleam@erlang@process:subject({ok, boolean()} | {error, binary()}),
        binary()} |
    {persistance_state, list(event(IHL))} |
    shutdown_store.

-spec configure(aggregate_config(IIW, IIX, IIY)) -> signal_config(IIW, any(), IIX, IIY).
configure(Agg) ->
    {signal_config, Agg, none, [], 100, none, true, true}.

-spec with_subscriber(signal_config(IJH, IJI, IJJ, IJK), subscriber(IJI, IJK)) -> signal_config(IJH, IJI, IJJ, IJK).
with_subscriber(Config, Sub) ->
    erlang:setelement(4, Config, [Sub | erlang:element(4, Config)]).

-spec with_persistance_layer(
    signal_config(IJV, IJW, IJX, IJY),
    gleam@erlang@process:subject(persistance_message(IJY))
) -> signal_config(IJV, IJW, IJX, IJY).
with_persistance_layer(Config, Persist) ->
    erlang:setelement(3, Config, {some, Persist}).

-spec with_pool_size_limit(signal_config(IKJ, IKK, IKL, IKM), integer()) -> signal_config(IKJ, IKK, IKL, IKM).
with_pool_size_limit(Config, Aggregates_in_memory) ->
    erlang:setelement(5, Config, Aggregates_in_memory).

-spec with_custom_logger(
    signal_config(IKS, IKT, IKU, IKV),
    gleam@erlang@process:subject(telemetry_message())
) -> signal_config(IKS, IKT, IKU, IKV).
with_custom_logger(Config, Logger) ->
    erlang:setelement(6, Config, {some, Logger}).

-spec without_info_logging(signal_config(ILF, ILG, ILH, ILI)) -> signal_config(ILF, ILG, ILH, ILI).
without_info_logging(Config) ->
    erlang:setelement(7, Config, false).

-spec without_debug_logging(signal_config(ILR, ILS, ILT, ILU)) -> signal_config(ILR, ILS, ILT, ILU).
without_debug_logging(Config) ->
    erlang:setelement(8, Config, false).

-spec emit_handler(signal_service(IOR, IOS, IOT)) -> fun((signal_message(IOR, IOS, IOT), nil) -> gleam@otp@actor:next(any(), nil)).
emit_handler(Cfg) ->
    fun(Message, _) -> case Message of
            {get_pool, S} ->
                gleam@erlang@process:send(S, erlang:element(2, Cfg)),
                gleam@otp@actor:continue(nil);

            shutdown ->
                {stop, normal}
        end end.

-spec apply_event(
    event(IQG),
    fun((IQI, event(IQG)) -> IQI),
    aggregate_state(IQI, IQL, IQG)
) -> aggregate_state(IQI, IQL, IQG).
apply_event(Event, Handler, Agg) ->
    {aggregate_state,
        erlang:element(2, Agg) + 1,
        Handler(erlang:element(3, Agg), Event)}.

-spec aggregate_init(list(event(IOY)), aggregate_config(IPB, any(), IOY)) -> fun(() -> gleam@otp@actor:init_result(aggregate_state(IPB, any(), IOY), 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 send_event_to_bus(event(IQS), aggregate_update_context(any(), any(), IQS)) -> event(IQS).
send_event_to_bus(Event, Ctx) ->
    gleam@erlang@process:send(erlang:element(3, Ctx), {push_event, Event}),
    Event.

-spec notify_store(event(IUD), gleam@erlang@process:subject(store_message(IUD))) -> nil.
notify_store(Event, Store) ->
    gleam@erlang@process:send(Store, {store_event, Event}).

-spec in_memory_persistance_handler(persistance_message(IUS), list(event(IUS))) -> gleam@otp@actor:next(any(), list(event(IUS))).
in_memory_persistance_handler(Message, State) ->
    case Message of
        {get_stored_events, S, Aggregate_id} ->
            case gleam@list:any(
                State,
                fun(E) -> erlang:element(3, E) =:= Aggregate_id end
            ) of
                true ->
                    gleam@erlang@process:send(
                        S,
                        {ok,
                            gleam@list:filter(
                                State,
                                fun(E@1) ->
                                    erlang:element(3, 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(3, 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_events, Events} ->
            gleam@otp@actor:continue(lists:append(State, Events));

        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 log_telemetry(
    gleam@erlang@process:subject(telemetry_message()),
    telemetry_event()
) -> nil.
log_telemetry(Logger, Event) ->
    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}).

-spec evict_aggregates_workflow(
    integer(),
    gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(IRK, IRL, IRM)),
        integer()}),
    gleam@erlang@process:subject(telemetry_message())
) -> gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(IRK, IRL, IRM)),
    integer()}).
evict_aggregates_workflow(Max_size, State, Logger) ->
    case maps:size(State) >= Max_size of
        true ->
            log_telemetry(Logger, {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(Logger, {pool_evicted_aggregate, Id}),
                        Id
                    end
                )
            end,
            log_telemetry(
                Logger,
                {pool_rebalancing_completed, erlang:length(Eviction_list)}
            ),
            gleam@dict:drop(State, Eviction_list);

        false ->
            State
    end.

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

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

        [{policy, T}] ->
            _ = gleam@otp@task:try_await(T, 5),
            nil;

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

        [{policy, T@1} | Rest@1] ->
            _ = gleam@otp@task:try_await(T@1, 5),
            notify_subscribers(Event, Rest@1, Logger)
    end,
    log_telemetry(
        Logger,
        {bus_subscribers_informed,
            erlang:element(4, Event),
            erlang:length(Consumers)}
    ).

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

            shutdown_bus ->
                {stop, normal}
        end 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(IPZ, aggregate_update_context(any(), any(), IPZ)) -> event(IPZ).
hydrate_event(Event, Ctx) ->
    {event,
        erlang:element(2, erlang:element(5, Ctx)) + 1,
        erlang:element(2, Ctx),
        type_name(Event),
        Event}.

-spec update_aggregate(aggregate_update_context(IPS, IPT, IPU), IPU) -> aggregate_state(IPS, IPT, IPU).
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 aggregate_handler(
    binary(),
    fun((IPI, IPH) -> {ok, list(IPJ)} | {error, binary()}),
    fun((IPH, event(IPJ)) -> IPH),
    gleam@erlang@process:subject(bus_message(IPJ)),
    gleam@erlang@process:subject(telemetry_message())
) -> fun((aggregate_message(IPH, IPI, IPJ), aggregate_state(IPH, IPI, IPJ)) -> gleam@otp@actor:next(any(), aggregate_state(IPH, IPI, IPJ))).
aggregate_handler(Id, Command_handler, Event_handler, Bus, Logger) ->
    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(
                    Logger,
                    {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,
                                        Id,
                                        Bus,
                                        Event_handler,
                                        Agg@1},
                                    E
                                )
                            end
                        ),
                        log_telemetry(
                            Logger,
                            {aggregate_processed_command,
                                type_name(Command),
                                Id}
                        ),
                        case Events of
                            [] ->
                                gleam@list:new();

                            E@1 ->
                                gleam@list:map(
                                    E@1,
                                    fun(Ev) ->
                                        log_telemetry(
                                            Logger,
                                            {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(
                            Logger,
                            {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 start_aggregate(
    aggregate_config(ITC, ITD, ITE),
    gleam@erlang@process:subject(bus_message(ITE)),
    gleam@erlang@process:subject(telemetry_message()),
    binary(),
    list(event(ITE))
) -> {ok, gleam@erlang@process:subject(aggregate_message(ITC, ITD, ITE))} |
    {error, binary()}.
start_aggregate(Config, Bus, Logger, Id, Events) ->
    case gleam@otp@actor:start_spec(
        {spec,
            aggregate_init(Events, Config),
            5,
            aggregate_handler(
                Id,
                erlang:element(3, Config),
                erlang:element(4, Config),
                Bus,
                Logger
            )}
    ) of
        {ok, Actor} ->
            {ok, Actor};

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

-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(signal_message(IMM, IMN, IMO)),
    binary()
) -> {ok, gleam@erlang@process:subject(aggregate_message(IMM, IMN, IMO))} |
    {error, binary()}.
aggregate(Signal, Id) ->
    Pool = gleam@erlang@process:call(
        Signal,
        fun(Field@0) -> {get_pool, Field@0} end,
        100
    ),
    gleam@erlang@process:call(
        Pool,
        fun(_capture) -> {get_aggregate, _capture, Id} end,
        100
    ).

-spec create(
    gleam@erlang@process:subject(signal_message(IMX, IMY, IMZ)),
    binary()
) -> {ok, gleam@erlang@process:subject(aggregate_message(IMX, IMY, IMZ))} |
    {error, binary()}.
create(Signal, Id) ->
    Pool = gleam@erlang@process:call(
        Signal,
        fun(Field@0) -> {get_pool, Field@0} end,
        100
    ),
    gleam@erlang@process:call(
        Pool,
        fun(_capture) -> {create_aggregate, _capture, Id} end,
        100
    ).

-spec handle_command(
    gleam@erlang@process:subject(aggregate_message(INI, INJ, any())),
    INJ
) -> {ok, INI} | {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(INQ, any(), any()))
) -> INQ.
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(signal_message(any(), any(), any()))
) -> integer().
get_current_pool_size(Signal) ->
    Pool = gleam@erlang@process:call(
        Signal,
        fun(_capture) -> {get_pool, _capture} end,
        100
    ),
    gleam@erlang@process:call(
        Pool,
        fun(_capture@1) -> {pool_size, _capture@1} end,
        100
    ).

-spec store_has_aggregate(
    gleam@erlang@process:subject(store_message(any())),
    binary()
) -> boolean().
store_has_aggregate(Store, Key) ->
    Response = gleam@erlang@process:call(
        Store,
        fun(_capture) -> {id_exists, _capture, Key} end,
        100
    ),
    case Response of
        {error, _} ->
            false;

        {ok, false} ->
            false;

        {ok, true} ->
            true
    end.

-spec gather_aggregate_from_store(
    fun((list(event(ISP))) -> {ok,
            gleam@erlang@process:subject(aggregate_message(ISS, IST, ISP))} |
        {error, binary()}),
    gleam@erlang@process:subject(store_message(ISP)),
    binary(),
    gleam@erlang@process:subject(telemetry_message())
) -> {ok, gleam@erlang@process:subject(aggregate_message(ISS, IST, ISP))} |
    {error, binary()}.
gather_aggregate_from_store(Aggregate_initializer, Store, Id, Logger) ->
    log_telemetry(Logger, {pool_hydrating_aggregate, Id}),
    case gleam@erlang@process:call(
        Store,
        fun(_capture) -> {get_events, _capture, Id} end,
        100
    ) of
        {error, Msg} ->
            {error, Msg};

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

-spec gather_aggregate(
    fun((list(event(IRX))) -> {ok,
            gleam@erlang@process:subject(aggregate_message(ISA, ISB, IRX))} |
        {error, binary()}),
    gleam@erlang@process:subject(store_message(IRX)),
    gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(ISA, ISB, IRX)),
        integer()}),
    binary(),
    gleam@erlang@process:subject(telemetry_message())
) -> {ok, gleam@erlang@process:subject(aggregate_message(ISA, ISB, IRX))} |
    {error, binary()}.
gather_aggregate(Aggregate_initializer, Store, Dict, Id, Logger) ->
    case gleam@dict:get(Dict, Id) of
        {ok, {Value, _}} ->
            {ok, Value};

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

-spec pool_handler(
    aggregate_config(IRA, IRB, IRC),
    gleam@erlang@process:subject(bus_message(IRC)),
    gleam@erlang@process:subject(telemetry_message()),
    gleam@erlang@process:subject(store_message(IRC)),
    integer()
) -> fun((pool_message(IRA, IRB, IRC), gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(IRA, IRB, IRC)),
    integer()})) -> gleam@otp@actor:next(any(), gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(IRA, IRB, IRC)),
    integer()}))).
pool_handler(Config, Bus, Logger, Store, Max_size) ->
    fun(Operation, State) -> case Operation of
            {create_aggregate, Client, With_id} ->
                log_telemetry(Logger, {pool_creating_aggregate, With_id}),
                Exists_in_store = store_has_aggregate(Store, 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(Config, Bus, Logger, With_id, []) of
                            {ok, Agg} ->
                                gleam@erlang@process:send(Client, {ok, Agg}),
                                log_telemetry(
                                    Logger,
                                    {pool_created_aggregate, With_id}
                                ),
                                gleam@otp@actor:continue(
                                    gleam@dict:insert(
                                        evict_aggregates_workflow(
                                            Max_size,
                                            State,
                                            Logger
                                        ),
                                        With_id,
                                        {Agg, maps:size(State) + 1}
                                    )
                                );

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

                    {_, _} ->
                        log_telemetry(
                            Logger,
                            {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(
                    fun(_capture) ->
                        start_aggregate(Config, Bus, Logger, Id, _capture)
                    end,
                    Store,
                    State,
                    Id,
                    Logger
                ) 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(Logger, {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 store_handler(
    gleam@erlang@process:subject(persistance_message(IUN)),
    gleam@erlang@process:subject(telemetry_message())
) -> fun((store_message(IUN), {list(event(IUN)), list(event(IUN))}) -> gleam@otp@actor:next(any(), {list(event(IUN)),
    list(event(IUN))})).
store_handler(Persistor, Logger) ->
    fun(Message, State) ->
        {Wal, Processing} = State,
        case erlang:length(Wal) of
            L when L > 100 ->
                log_telemetry(Logger, {store_write_ahead_log_size_warning, L});

            _ ->
                nil
        end,
        case Message of
            {store_event, E} ->
                log_telemetry(Logger, {store_persistance_completed, 1, 0}),
                gleam@otp@actor:send(Persistor, {store_events, [E]}),
                gleam@otp@actor:continue({[], []});

            {get_events, S, Id} ->
                Events = gleam@erlang@process:call(
                    Persistor,
                    fun(_capture) -> {get_stored_events, _capture, Id} end,
                    100
                ),
                gleam@erlang@process:send(S, Events),
                gleam@otp@actor:continue(State);

            {id_exists, S@1, Id@1} ->
                Result = gleam@erlang@process:call(
                    Persistor,
                    fun(_capture@1) ->
                        {is_identity_available, _capture@1, Id@1}
                    end,
                    100
                ),
                gleam@erlang@process:send(S@1, Result),
                gleam@otp@actor:continue(State);

            {persistance_state, Processed} ->
                gleam@otp@actor:continue({[], []});

            shutdown_store ->
                {stop, normal}
        end
    end.

-spec set_up_store_handler(
    gleam@erlang@process:subject(telemetry_message()),
    gleam@option:option(gleam@erlang@process:subject(persistance_message(IUI)))
) -> {ok,
        fun((store_message(IUI), {list(event(IUI)), list(event(IUI))}) -> gleam@otp@actor:next(any(), {list(event(IUI)),
            list(event(IUI))}))} |
    {error, binary()}.
set_up_store_handler(Logger, Persistor) ->
    case Persistor of
        {some, P} ->
            {ok, store_handler(P, Logger)};

        none ->
            case gleam@otp@actor:start([], fun in_memory_persistance_handler/2) of
                {ok, S} ->
                    {ok, store_handler(S, Logger)};

                _ ->
                    {error,
                        <<"Failed to spin up the in memory persistance layer!"/utf8>>}
            end
    end.

-spec signal_init(signal_config(IOI, any(), IOK, IOL)) -> {ok,
        signal_service(IOI, IOK, IOL)} |
    {error, gleam@otp@actor:start_error()}.
signal_init(Config) ->
    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@result:'try'(
                gleam@result:replace_error(
                    set_up_store_handler(Logger@1, erlang:element(3, Config)),
                    init_timeout
                ),
                fun(Store_handler) ->
                    gleam@result:'try'(
                        gleam@otp@actor:start({[], []}, Store_handler),
                        fun(Store) ->
                            gleam@result:'try'(
                                gleam@otp@actor:start(
                                    nil,
                                    bus_handler(
                                        Logger@1,
                                        erlang:element(4, Config),
                                        Store
                                    )
                                ),
                                fun(Bus) ->
                                    gleam@result:'try'(
                                        gleam@otp@actor:start(
                                            gleam@dict:new(),
                                            pool_handler(
                                                erlang:element(2, Config),
                                                Bus,
                                                Logger@1,
                                                Store,
                                                erlang:element(5, Config)
                                            )
                                        ),
                                        fun(Pool) ->
                                            {ok,
                                                {signal_service,
                                                    Pool,
                                                    Bus,
                                                    Store}}
                                        end
                                    )
                                end
                            )
                        end
                    )
                end
            )
        end).

-spec start(signal_config(IMD, any(), IMF, IMG)) -> {ok,
        gleam@erlang@process:subject(signal_message(IMD, IMF, IMG))} |
    {error, gleam@otp@actor:start_error()}.
start(Config) ->
    gleam@result:'try'(
        signal_init(Config),
        fun(Service) -> gleam@otp@actor:start(nil, emit_handler(Service)) end
    ).