glubsub

-module(glubsub).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).
-define(FILEPATH, "src/glubsub.gleam").
-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]).

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

?MODULEDOC(
    " This module implements a simple pubsub system using gleam actors.\n"
    "\n"
).

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

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

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

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

-type subscriber(FFO) :: {subscriber,
        gleam@erlang@process:subject(FFO),
        gleam@erlang@process:monitor()}.

-file("src/glubsub.gleam", 179).
-spec topic_to_subject(topic(FGU)) -> gleam@erlang@process:subject(message(FGU)).
topic_to_subject(Topic) ->
    {topic, Subject} = Topic,
    Subject.

-file("src/glubsub.gleam", 63).
?DOC(" Destroys the given topic, unsubscribing all clients.\n").
-spec destroy_topic(topic(any())) -> nil.
destroy_topic(Topic) ->
    gleam@otp@actor:send(topic_to_subject(Topic), shutdown).

-file("src/glubsub.gleam", 88).
?DOC(" Broadcasts a message to all subscribers of the given topic.\n").
-spec broadcast(topic(FGF), FGF) -> {ok, nil} | {error, nil}.
broadcast(Topic, Message) ->
    _pipe = gleam@otp@actor:send(topic_to_subject(Topic), {broadcast, Message}),
    {ok, _pipe}.

-file("src/glubsub.gleam", 184).
-spec remove_subscriber(list(subscriber(FGY)), subscriber(FGY)) -> list(subscriber(FGY)).
remove_subscriber(Subscribers, Unsubscriber) ->
    gleam@list:filter(Subscribers, fun(Sub) -> Sub /= Unsubscriber end).

-file("src/glubsub.gleam", 98).
-spec handle_message(state(FGN), message(FGN)) -> gleam@otp@actor:next(state(FGN), message(FGN)).
handle_message(State, Message) ->
    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} ->
                    case gleam@erlang@process:subject_owner(Client) of
                        {ok, Pid} ->
                            Monitor = gleam@erlang@process:monitor(Pid),
                            New_subs = [{subscriber, Client, Monitor} |
                                erlang:element(2, State)],
                            gleam@otp@actor:send(Reply, {ok, nil}),
                            gleam@otp@actor:continue(
                                begin
                                    _record = State,
                                    {state,
                                        New_subs,
                                        erlang:element(3, _record)}
                                end
                            );

                        {error, _} ->
                            gleam@otp@actor:continue(State)
                    end
            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
                    ),
                    gleam@otp@actor:send(Reply@1, {ok, nil}),
                    gleam@otp@actor:continue(
                        begin
                            _record@1 = State,
                            {state, New_subs@1, erlang:element(3, _record@1)}
                        end
                    )
            end;

        {broadcast, Message@1} ->
            _pipe = erlang:element(2, State),
            gleam@list:each(
                _pipe,
                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, {process_down, _, Pid@1, _}} ->
            Ok_pid = {ok, Pid@1},
            case begin
                _pipe@1 = erlang:element(2, State),
                gleam@list:find(
                    _pipe@1,
                    fun(Sub@3) ->
                        gleam@erlang@process:subject_owner(
                            erlang:element(2, Sub@3)
                        )
                        =:= Ok_pid
                    end
                )
            end of
                {error, nil} ->
                    gleam@otp@actor:continue(State);

                {ok, Unsub@1} ->
                    gleam@erlang@process:demonitor_process(
                        erlang:element(3, Unsub@1)
                    ),
                    New_subs@2 = remove_subscriber(
                        erlang:element(2, State),
                        Unsub@1
                    ),
                    gleam@otp@actor:continue(
                        begin
                            _record@2 = State,
                            {state, New_subs@2, erlang:element(3, _record@2)}
                        end
                    )
            end;

        {subscriber_down, _} ->
            gleam@otp@actor:continue(State);

        shutdown ->
            gleam@otp@actor:stop()
    end.

-file("src/glubsub.gleam", 42).
?DOC(" Creates a new topic. Which is a pubsub channel that clients can subscribe to.\n").
-spec new_topic() -> {ok, topic(any())} | {error, glubsub_error()}.
new_topic() ->
    _pipe@5 = gleam@otp@actor:new_with_initialiser(
        1000,
        fun(Subject) ->
            Selector = begin
                _pipe = gleam_erlang_ffi:new_selector(),
                _pipe@1 = gleam@erlang@process:select(_pipe, Subject),
                gleam@erlang@process:select_monitors(
                    _pipe@1,
                    fun(Field@0) -> {subscriber_down, Field@0} end
                )
            end,
            _pipe@2 = gleam@otp@actor:initialised({state, [], Selector}),
            _pipe@3 = gleam@otp@actor:selecting(_pipe@2, Selector),
            _pipe@4 = gleam@otp@actor:returning(_pipe@3, Subject),
            {ok, _pipe@4}
        end
    ),
    _pipe@6 = gleam@otp@actor:on_message(_pipe@5, fun handle_message/2),
    _pipe@7 = gleam@otp@actor:start(_pipe@6),
    _pipe@8 = gleam@result:map(
        _pipe@7,
        fun(Started) ->
            {started, _, Subject@1} = Started,
            {topic, Subject@1}
        end
    ),
    gleam@result:map_error(_pipe@8, fun(Err) -> {start_error, Err} end).

-file("src/glubsub.gleam", 68).
?DOC(" Subscribes the given client to the given topic.\n").
-spec subscribe(topic(FFV), gleam@erlang@process:subject(FFV)) -> {ok, nil} |
    {error, glubsub_error()}.
subscribe(Topic, Client) ->
    gleam@otp@actor:call(
        topic_to_subject(Topic),
        1000,
        fun(Self) -> {subscribe, Self, Client} end
    ).

-file("src/glubsub.gleam", 78).
?DOC(" Unsubscribes the given client from the given topic.\n").
-spec unsubscribe(topic(FGA), gleam@erlang@process:subject(FGA)) -> {ok, nil} |
    {error, glubsub_error()}.
unsubscribe(Topic, Client) ->
    gleam@otp@actor:call(
        topic_to_subject(Topic),
        1000,
        fun(Self) -> {unsubscribe, Self, Client} end
    ).

-file("src/glubsub.gleam", 94).
?DOC(" Returns a set of all subscribers to the given topic.\n").
-spec get_subscribers(topic(FGJ)) -> list(subscriber(FGJ)).
get_subscribers(Topic) ->
    gleam@otp@actor:call(
        topic_to_subject(Topic),
        1000,
        fun(Field@0) -> {get_subscribers, Field@0} end
    ).