-module(lightspeed@channel).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/lightspeed/channel.gleam").
-export([new_single_node/0, new_cluster_ready/2, event_label/1, handle/2, flush_outbox/1, crashed/1, adapter_label/1, subscribers/2, presences/2, membership_count/1, outbound_label/1]).
-export_type([event/0, outbound/0, membership/0, channel/0]).

-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(" Channel runtime model with topic routing, pubsub, and presence.\n").

-type event() :: {join, binary(), binary(), binary(), integer()} |
    {rejoin, binary(), binary(), binary(), integer()} |
    {leave, binary(), binary()} |
    {broadcast, binary(), binary(), binary()} |
    {push, binary(), binary(), binary(), binary()} |
    {crash, binary()} |
    restart.

-type outbound() :: {reply, binary(), binary(), binary(), binary()} |
    {push_message, binary(), binary(), binary(), binary()} |
    {broadcast_message, binary(), binary(), list(binary())} |
    {presence_diff, binary(), lightspeed@presence:diff()} |
    {system, binary()}.

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

-opaque channel() :: {channel,
        lightspeed@pubsub:broker(binary()),
        lightspeed@presence:tracker(),
        list(membership()),
        list(outbound()),
        boolean()}.

-file("src/lightspeed/channel.gleam", 50).
?DOC(" Start a single-node channel.\n").
-spec new_single_node() -> channel().
new_single_node() ->
    {channel,
        lightspeed@pubsub:new_single_node(),
        lightspeed@presence:new(),
        [],
        [],
        false}.

-file("src/lightspeed/channel.gleam", 61).
?DOC(" Start a cluster-ready channel.\n").
-spec new_cluster_ready(binary(), binary()) -> channel().
new_cluster_ready(Node_id, Cluster) ->
    {channel,
        lightspeed@pubsub:new_cluster_ready(Node_id, Cluster),
        lightspeed@presence:new(),
        [],
        [],
        false}.

-file("src/lightspeed/channel.gleam", 359).
-spec emit(channel(), outbound()) -> channel().
emit(Channel, Outbound) ->
    {channel,
        erlang:element(2, Channel),
        erlang:element(3, Channel),
        erlang:element(4, Channel),
        [Outbound | erlang:element(5, Channel)],
        erlang:element(6, Channel)}.

-file("src/lightspeed/channel.gleam", 370).
-spec find_membership(list(membership()), binary(), binary()) -> gleam@option:option(membership()).
find_membership(Memberships_rev, Client_id, Topic) ->
    case Memberships_rev of
        [] ->
            none;

        [Membership | Rest] ->
            case (erlang:element(2, Membership) =:= Client_id) andalso (erlang:element(
                3,
                Membership
            )
            =:= Topic) of
                true ->
                    {some, Membership};

                false ->
                    find_membership(Rest, Client_id, Topic)
            end
    end.

-file("src/lightspeed/channel.gleam", 363).
-spec has_membership(channel(), binary(), binary()) -> boolean().
has_membership(Channel, Client_id, Topic) ->
    case find_membership(erlang:element(4, Channel), Client_id, Topic) of
        {some, _} ->
            true;

        none ->
            false
    end.

-file("src/lightspeed/channel.gleam", 337).
-spec push_deliveries(channel(), list(lightspeed@pubsub:delivery(binary()))) -> channel().
push_deliveries(Channel, Deliveries) ->
    case Deliveries of
        [] ->
            Channel;

        [Delivery | Rest] ->
            push_deliveries(
                emit(
                    Channel,
                    {push_message,
                        lightspeed@pubsub:delivery_subscriber(Delivery),
                        lightspeed@pubsub:delivery_topic(Delivery),
                        <<"broadcast"/utf8>>,
                        lightspeed@pubsub:delivery_message(Delivery)}
                ),
                Rest
            )
    end.

-file("src/lightspeed/channel.gleam", 412).
-spec remove_membership_loop(list(membership()), binary(), binary()) -> {list(membership()),
    gleam@option:option(membership())}.
remove_membership_loop(Memberships_rev, Client_id, Topic) ->
    case Memberships_rev of
        [] ->
            {[], none};

        [Membership | Rest] ->
            case (erlang:element(2, Membership) =:= Client_id) andalso (erlang:element(
                3,
                Membership
            )
            =:= Topic) of
                true ->
                    {Rest, {some, Membership}};

                false ->
                    {Updated_rest, Removed} = remove_membership_loop(
                        Rest,
                        Client_id,
                        Topic
                    ),
                    {[Membership | Updated_rest], Removed}
            end
    end.

-file("src/lightspeed/channel.gleam", 402).
-spec remove_membership(channel(), binary(), binary()) -> {channel(),
    gleam@option:option(membership())}.
remove_membership(Channel, Client_id, Topic) ->
    {Memberships_rev, Removed} = remove_membership_loop(
        erlang:element(4, Channel),
        Client_id,
        Topic
    ),
    {{channel,
            erlang:element(2, Channel),
            erlang:element(3, Channel),
            Memberships_rev,
            erlang:element(5, Channel),
            erlang:element(6, Channel)},
        Removed}.

-file("src/lightspeed/channel.gleam", 300).
-spec leave(channel(), binary(), binary()) -> channel().
leave(Channel, Client_id, Topic) ->
    {Updated, Membership} = remove_membership(Channel, Client_id, Topic),
    case Membership of
        none ->
            emit(
                Updated,
                {reply,
                    Client_id,
                    Topic,
                    <<"error"/utf8>>,
                    <<"not_joined"/utf8>>}
            );

        {some, Membership@1} ->
            Broker = lightspeed@pubsub:unsubscribe(
                erlang:element(2, Updated),
                Topic,
                Client_id
            ),
            {Tracker, Diff} = lightspeed@presence:leave(
                erlang:element(3, Updated),
                erlang:element(3, Membership@1),
                erlang:element(4, Membership@1),
                erlang:element(5, Membership@1)
            ),
            _pipe = {channel,
                Broker,
                Tracker,
                erlang:element(4, Updated),
                erlang:element(5, Updated),
                erlang:element(6, Updated)},
            _pipe@1 = emit(
                _pipe,
                {reply, Client_id, Topic, <<"ok"/utf8>>, <<"left"/utf8>>}
            ),
            emit(_pipe@1, {presence_diff, Topic, Diff})
    end.

-file("src/lightspeed/channel.gleam", 385).
-spec upsert_membership(list(membership()), membership()) -> list(membership()).
upsert_membership(Memberships_rev, Target) ->
    case Memberships_rev of
        [] ->
            [Target];

        [Membership | Rest] ->
            case (erlang:element(2, Membership) =:= erlang:element(2, Target))
            andalso (erlang:element(3, Membership) =:= erlang:element(3, Target)) of
                true ->
                    [Target | Rest];

                false ->
                    [Membership | upsert_membership(Rest, Target)]
            end
    end.

-file("src/lightspeed/channel.gleam", 431).
-spec membership_ref(binary(), binary()) -> binary().
membership_ref(Client_id, Topic) ->
    <<<<Client_id/binary, "@"/utf8>>/binary, Topic/binary>>.

-file("src/lightspeed/channel.gleam", 257).
-spec join(channel(), binary(), binary(), binary(), integer(), binary()) -> channel().
join(Channel, Client_id, Topic, User_id, Now_ms, Reply_payload) ->
    Presence_ref = membership_ref(Client_id, Topic),
    Broker = lightspeed@pubsub:subscribe(
        erlang:element(2, Channel),
        Topic,
        Client_id
    ),
    {Tracker, Diff} = lightspeed@presence:join(
        erlang:element(3, Channel),
        Topic,
        User_id,
        {meta, Presence_ref, Now_ms}
    ),
    Memberships_rev = upsert_membership(
        erlang:element(4, Channel),
        {membership, Client_id, Topic, User_id, Presence_ref}
    ),
    _pipe = {channel,
        Broker,
        Tracker,
        Memberships_rev,
        erlang:element(5, Channel),
        erlang:element(6, Channel)},
    _pipe@1 = emit(
        _pipe,
        {reply, Client_id, Topic, <<"ok"/utf8>>, Reply_payload}
    ),
    emit(_pipe@1, {presence_diff, Topic, Diff}).

-file("src/lightspeed/channel.gleam", 150).
-spec apply_event(channel(), event()) -> channel().
apply_event(Channel, Event) ->
    case Event of
        {join, Client_id, Topic, User_id, Now_ms} ->
            join(Channel, Client_id, Topic, User_id, Now_ms, <<"joined"/utf8>>);

        {rejoin, Client_id@1, Topic@1, User_id@1, Now_ms@1} ->
            {Membership_state, Previous} = remove_membership(
                Channel,
                Client_id@1,
                Topic@1
            ),
            Membership_state@1 = case Previous of
                none ->
                    Membership_state;

                {some, Membership} ->
                    Broker = lightspeed@pubsub:unsubscribe(
                        erlang:element(2, Membership_state),
                        erlang:element(3, Membership),
                        erlang:element(2, Membership)
                    ),
                    {Tracker, Diff} = lightspeed@presence:leave(
                        erlang:element(3, Membership_state),
                        erlang:element(3, Membership),
                        erlang:element(4, Membership),
                        erlang:element(5, Membership)
                    ),
                    _pipe = {channel,
                        Broker,
                        Tracker,
                        erlang:element(4, Membership_state),
                        erlang:element(5, Membership_state),
                        erlang:element(6, Membership_state)},
                    emit(
                        _pipe,
                        {presence_diff, erlang:element(3, Membership), Diff}
                    )
            end,
            join(
                Membership_state@1,
                Client_id@1,
                Topic@1,
                User_id@1,
                Now_ms@1,
                <<"rejoined"/utf8>>
            );

        {leave, Client_id@2, Topic@2} ->
            leave(Channel, Client_id@2, Topic@2);

        {broadcast, Client_id@3, Topic@3, Payload} ->
            case has_membership(Channel, Client_id@3, Topic@3) of
                false ->
                    emit(
                        Channel,
                        {reply,
                            Client_id@3,
                            Topic@3,
                            <<"error"/utf8>>,
                            <<"not_joined"/utf8>>}
                    );

                true ->
                    {_, Deliveries} = lightspeed@pubsub:publish(
                        erlang:element(2, Channel),
                        Topic@3,
                        Payload
                    ),
                    Recipients = gleam@list:map(
                        Deliveries,
                        fun lightspeed@pubsub:delivery_subscriber/1
                    ),
                    Pushed = push_deliveries(Channel, Deliveries),
                    _pipe@1 = Pushed,
                    _pipe@2 = emit(
                        _pipe@1,
                        {broadcast_message, Topic@3, Payload, Recipients}
                    ),
                    emit(
                        _pipe@2,
                        {reply,
                            Client_id@3,
                            Topic@3,
                            <<"ok"/utf8>>,
                            <<"broadcast:"/utf8,
                                (erlang:integer_to_binary(
                                    erlang:length(Recipients)
                                ))/binary>>}
                    )
            end;

        {push, Client_id@4, Topic@4, Event@1, Payload@1} ->
            case has_membership(Channel, Client_id@4, Topic@4) of
                false ->
                    emit(
                        Channel,
                        {reply,
                            Client_id@4,
                            Topic@4,
                            <<"error"/utf8>>,
                            <<"not_joined"/utf8>>}
                    );

                true ->
                    _pipe@3 = Channel,
                    _pipe@4 = emit(
                        _pipe@3,
                        {push_message, Client_id@4, Topic@4, Event@1, Payload@1}
                    ),
                    emit(
                        _pipe@4,
                        {reply,
                            Client_id@4,
                            Topic@4,
                            <<"ok"/utf8>>,
                            <<"pushed"/utf8>>}
                    )
            end;

        {crash, Reason} ->
            _pipe@5 = {channel,
                erlang:element(2, Channel),
                erlang:element(3, Channel),
                erlang:element(4, Channel),
                erlang:element(5, Channel),
                true},
            emit(_pipe@5, {system, <<"crashed:"/utf8, Reason/binary>>});

        restart ->
            Channel
    end.

-file("src/lightspeed/channel.gleam", 138).
?DOC(" Stable event label.\n").
-spec event_label(event()) -> binary().
event_label(Event) ->
    case Event of
        {join, _, Topic, _, _} ->
            <<"join:"/utf8, Topic/binary>>;

        {rejoin, _, Topic@1, _, _} ->
            <<"rejoin:"/utf8, Topic@1/binary>>;

        {leave, _, Topic@2} ->
            <<"leave:"/utf8, Topic@2/binary>>;

        {broadcast, _, Topic@3, _} ->
            <<"broadcast:"/utf8, Topic@3/binary>>;

        {push, _, Topic@4, Event@1, _} ->
            <<<<<<"push:"/utf8, Topic@4/binary>>/binary, ":"/utf8>>/binary,
                Event@1/binary>>;

        {crash, Reason} ->
            <<"crash:"/utf8, Reason/binary>>;

        restart ->
            <<"restart"/utf8>>
    end.

-file("src/lightspeed/channel.gleam", 72).
?DOC(" Handle one channel event.\n").
-spec handle(channel(), event()) -> channel().
handle(Channel, Event) ->
    case erlang:element(6, Channel) of
        true ->
            case Event of
                restart ->
                    _pipe = {channel,
                        erlang:element(2, Channel),
                        erlang:element(3, Channel),
                        erlang:element(4, Channel),
                        erlang:element(5, Channel),
                        false},
                    emit(_pipe, {system, <<"restarted"/utf8>>});

                _ ->
                    emit(
                        Channel,
                        {system,
                            <<"ignored_while_crashed:"/utf8,
                                (event_label(Event))/binary>>}
                    )
            end;

        false ->
            apply_event(Channel, Event)
    end.

-file("src/lightspeed/channel.gleam", 92).
?DOC(" Flush outbox in emit order.\n").
-spec flush_outbox(channel()) -> {channel(), list(outbound())}.
flush_outbox(Channel) ->
    Outbox = lists:reverse(erlang:element(5, Channel)),
    {{channel,
            erlang:element(2, Channel),
            erlang:element(3, Channel),
            erlang:element(4, Channel),
            [],
            erlang:element(6, Channel)},
        Outbox}.

-file("src/lightspeed/channel.gleam", 98).
?DOC(" True when channel is crashed.\n").
-spec crashed(channel()) -> boolean().
crashed(Channel) ->
    erlang:element(6, Channel).

-file("src/lightspeed/channel.gleam", 103).
?DOC(" PubSub adapter label.\n").
-spec adapter_label(channel()) -> binary().
adapter_label(Channel) ->
    _pipe = erlang:element(2, Channel),
    _pipe@1 = lightspeed@pubsub:adapter(_pipe),
    lightspeed@pubsub:adapter_label(_pipe@1).

-file("src/lightspeed/channel.gleam", 110).
?DOC(" Subscribers for one topic.\n").
-spec subscribers(channel(), binary()) -> list(binary()).
subscribers(Channel, Topic) ->
    lightspeed@pubsub:subscribers(erlang:element(2, Channel), Topic).

-file("src/lightspeed/channel.gleam", 115).
?DOC(" Presence entries for one topic.\n").
-spec presences(channel(), binary()) -> list(lightspeed@presence:entry()).
presences(Channel, Topic) ->
    lightspeed@presence:list_topic(erlang:element(3, Channel), Topic).

-file("src/lightspeed/channel.gleam", 120).
?DOC(" Membership count.\n").
-spec membership_count(channel()) -> integer().
membership_count(Channel) ->
    erlang:length(erlang:element(4, Channel)).

-file("src/lightspeed/channel.gleam", 125).
?DOC(" Stable outbound label.\n").
-spec outbound_label(outbound()) -> binary().
outbound_label(Outbound) ->
    case Outbound of
        {reply, _, Topic, Status, _} ->
            <<<<<<"reply:"/utf8, Topic/binary>>/binary, ":"/utf8>>/binary,
                Status/binary>>;

        {push_message, _, Topic@1, Event, _} ->
            <<<<<<"push:"/utf8, Topic@1/binary>>/binary, ":"/utf8>>/binary,
                Event/binary>>;

        {broadcast_message, Topic@2, _, Recipients} ->
            <<<<<<"broadcast:"/utf8, Topic@2/binary>>/binary, ":"/utf8>>/binary,
                (erlang:integer_to_binary(erlang:length(Recipients)))/binary>>;

        {presence_diff, Topic@3, Diff} ->
            <<<<<<"presence:"/utf8, Topic@3/binary>>/binary, ":"/utf8>>/binary,
                (lightspeed@presence:diff_label(Diff))/binary>>;

        {system, Label} ->
            <<"system:"/utf8, Label/binary>>
    end.