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

-export([destroy_topic/1, broadcast/2, new_topic/0, subscribe/2, unsubscribe/2, get_subscribers/1]).
-export_type([topic/1, message/1, glubsub_error/0, state/1, subscriber/1]).

-type topic(GYN) :: {topic, gleam@erlang@process:subject(message(GYN))}.

-opaque message(GYO) :: {subscribe,
        gleam@erlang@process:subject({ok, nil} | {error, glubsub_error()}),
        gleam@erlang@process:subject(GYO)} |
    {unsubscribe,
        gleam@erlang@process:subject({ok, nil} | {error, glubsub_error()}),
        gleam@erlang@process:subject(GYO)} |
    {broadcast, GYO} |
    {get_subscribers, gleam@erlang@process:subject(list(subscriber(GYO)))} |
    {subscriber_down, gleam@erlang@process:process_down()} |
    shutdown.

-opaque glubsub_error() :: already_subscribed |
    not_subscribed |
    {start_error, gleam@otp@actor:start_error()}.

-type state(GYP) :: {state,
        list(subscriber(GYP)),
        gleam@erlang@process:selector(message(GYP))}.

-type subscriber(GYQ) :: {subscriber,
        gleam@erlang@process:subject(GYQ),
        gleam@erlang@process:process_monitor()}.

-spec topic_to_subject(topic(GZW)) -> gleam@erlang@process:subject(message(GZW)).
topic_to_subject(Topic) ->
    {topic, Subject} = Topic,
    Subject.

-spec destroy_topic(topic(any())) -> nil.
destroy_topic(Topic) ->
    gleam@otp@actor:send(topic_to_subject(Topic), shutdown).

-spec broadcast(topic(GZH), GZH) -> {ok, nil} | {error, nil}.
broadcast(Topic, Message) ->
    _pipe = gleam@otp@actor:send(topic_to_subject(Topic), {broadcast, Message}),
    {ok, _pipe}.

-spec remove_subscriber(list(subscriber(HAA)), subscriber(HAA)) -> list(subscriber(HAA)).
remove_subscriber(Subscribers, Unsubscriber) ->
    _pipe = Subscribers,
    _pipe@1 = gleam@list:pop(_pipe, fun(Sub) -> Sub =:= Unsubscriber end),
    _pipe@2 = gleam@result:map(_pipe@1, fun(Res) -> erlang:element(2, Res) end),
    gleam@result:unwrap(_pipe@2, Subscribers).

-spec selector_down_from_subscribers(
    gleam@erlang@process:selector(message(HAG)),
    list(subscriber(HAG))
) -> gleam@erlang@process:selector(message(HAG)).
selector_down_from_subscribers(Selector, Subscribers) ->
    gleam@list:fold(
        Subscribers,
        Selector,
        fun(Selector@1, Sub) ->
            gleam@erlang@process:selecting_process_down(
                Selector@1,
                erlang:element(3, Sub),
                fun(Field@0) -> {subscriber_down, Field@0} end
            )
        end
    ).

-spec handle_message(message(GZP), state(GZP)) -> gleam@otp@actor:next(message(GZP), state(GZP)).
handle_message(Message, State) ->
    case Message of
        {subscribe, Reply, Client} ->
            case gleam@list:find(
                erlang:element(2, State),
                fun(Sub) -> erlang:element(2, Sub) =:= Client end
            ) of
                {ok, _} ->
                    gleam@otp@actor:send(Reply, {error, already_subscribed}),
                    gleam@otp@actor:continue(State);

                {error, nil} ->
                    Monitor = gleam@erlang@process:monitor_process(
                        gleam@erlang@process:subject_owner(Client)
                    ),
                    New_selector = begin
                        _pipe = erlang:element(3, State),
                        gleam@erlang@process:selecting_process_down(
                            _pipe,
                            Monitor,
                            fun(Field@0) -> {subscriber_down, Field@0} end
                        )
                    end,
                    New_subs = [{subscriber, Client, Monitor} |
                        erlang:element(2, State)],
                    gleam@otp@actor:send(Reply, {ok, nil}),
                    _pipe@1 = gleam@otp@actor:continue(
                        {state, New_subs, New_selector}
                    ),
                    gleam@otp@actor:with_selector(_pipe@1, New_selector)
            end;

        {unsubscribe, Reply@1, Client@1} ->
            case gleam@list:find(
                erlang:element(2, State),
                fun(Sub@1) -> erlang:element(2, Sub@1) =:= Client@1 end
            ) of
                {error, nil} ->
                    gleam@otp@actor:send(Reply@1, {error, not_subscribed}),
                    gleam@otp@actor:continue(State);

                {ok, Unsub} ->
                    New_subs@1 = remove_subscriber(
                        erlang:element(2, State),
                        Unsub
                    ),
                    New_selector@1 = begin
                        _pipe@2 = gleam_erlang_ffi:new_selector(),
                        selector_down_from_subscribers(_pipe@2, New_subs@1)
                    end,
                    gleam@otp@actor:send(Reply@1, {ok, nil}),
                    gleam@otp@actor:continue(
                        {state, New_subs@1, New_selector@1}
                    )
            end;

        {broadcast, Message@1} ->
            _pipe@3 = erlang:element(2, State),
            gleam@list:each(
                _pipe@3,
                fun(Sub@2) ->
                    gleam@otp@actor:send(erlang:element(2, Sub@2), Message@1)
                end
            ),
            gleam@otp@actor:continue(State);

        {get_subscribers, Reply@2} ->
            gleam@otp@actor:send(Reply@2, erlang:element(2, State)),
            gleam@otp@actor:continue(State);

        {subscriber_down, Client@2} ->
            case begin
                _pipe@4 = erlang:element(2, State),
                gleam@list:find(
                    _pipe@4,
                    fun(Sub@3) ->
                        gleam@erlang@process:subject_owner(
                            erlang:element(2, Sub@3)
                        )
                        =:= erlang:element(2, Client@2)
                    end
                )
            end of
                {error, nil} ->
                    gleam@otp@actor:continue(State);

                {ok, Unsub@1} ->
                    gleam_erlang_ffi:demonitor(erlang:element(3, Unsub@1)),
                    New_subs@2 = remove_subscriber(
                        erlang:element(2, State),
                        Unsub@1
                    ),
                    New_selector@2 = begin
                        _pipe@5 = gleam_erlang_ffi:new_selector(),
                        selector_down_from_subscribers(_pipe@5, New_subs@2)
                    end,
                    gleam@otp@actor:continue(
                        {state, New_subs@2, New_selector@2}
                    )
            end;

        shutdown ->
            {stop, normal}
    end.

-spec new_topic() -> {ok, topic(any())} | {error, glubsub_error()}.
new_topic() ->
    _pipe = gleam@otp@actor:start_spec(
        {spec,
            fun() ->
                Selector = gleam_erlang_ffi:new_selector(),
                {ready, {state, [], Selector}, Selector}
            end,
            1000,
            fun handle_message/2}
    ),
    _pipe@1 = gleam@result:map(_pipe, fun(Subject) -> {topic, Subject} end),
    gleam@result:map_error(_pipe@1, fun(Err) -> {start_error, Err} end).

-spec subscribe(topic(GYX), gleam@erlang@process:subject(GYX)) -> {ok, nil} |
    {error, glubsub_error()}.
subscribe(Topic, Client) ->
    gleam@otp@actor:call(
        topic_to_subject(Topic),
        fun(Self) -> {subscribe, Self, Client} end,
        1000
    ).

-spec unsubscribe(topic(GZC), gleam@erlang@process:subject(GZC)) -> {ok, nil} |
    {error, glubsub_error()}.
unsubscribe(Topic, Client) ->
    gleam@otp@actor:call(
        topic_to_subject(Topic),
        fun(Self) -> {unsubscribe, Self, Client} end,
        1000
    ).

-spec get_subscribers(topic(GZL)) -> list(subscriber(GZL)).
get_subscribers(Topic) ->
    gleam@otp@actor:call(
        topic_to_subject(Topic),
        fun(Field@0) -> {get_subscribers, Field@0} end,
        1000
    ).