-module(eventsourcing_glyn).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/eventsourcing_glyn.gleam").
-export([get_glyn_store/1, memory_store_event_envelop_decoder/1, metadata_decoder/0, serialized_event_envelop_decoder/1, payload_decoder/1, supervised/4, subscribers/1, publish/3]).
-export_type([glyn_store/6, glyn_config/0, glyn_instances/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 glyn_store(IMP, IMQ, IMR, IMS, IMT, IMU) :: {glyn_store,
        binary(),
        binary(),
        glyn_instances(IMS),
        eventsourcing:event_store(IMP, IMQ, IMR, IMS, IMT, IMU)}.

-type glyn_config() :: {glyn_config, binary(), binary()}.

-type glyn_instances(IMV) :: {glyn_instances,
        glyn@pubsub:pub_sub(eventsourcing:query_message(IMV))}.

-file("src/eventsourcing_glyn.gleam", 256).
?DOC(false).
-spec get_glyn_store(
    eventsourcing:event_store(glyn_store(IOC, IOD, IOE, IOF, IOG, IOH), IOD, IOE, IOF, IOG, IOH)
) -> glyn_store(IOC, IOD, IOE, IOF, IOG, IOH).
get_glyn_store(Eventstore) ->
    erlang:element(10, Eventstore).

-file("src/eventsourcing_glyn.gleam", 306).
?DOC(false).
-spec memory_store_event_envelop_decoder(gleam@dynamic@decode:decoder(IPE)) -> gleam@dynamic@decode:decoder(eventsourcing:event_envelop(IPE)).
memory_store_event_envelop_decoder(Event_decoder) ->
    gleam@dynamic@decode:field(
        1,
        {decoder, fun gleam@dynamic@decode:decode_string/1},
        fun(Aggregate_id) ->
            gleam@dynamic@decode:field(
                2,
                {decoder, fun gleam@dynamic@decode:decode_int/1},
                fun(Sequence) ->
                    gleam@dynamic@decode:field(
                        3,
                        Event_decoder,
                        fun(Payload) ->
                            gleam@dynamic@decode:success(
                                {memory_store_event_envelop,
                                    Aggregate_id,
                                    Sequence,
                                    Payload,
                                    []}
                            )
                        end
                    )
                end
            )
        end
    ).

-file("src/eventsourcing_glyn.gleam", 332).
?DOC(false).
-spec metadata_decoder() -> gleam@dynamic@decode:decoder({binary(), binary()}).
metadata_decoder() ->
    gleam@dynamic@decode:field(
        0,
        {decoder, fun gleam@dynamic@decode:decode_string/1},
        fun(Key) ->
            gleam@dynamic@decode:field(
                1,
                {decoder, fun gleam@dynamic@decode:decode_string/1},
                fun(Value) -> gleam@dynamic@decode:success({Key, Value}) end
            )
        end
    ).

-file("src/eventsourcing_glyn.gleam", 284).
?DOC(false).
-spec serialized_event_envelop_decoder(gleam@dynamic@decode:decoder(IPA)) -> gleam@dynamic@decode:decoder(eventsourcing:event_envelop(IPA)).
serialized_event_envelop_decoder(Event_decoder) ->
    gleam@dynamic@decode:field(
        1,
        {decoder, fun gleam@dynamic@decode:decode_string/1},
        fun(Aggregate_id) ->
            gleam@dynamic@decode:field(
                2,
                {decoder, fun gleam@dynamic@decode:decode_int/1},
                fun(Sequence) ->
                    gleam@dynamic@decode:field(
                        3,
                        Event_decoder,
                        fun(Payload) ->
                            gleam@dynamic@decode:field(
                                4,
                                gleam@dynamic@decode:list(metadata_decoder()),
                                fun(Metadata) ->
                                    gleam@dynamic@decode:field(
                                        5,
                                        {decoder,
                                            fun gleam@dynamic@decode:decode_string/1},
                                        fun(Event_type) ->
                                            gleam@dynamic@decode:field(
                                                6,
                                                {decoder,
                                                    fun gleam@dynamic@decode:decode_string/1},
                                                fun(Event_version) ->
                                                    gleam@dynamic@decode:field(
                                                        7,
                                                        {decoder,
                                                            fun gleam@dynamic@decode:decode_string/1},
                                                        fun(Aggregate_type) ->
                                                            gleam@dynamic@decode:success(
                                                                {serialized_event_envelop,
                                                                    Aggregate_id,
                                                                    Sequence,
                                                                    Payload,
                                                                    Metadata,
                                                                    Event_type,
                                                                    Event_version,
                                                                    Aggregate_type}
                                                            )
                                                        end
                                                    )
                                                end
                                            )
                                        end
                                    )
                                end
                            )
                        end
                    )
                end
            )
        end
    ).

-file("src/eventsourcing_glyn.gleam", 323).
?DOC(false).
-spec payload_decoder(gleam@dynamic@decode:decoder(IPI)) -> gleam@dynamic@decode:decoder(eventsourcing:event_envelop(IPI)).
payload_decoder(Event_decoder) ->
    gleam@dynamic@decode:one_of(
        memory_store_event_envelop_decoder(Event_decoder),
        [serialized_event_envelop_decoder(Event_decoder)]
    ).

-file("src/eventsourcing_glyn.gleam", 367).
?DOC(" Publish an event to Glyn PubSub for distribution to all query subscribers\n").
-spec publish_to_glyn_pubsub(
    glyn_store(any(), any(), any(), IQN, any(), any()),
    binary(),
    list(eventsourcing:event_envelop(IQN))
) -> {ok, integer()} | {error, glyn@pubsub:pub_sub_error()}.
publish_to_glyn_pubsub(Glyn_instances, Aggregate_id, Events) ->
    Message = {process_events, Aggregate_id, Events},
    glyn@pubsub:publish(
        erlang:element(2, erlang:element(4, Glyn_instances)),
        erlang:element(3, Glyn_instances),
        Message
    ).

-file("src/eventsourcing_glyn.gleam", 338).
-spec glyn_commit_events(
    glyn_store(any(), IPO, IPP, IPQ, IPR, IPS),
    IPS,
    eventsourcing:aggregate(IPO, IPP, IPQ, IPR),
    list(IPQ),
    list({binary(), binary()})
) -> {ok, {list(eventsourcing:event_envelop(IPQ)), integer()}} |
    {error, eventsourcing:event_sourcing_error(IPR)}.
glyn_commit_events(Glyn_store, Tx, Aggregate, Events, Metadata) ->
    begin
        Result = (erlang:element(6, erlang:element(5, Glyn_store)))(
            Tx,
            Aggregate,
            Events,
            Metadata
        ),
        case Result of
            {ok, X} ->
                {Events@1, Version} = X,
                begin
                    Result@1 = begin
                        _pipe = publish_to_glyn_pubsub(
                            Glyn_store,
                            erlang:element(2, Aggregate),
                            Events@1
                        ),
                        case _pipe of
                            {ok, X@1} ->
                                {ok, X@1};

                            {error, Error} ->
                                {error,
                                    begin
                                        _@1 = Error,
                                        {event_store_error,
                                            <<"Failed to publish to Glyn PubSub"/utf8>>}
                                    end}
                        end
                    end,
                    case Result@1 of
                        {ok, X@2} ->
                            _@2 = X@2,
                            {ok, {Events@1, Version}};

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

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

-file("src/eventsourcing_glyn.gleam", 388).
-spec proxy_load_events(
    glyn_store(any(), any(), any(), IRD, IRE, any()),
    binary(),
    integer()
) -> {ok, list(eventsourcing:event_envelop(IRD))} |
    {error, eventsourcing:event_sourcing_error(IRE)}.
proxy_load_events(Glyn_store, Aggregate_id, Start_from) ->
    (erlang:element(4, erlang:element(5, Glyn_store)))(
        fun(Underlying_tx) ->
            (erlang:element(7, erlang:element(5, Glyn_store)))(
                erlang:element(10, erlang:element(5, Glyn_store)),
                Underlying_tx,
                Aggregate_id,
                Start_from
            )
        end
    ).

-file("src/eventsourcing_glyn.gleam", 412).
-spec proxy_load_snapshot(
    glyn_store(any(), IRS, any(), any(), IRV, any()),
    binary()
) -> {ok, gleam@option:option(eventsourcing:snapshot(IRS))} |
    {error, eventsourcing:event_sourcing_error(IRV)}.
proxy_load_snapshot(Glyn_store, Aggregate_id) ->
    (erlang:element(5, erlang:element(5, Glyn_store)))(
        fun(Underlying_tx) ->
            (erlang:element(8, erlang:element(5, Glyn_store)))(
                Underlying_tx,
                Aggregate_id
            )
        end
    ).

-file("src/eventsourcing_glyn.gleam", 428).
-spec proxy_save_snapshot(
    glyn_store(any(), ISJ, any(), any(), ISM, any()),
    eventsourcing:snapshot(ISJ)
) -> {ok, nil} | {error, eventsourcing:event_sourcing_error(ISM)}.
proxy_save_snapshot(Glyn_store, Snapshot) ->
    (erlang:element(2, erlang:element(5, Glyn_store)))(
        fun(Underlying_tx) ->
            (erlang:element(9, erlang:element(5, Glyn_store)))(
                Underlying_tx,
                Snapshot
            )
        end
    ).

-file("src/eventsourcing_glyn.gleam", 444).
-spec proxy_execute_transaction(
    glyn_store(any(), any(), any(), any(), ITC, ITD),
    fun((ITD) -> {ok, nil} | {error, eventsourcing:event_sourcing_error(ITC)})
) -> {ok, nil} | {error, eventsourcing:event_sourcing_error(ITC)}.
proxy_execute_transaction(Glyn_store, Transaction_fn) ->
    (erlang:element(2, erlang:element(5, Glyn_store)))(Transaction_fn).

-file("src/eventsourcing_glyn.gleam", 458).
-spec proxy_get_latest_snapshot_transaction(
    glyn_store(any(), ITP, any(), any(), ITS, ITT),
    fun((ITT) -> {ok, gleam@option:option(eventsourcing:snapshot(ITP))} |
        {error, eventsourcing:event_sourcing_error(ITS)})
) -> {ok, gleam@option:option(eventsourcing:snapshot(ITP))} |
    {error, eventsourcing:event_sourcing_error(ITS)}.
proxy_get_latest_snapshot_transaction(Glyn_store, Transaction_fn) ->
    (erlang:element(5, erlang:element(5, Glyn_store)))(Transaction_fn).

-file("src/eventsourcing_glyn.gleam", 472).
-spec proxy_load_aggregate_transaction(
    glyn_store(any(), IUH, IUI, IUJ, IUK, IUL),
    fun((IUL) -> {ok, eventsourcing:aggregate(IUH, IUI, IUJ, IUK)} |
        {error, eventsourcing:event_sourcing_error(IUK)})
) -> {ok, eventsourcing:aggregate(IUH, IUI, IUJ, IUK)} |
    {error, eventsourcing:event_sourcing_error(IUK)}.
proxy_load_aggregate_transaction(Glyn_store, Transaction_fn) ->
    (erlang:element(3, erlang:element(5, Glyn_store)))(Transaction_fn).

-file("src/eventsourcing_glyn.gleam", 486).
-spec proxy_load_events_transaction(
    glyn_store(any(), any(), any(), IVD, IVE, IVF),
    fun((IVF) -> {ok, list(eventsourcing:event_envelop(IVD))} |
        {error, eventsourcing:event_sourcing_error(IVE)})
) -> {ok, list(eventsourcing:event_envelop(IVD))} |
    {error, eventsourcing:event_sourcing_error(IVE)}.
proxy_load_events_transaction(Glyn_store, Transaction_fn) ->
    (erlang:element(4, erlang:element(5, Glyn_store)))(Transaction_fn).

-file("src/eventsourcing_glyn.gleam", 500).
-spec start_glyn_query_actor(
    glyn_instances(IVS),
    gleam@erlang@process:name(eventsourcing:query_message(IVS)),
    fun((binary(), list(eventsourcing:event_envelop(IVS))) -> nil),
    binary()
) -> {ok,
        gleam@otp@actor:started(gleam@erlang@process:subject(eventsourcing:query_message(IVS)))} |
    {error, gleam@otp@actor:start_error()}.
start_glyn_query_actor(Glyn_instances, Query_name, Query_fn, Pubsub_group) ->
    begin
        Result = begin
            _pipe@4 = gleam@otp@actor:new_with_initialiser(
                500,
                fun(Subject) ->
                    Subscription_selector = glyn@pubsub:subscribe(
                        erlang:element(2, Glyn_instances),
                        Pubsub_group
                    ),
                    Selector = begin
                        _pipe = gleam_erlang_ffi:new_selector(),
                        gleam_erlang_ffi:merge_selector(
                            _pipe,
                            Subscription_selector
                        )
                    end,
                    _pipe@1 = gleam@otp@actor:initialised(Subject),
                    _pipe@2 = gleam@otp@actor:selecting(_pipe@1, Selector),
                    _pipe@3 = gleam@otp@actor:returning(_pipe@2, Subject),
                    {ok, _pipe@3}
                end
            ),
            _pipe@5 = gleam@otp@actor:on_message(
                _pipe@4,
                fun(State, Message) -> case Message of
                        {process_events, Aggregate_id, Events} ->
                            gleam_stdlib:println(
                                <<<<<<"📥 Query actor received "/utf8,
                                            (erlang:integer_to_binary(
                                                erlang:length(Events)
                                            ))/binary>>/binary,
                                        " events for aggregate: "/utf8>>/binary,
                                    Aggregate_id/binary>>
                            ),
                            Query_fn(Aggregate_id, Events),
                            gleam@otp@actor:continue(State)
                    end end
            ),
            _pipe@6 = gleam@otp@actor:named(_pipe@5, Query_name),
            gleam@otp@actor:start(_pipe@6)
        end,
        case Result of
            {ok, X} ->
                {ok, X};

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

-file("src/eventsourcing_glyn.gleam", 149).
?DOC(
    " Creates a supervised Glyn-enhanced event store that wraps an existing event store\n"
    " and adds distributed PubSub capabilities for query actors and event broadcasting.\n"
    " \n"
    " This function creates a complete supervision tree that includes:\n"
    " - Query actors that automatically subscribe to Glyn PubSub for event notifications\n"
    " - The enhanced event store that publishes committed events to all subscribers\n"
    " - Integration with Glyn's distributed coordination system\n"
    " - Fault-tolerant supervision of all distributed components\n"
    "\n"
    " ## Example\n"
    " ```gleam\n"
    " import eventsourcing_glyn\n"
    " import eventsourcing/memory_store\n"
    " import gleam/otp/static_supervisor\n"
    " \n"
    " // 1. Set up underlying store\n"
    " let #(underlying_store, memory_spec) = memory_store.supervised(\n"
    "   process.new_name(\"events\"),\n"
    "   process.new_name(\"snapshots\"), \n"
    "   static_supervisor.OneForOne\n"
    " )\n"
    " \n"
    " // 2. Configure distributed capabilities\n"
    " let config = eventsourcing_glyn.GlynConfig(\n"
    "   pubsub_scope: \"bank_events\",\n"
    "   pubsub_group: \"bank_services\"\n"
    " )\n"
    " \n"
    " // 3. Define query actors\n"
    " let queries = [\n"
    "   #(process.new_name(\"balance_query\"), balance_query_fn),\n"
    "   #(process.new_name(\"audit_query\"), audit_query_fn),\n"
    " ]\n"
    " \n"
    " // 4. Create distributed event store\n"
    " let assert Ok(#(glyn_eventstore, glyn_spec)) = eventsourcing_glyn.supervised(\n"
    "   config,\n"
    "   underlying_store,\n"
    "   queries,\n"
    "   my_event_decoder()\n"
    " )\n"
    " \n"
    " // 5. Start supervision tree\n"
    " let assert Ok(_) =\n"
    "   static_supervisor.new(static_supervisor.OneForOne)\n"
    "   |> static_supervisor.add(memory_spec)\n"
    "   |> static_supervisor.add(glyn_spec)\n"
    "   |> static_supervisor.start()\n"
    " ```\n"
    "\n"
    " ## Query Actor Behavior\n"
    " Each query actor receives a `QueryMessage(event)` containing:\n"
    " - `ProcessEvents(aggregate_id, events)`: List of events to process\n"
    " \n"
    " Query functions have the signature:\n"
    " ```gleam\n"
    " fn(aggregate_id: String, events: List(EventEnvelop(event))) -> Nil\n"
    " ```\n"
    "\n"
    " ## Distributed Event Flow\n"
    " 1. Service A commits events via `eventsourcing.execute()`\n"
    " 2. Events are persisted to the underlying store\n"
    " 3. Events are automatically broadcast via Glyn PubSub\n"
    " 4. All query actors across all nodes receive and process the events\n"
    " 5. Query actors update their projections/views accordingly\n"
).
-spec supervised(
    glyn_config(),
    eventsourcing:event_store(IMW, IMX, IMY, IMZ, INA, INB),
    list({gleam@erlang@process:name(eventsourcing:query_message(IMZ)),
        fun((binary(), list(eventsourcing:event_envelop(IMZ))) -> nil)}),
    gleam@dynamic@decode:decoder(IMZ)
) -> {ok,
        {eventsourcing:event_store(glyn_store(IMW, IMX, IMY, IMZ, INA, INB), IMX, IMY, IMZ, INA, INB),
            gleam@otp@supervision:child_specification(gleam@otp@static_supervisor:supervisor())}} |
    {error, binary()}.
supervised(Config, Underlying_store, Queries, Event_decoder) ->
    Message_decoder = begin
        gleam@dynamic@decode:field(
            1,
            {decoder, fun gleam@dynamic@decode:decode_string/1},
            fun(Aggregate_id) ->
                gleam@dynamic@decode:field(
                    2,
                    gleam@dynamic@decode:list(payload_decoder(Event_decoder)),
                    fun(Events) ->
                        gleam@dynamic@decode:success(
                            {process_events, Aggregate_id, Events}
                        )
                    end
                )
            end
        )
    end,
    Glyn_pubsub = glyn@pubsub:new(
        erlang:element(2, Config),
        Message_decoder,
        {process_events, <<"error"/utf8>>, []}
    ),
    Glyn_instances = {glyn_instances, Glyn_pubsub},
    Glyn_store = {glyn_store,
        erlang:element(2, Config),
        erlang:element(3, Config),
        Glyn_instances,
        Underlying_store},
    Query_specs = gleam@list:map(
        Queries,
        fun(Query_def) ->
            {Query_name, Query_fn} = Query_def,
            gleam@otp@supervision:worker(
                fun() ->
                    start_glyn_query_actor(
                        Glyn_instances,
                        Query_name,
                        Query_fn,
                        erlang:element(3, Config)
                    )
                end
            )
        end
    ),
    Eventstore = {event_store,
        fun(F) -> proxy_execute_transaction(Glyn_store, F) end,
        fun(F@1) -> proxy_load_aggregate_transaction(Glyn_store, F@1) end,
        fun(F@2) -> proxy_load_events_transaction(Glyn_store, F@2) end,
        fun(F@3) -> proxy_get_latest_snapshot_transaction(Glyn_store, F@3) end,
        fun(Tx, Aggregate, Events@1, Metadata) ->
            glyn_commit_events(Glyn_store, Tx, Aggregate, Events@1, Metadata)
        end,
        fun(Glyn_store@1, _, Aggregate_id@1, Start_from) ->
            proxy_load_events(Glyn_store@1, Aggregate_id@1, Start_from)
        end,
        fun(_, Aggregate_id@2) ->
            proxy_load_snapshot(Glyn_store, Aggregate_id@2)
        end,
        fun(_, Snapshot) -> proxy_save_snapshot(Glyn_store, Snapshot) end,
        Glyn_store},
    Supervisor_spec = begin
        _pipe = gleam@otp@static_supervisor:new(one_for_one),
        _pipe@1 = gleam@list:fold(
            Query_specs,
            _pipe,
            fun(Sup, Spec) -> gleam@otp@static_supervisor:add(Sup, Spec) end
        ),
        gleam@otp@static_supervisor:supervised(_pipe@1)
    end,
    {ok, {Eventstore, Supervisor_spec}}.

-file("src/eventsourcing_glyn.gleam", 557).
?DOC(
    " Get the list of subscribers to this Glyn event store's PubSub scope.\n"
    "\n"
    " This function returns information about all query actors currently subscribed\n"
    " to receive events from this distributed event store across all connected nodes.\n"
    "\n"
    " ## Example\n"
    " ```gleam\n"
    " let subscriber_info = eventsourcing_glyn.subscribers(glyn_eventstore)\n"
    " // Use for monitoring or debugging distributed event flow\n"
    " ```\n"
    "\n"
    " ## Use Cases\n"
    " - **Health Monitoring**: Check if expected query actors are connected\n"
    " - **Debugging**: Verify event distribution is working correctly  \n"
    " - **Load Balancing**: Understand the distribution of query actors\n"
    " - **Troubleshooting**: Identify disconnected or failed nodes\n"
).
-spec subscribers(glyn_store(any(), any(), any(), any(), any(), any())) -> list(gleam@erlang@process:pid_()).
subscribers(Glyn_store) ->
    glyn@pubsub:subscribers(
        erlang:element(2, erlang:element(4, Glyn_store)),
        erlang:element(2, Glyn_store)
    ).

-file("src/eventsourcing_glyn.gleam", 596).
?DOC(
    " Manually publish an event message to distributed query subscribers.\n"
    "\n"
    " This function allows you to send custom messages to query actors without\n"
    " going through the normal event store commit process. This is useful for:\n"
    " - Broadcasting system events or notifications\n"
    " - Triggering query actor maintenance or refresh operations\n"
    " - Testing distributed event handling\n"
    "\n"
    " ## Example\n"
    " ```gleam\n"
    " import eventsourcing_glyn\n"
    " import eventsourcing.{ProcessEvents}\n"
    " \n"
    " // Send a custom event to all query actors\n"
    " let message = ProcessEvents(\"system\", [maintenance_event])\n"
    " case eventsourcing_glyn.publish(glyn_store, \"bank_services\", message) {\n"
    "   Ok(count) -> io.println(\"Notified \" <> int.to_string(count) <> \" subscribers\")\n"
    "   Error(error) -> io.println(\"Failed to publish: \" <> string.inspect(error))\n"
    " }\n"
    " ```\n"
    "\n"
    " ## Important Notes\n"
    " - **Normal Usage**: Most events should go through `eventsourcing.execute()` for proper persistence\n"
    " - **Manual Publishing**: Only use this for system events or testing scenarios\n"
    " - **Group Matching**: The group parameter should typically match your `GlynConfig.pubsub_group`\n"
    " - **Return Value**: The count indicates how many query actors received the message\n"
).
-spec publish(
    glyn_store(any(), any(), any(), IWQ, IWR, any()),
    binary(),
    eventsourcing:query_message(IWQ)
) -> {ok, integer()} | {error, eventsourcing:event_sourcing_error(IWR)}.
publish(Glyn_store, Group, Message) ->
    _pipe = glyn@pubsub:publish(
        erlang:element(2, erlang:element(4, Glyn_store)),
        Group,
        Message
    ),
    case _pipe of
        {ok, X} ->
            {ok, X};

        {error, Error} ->
            {error, case Error of
                    {publish_failed, Error@1} ->
                        {event_store_error,
                            <<"Failed to publish to Glyn PubSub: "/utf8,
                                Error@1/binary>>}
                end}
    end.