glupbit

-module(glupbit@websocket@connection).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/glupbit/websocket/connection.gleam").
-export([subscribe/3, list_subscriptions/1, decode_ws_message/1, connect_public/2, connect_private/3, connect_public_with_reconnect/3]).
-export_type([ws_message/0, ws_command/0, ws_state/1, reconnect_config/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(
    " WebSocket connection management via stratus actors.\n"
    "\n"
    " Handles public and private (authenticated) connections with automatic\n"
    " message routing based on `type` / `method` fields in incoming JSON.\n"
    " Supports automatic reconnection with exponential backoff.\n"
).

-type ws_message() :: {ticker_msg, glupbit@websocket@subscription:ticker_data()} |
    {trade_msg, glupbit@websocket@subscription:trade_data()} |
    {orderbook_msg, glupbit@websocket@subscription:orderbook_data()} |
    {subscription_list_msg,
        glupbit@websocket@subscription:list_subscriptions_response()} |
    {status_msg, binary()} |
    {error_msg, binary(), binary()} |
    {raw_msg, binary()}.

-type ws_command() :: {subscribe,
        list(glupbit@websocket@subscription:subscription()),
        glupbit@websocket@subscription:ws_format()} |
    list_subscriptions.

-type ws_state(ACLY) :: {ws_state, fun((ACLY, ws_message()) -> ACLY), ACLY}.

-type reconnect_config() :: {reconnect_config,
        list(glupbit@websocket@subscription:subscription()),
        glupbit@websocket@subscription:ws_format(),
        integer(),
        integer(),
        fun(() -> nil)}.

-file("src/glupbit/websocket/connection.gleam", 105).
?DOC(" Send a subscription request over the WebSocket.\n").
-spec subscribe(
    gleam@erlang@process:subject(stratus:internal_message(ws_command())),
    list(glupbit@websocket@subscription:subscription()),
    glupbit@websocket@subscription:ws_format()
) -> nil.
subscribe(Conn, Subscriptions, Format) ->
    gleam@erlang@process:send(
        Conn,
        stratus:to_user_message({subscribe, Subscriptions, Format})
    ).

-file("src/glupbit/websocket/connection.gleam", 114).
?DOC(" Query the list of active subscriptions on the WebSocket.\n").
-spec list_subscriptions(
    gleam@erlang@process:subject(stratus:internal_message(ws_command()))
) -> nil.
list_subscriptions(Conn) ->
    gleam@erlang@process:send(Conn, stratus:to_user_message(list_subscriptions)).

-file("src/glupbit/websocket/connection.gleam", 235).
-spec build_command_message(ws_command()) -> binary().
build_command_message(Cmd) ->
    Ticket = youid@uuid:v4_string(),
    case Cmd of
        {subscribe, Subs, Format} ->
            glupbit@websocket@subscription:build_subscription_message(
                Ticket,
                Subs,
                Format
            );

        list_subscriptions ->
            glupbit@websocket@subscription:build_list_subscriptions_message(
                Ticket
            )
    end.

-file("src/glupbit/websocket/connection.gleam", 289).
-spec try_decode(
    binary(),
    gleam@dynamic@decode:decoder(ACNC),
    fun((ACNC) -> ws_message())
) -> ws_message().
try_decode(Text, Decoder, To_msg) ->
    case gleam@json:parse(Text, Decoder) of
        {ok, Data} ->
            To_msg(Data);

        {error, _} ->
            {raw_msg, Text}
    end.

-file("src/glupbit/websocket/connection.gleam", 279).
-spec decode_by_type(binary(), binary()) -> ws_message().
decode_by_type(Text, Type_name) ->
    case Type_name of
        <<"ticker"/utf8>> ->
            try_decode(
                Text,
                glupbit@websocket@subscription:ticker_data_decoder(),
                fun(Field@0) -> {ticker_msg, Field@0} end
            );

        <<"trade"/utf8>> ->
            try_decode(
                Text,
                glupbit@websocket@subscription:trade_data_decoder(),
                fun(Field@0) -> {trade_msg, Field@0} end
            );

        <<"orderbook"/utf8>> ->
            try_decode(
                Text,
                glupbit@websocket@subscription:orderbook_data_decoder(),
                fun(Field@0) -> {orderbook_msg, Field@0} end
            );

        _ ->
            {raw_msg, Text}
    end.

-file("src/glupbit/websocket/connection.gleam", 300).
-spec field_decoder(binary()) -> gleam@dynamic@decode:decoder(binary()).
field_decoder(Name) ->
    gleam@dynamic@decode:field(
        Name,
        {decoder, fun gleam@dynamic@decode:decode_string/1},
        fun(Value) -> gleam@dynamic@decode:success(Value) end
    ).

-file("src/glupbit/websocket/connection.gleam", 262).
-spec decode_data_message(binary()) -> ws_message().
decode_data_message(Text) ->
    case gleam@json:parse(Text, field_decoder(<<"method"/utf8>>)) of
        {ok, <<"LIST_SUBSCRIPTIONS"/utf8>>} ->
            try_decode(
                Text,
                glupbit@websocket@subscription:list_subscriptions_response_decoder(
                    
                ),
                fun(Field@0) -> {subscription_list_msg, Field@0} end
            );

        {ok, _} ->
            {raw_msg, Text};

        {error, _} ->
            case gleam@json:parse(Text, field_decoder(<<"type"/utf8>>)) of
                {ok, Type_name} ->
                    decode_by_type(Text, Type_name);

                {error, _} ->
                    {raw_msg, Text}
            end
    end.

-file("src/glupbit/websocket/connection.gleam", 305).
-spec error_decoder() -> gleam@dynamic@decode:decoder({binary(), binary()}).
error_decoder() ->
    gleam@dynamic@decode:subfield(
        [<<"error"/utf8>>, <<"name"/utf8>>],
        {decoder, fun gleam@dynamic@decode:decode_string/1},
        fun(Name) ->
            gleam@dynamic@decode:subfield(
                [<<"error"/utf8>>, <<"message"/utf8>>],
                {decoder, fun gleam@dynamic@decode:decode_string/1},
                fun(Message) ->
                    gleam@dynamic@decode:success({Name, Message})
                end
            )
        end
    ).

-file("src/glupbit/websocket/connection.gleam", 251).
-spec decode_ws_message(binary()) -> ws_message().
decode_ws_message(Text) ->
    case gleam@json:parse(Text, field_decoder(<<"status"/utf8>>)) of
        {ok, Status} ->
            {status_msg, Status};

        {error, _} ->
            case gleam@json:parse(Text, error_decoder()) of
                {ok, {Name, Message}} ->
                    {error_msg, Name, Message};

                {error, _} ->
                    decode_data_message(Text)
            end
    end.

-file("src/glupbit/websocket/connection.gleam", 162).
-spec ws_message_handler(
    ws_state(ACMT),
    stratus:message(ws_command()),
    stratus:connection()
) -> stratus:next(ws_state(ACMT), ws_command()).
ws_message_handler(State, Msg, Conn) ->
    case Msg of
        {text, Text} ->
            Ws_msg = decode_ws_message(Text),
            New_user_state = (erlang:element(2, State))(
                erlang:element(3, State),
                Ws_msg
            ),
            stratus:continue(
                {ws_state, erlang:element(2, State), New_user_state}
            );

        {binary, Bytes} ->
            case gleam@bit_array:to_string(Bytes) of
                {ok, Text@1} ->
                    Ws_msg@1 = decode_ws_message(Text@1),
                    New_user_state@1 = (erlang:element(2, State))(
                        erlang:element(3, State),
                        Ws_msg@1
                    ),
                    stratus:continue(
                        {ws_state, erlang:element(2, State), New_user_state@1}
                    );

                {error, _} ->
                    stratus:continue(State)
            end;

        {user, Cmd} ->
            Msg_text = build_command_message(Cmd),
            _ = stratus:send_text_message(Conn, Msg_text),
            stratus:continue(State)
    end.

-file("src/glupbit/websocket/connection.gleam", 120).
-spec start_ws(
    gleam@http@request:request(binary()),
    ACMM,
    fun((ACMM, ws_message()) -> ACMM)
) -> {ok, gleam@erlang@process:subject(stratus:internal_message(ws_command()))} |
    {error, stratus:initialization_error()}.
start_ws(Req, User_state, Handler) ->
    Initial_state = {ws_state, Handler, User_state},
    _pipe = stratus:new(Req, Initial_state),
    _pipe@1 = stratus:on_message(_pipe, fun ws_message_handler/3),
    _pipe@2 = stratus:on_close(_pipe@1, fun(_, _) -> nil end),
    _pipe@3 = stratus:start(_pipe@2),
    gleam@result:map(_pipe@3, fun(Started) -> erlang:element(3, Started) end).

-file("src/glupbit/websocket/connection.gleam", 72).
?DOC(" Connect to the public WebSocket.\n").
-spec connect_public(ACMB, fun((ACMB, ws_message()) -> ACMB)) -> {ok,
        gleam@erlang@process:subject(stratus:internal_message(ws_command()))} |
    {error, stratus:initialization_error()}.
connect_public(User_state, Handler) ->
    Req@1 = case gleam@http@request:to(
        <<"https://api.upbit.com/websocket/v1"/utf8>>
    ) of
        {ok, Req} -> Req;
        _assert_fail ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
                        file => <<?FILEPATH/utf8>>,
                        module => <<"glupbit/websocket/connection"/utf8>>,
                        function => <<"connect_public"/utf8>>,
                        line => 76,
                        value => _assert_fail,
                        start => 1993,
                        'end' => 2039,
                        pattern_start => 2004,
                        pattern_end => 2011})
    end,
    start_ws(Req@1, User_state, Handler).

-file("src/glupbit/websocket/connection.gleam", 93).
?DOC(" Connect to the private WebSocket with authentication.\n").
-spec connect_private(
    glupbit@auth:credentials(),
    ACMH,
    fun((ACMH, ws_message()) -> ACMH)
) -> {ok, gleam@erlang@process:subject(stratus:internal_message(ws_command()))} |
    {error, stratus:initialization_error()}.
connect_private(Creds, User_state, Handler) ->
    Req@1 = case gleam@http@request:to(
        <<"https://api.upbit.com/websocket/v1/private"/utf8>>
    ) of
        {ok, Req} -> Req;
        _assert_fail ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
                        file => <<?FILEPATH/utf8>>,
                        module => <<"glupbit/websocket/connection"/utf8>>,
                        function => <<"connect_private"/utf8>>,
                        line => 98,
                        value => _assert_fail,
                        start => 2907,
                        'end' => 2954,
                        pattern_start => 2918,
                        pattern_end => 2925})
    end,
    Token@1 = case glupbit@auth:generate_token(Creds) of
        {ok, Token} -> Token;
        _assert_fail@1 ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
                        file => <<?FILEPATH/utf8>>,
                        module => <<"glupbit/websocket/connection"/utf8>>,
                        function => <<"connect_private"/utf8>>,
                        line => 99,
                        value => _assert_fail@1,
                        start => 2957,
                        'end' => 3006,
                        pattern_start => 2968,
                        pattern_end => 2977})
    end,
    Req@2 = gleam@http@request:set_header(
        Req@1,
        <<"authorization"/utf8>>,
        <<"Bearer "/utf8, Token@1/binary>>
    ),
    start_ws(Req@2, User_state, Handler).

-file("src/glupbit/websocket/connection.gleam", 199).
?DOC(
    " Exponential backoff reconnection loop.\n"
    " Retries until successful, doubling delay each time up to max_delay_ms.\n"
).
-spec reconnect_loop(
    gleam@http@request:request(binary()),
    ACNA,
    fun((ACNA, ws_message()) -> ACNA),
    reconnect_config(),
    integer()
) -> nil.
reconnect_loop(Req, User_state, Handler, Config, Delay_ms) ->
    logger:notice(
        <<<<"[WS] Reconnecting in "/utf8,
                (erlang:integer_to_binary(Delay_ms))/binary>>/binary,
            "ms..."/utf8>>
    ),
    gleam_erlang_ffi:sleep(Delay_ms),
    case start_ws_reconnectable(Req, User_state, Handler, Config) of
        {ok, New_handle} ->
            subscribe(
                New_handle,
                erlang:element(2, Config),
                erlang:element(3, Config)
            ),
            (erlang:element(6, Config))(),
            logger:notice(<<"[WS] Reconnected and re-subscribed"/utf8>>);

        {error, _} ->
            Next_delay = erlang:trunc(
                gleam@float:min(
                    erlang:float(Delay_ms * 2),
                    erlang:float(erlang:element(5, Config))
                )
            ),
            logger:notice(<<"[WS] Reconnection failed, retrying..."/utf8>>),
            reconnect_loop(Req, User_state, Handler, Config, Next_delay)
    end.

-file("src/glupbit/websocket/connection.gleam", 134).
-spec start_ws_reconnectable(
    gleam@http@request:request(binary()),
    ACMQ,
    fun((ACMQ, ws_message()) -> ACMQ),
    reconnect_config()
) -> {ok, gleam@erlang@process:subject(stratus:internal_message(ws_command()))} |
    {error, stratus:initialization_error()}.
start_ws_reconnectable(Req, User_state, Handler, Config) ->
    Initial_state = {ws_state, Handler, User_state},
    _pipe = stratus:new(Req, Initial_state),
    _pipe@1 = stratus:on_message(_pipe, fun ws_message_handler/3),
    _pipe@2 = stratus:on_close(
        _pipe@1,
        fun(State, _) ->
            _ = erlang:spawn(
                fun() ->
                    reconnect_loop(
                        Req,
                        erlang:element(3, State),
                        erlang:element(2, State),
                        Config,
                        erlang:element(4, Config)
                    )
                end
            ),
            nil
        end
    ),
    _pipe@3 = stratus:start(_pipe@2),
    gleam@result:map(_pipe@3, fun(Started) -> erlang:element(3, Started) end).

-file("src/glupbit/websocket/connection.gleam", 83).
?DOC(
    " Connect to the public WebSocket with automatic reconnection.\n"
    " On disconnect, retries with exponential backoff (1s, 2s, 4s, ... up to max_delay_ms).\n"
    " Re-subscribes to the same feeds after successful reconnection.\n"
).
-spec connect_public_with_reconnect(
    ACME,
    fun((ACME, ws_message()) -> ACME),
    reconnect_config()
) -> {ok, gleam@erlang@process:subject(stratus:internal_message(ws_command()))} |
    {error, stratus:initialization_error()}.
connect_public_with_reconnect(User_state, Handler, Config) ->
    Req@1 = case gleam@http@request:to(
        <<"https://api.upbit.com/websocket/v1"/utf8>>
    ) of
        {ok, Req} -> Req;
        _assert_fail ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
                        file => <<?FILEPATH/utf8>>,
                        module => <<"glupbit/websocket/connection"/utf8>>,
                        function => <<"connect_public_with_reconnect"/utf8>>,
                        line => 88,
                        value => _assert_fail,
                        start => 2527,
                        'end' => 2573,
                        pattern_start => 2538,
                        pattern_end => 2545})
    end,
    start_ws_reconnectable(Req@1, User_state, Handler, Config).