gftp

-module(gftp@stream).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/gftp/stream.gleam").
-export(['receive'/2, receive_exact/3, send/2, upgrade_to_ssl/2, downgrade_to_tcp/1, local_address/1, peer_address/1, set_line_mode/1, set_raw_mode/1, receive_next_packet_as_message/1, select_stream_messages/2, controlling_process/2, shutdown/1]).
-export_type([data_stream/0, stream_message/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(
    " The stream module provides an interface for working with a network stream, which can either be a TCP stream or a TLS stream.\n"
    " It abstracts away the differences between the two types of streams and provides a unified interface for reading and writing data.\n"
    "\n"
    " ## Message-based mode\n"
    "\n"
    " For integration with OTP actors and other message-based architectures, the stream module provides\n"
    " a non-blocking alternative to synchronous `receive`. Instead of blocking the calling process,\n"
    " you can use `receive_next_packet_as_message` to arm the socket for a single message delivery,\n"
    " then use `select_stream_messages` to build a `process.Selector` that maps incoming socket\n"
    " messages to your actor's message type.\n"
    "\n"
    " This is the standard BEAM approach: set `{active, once}` on the socket, receive one message,\n"
    " then re-arm. This gives backpressure control and integrates naturally with `gleam_erlang`\n"
    " selectors and OTP actors.\n"
).

-type data_stream() :: {ssl, kafein:ssl_socket(), mug:socket()} |
    {tcp, mug:socket()}.

-type stream_message() :: {packet, bitstring()} |
    stream_closed |
    {stream_error, gftp@result:ftp_error()}.

-file("src/gftp/stream.gleam", 84).
?DOC(
    " Receive a message from the peer.\n"
    "\n"
    " Errors if the socket is closed, if the timeout is reached, or if any other error occurs during the receive operation.\n"
    " The error is returned as a [`mug.Error`]. In case of success, it returns the received data as a `BitArray`.\n"
).
-spec 'receive'(data_stream(), integer()) -> {ok, bitstring()} |
    {error, mug:error()}.
'receive'(Stream, Timeout) ->
    case Stream of
        {ssl, Ssl_socket, _} ->
            kafein:'receive'(Ssl_socket, Timeout);

        {tcp, Tcp_socket} ->
            mug:'receive'(Tcp_socket, Timeout)
    end.

-file("src/gftp/stream.gleam", 95).
?DOC(
    " Receive an exact number of bytes from the peer.\n"
    "\n"
    " Errors if the socket is closed, if the timeout is reached, or if any other error occurs during the receive operation.\n"
    " The error is returned as a [`mug.Error`]. In case of success, it returns the received data as a `BitArray`.\n"
).
-spec receive_exact(data_stream(), integer(), integer()) -> {ok, bitstring()} |
    {error, mug:error()}.
receive_exact(Stream, Size, Timeout) ->
    case Stream of
        {ssl, Ssl_socket, _} ->
            kafein:receive_exact(Ssl_socket, Size, Timeout);

        {tcp, Tcp_socket} ->
            mug:receive_exact(Tcp_socket, Size, Timeout)
    end.

-file("src/gftp/stream.gleam", 110).
?DOC(
    " Sends data over the provided stream.\n"
    " It handles both SSL and TCP streams, abstracting away the differences between them.\n"
    "\n"
    " Returns a Result indicating success (`Nil`) or an error ([`mug.Error`]) if the send operation fails.\n"
).
-spec send(data_stream(), bitstring()) -> {ok, nil} | {error, mug:error()}.
send(Stream, Data) ->
    case Stream of
        {ssl, Ssl_socket, _} ->
            kafein:send(Ssl_socket, Data);

        {tcp, Tcp_socket} ->
            mug:send(Tcp_socket, Data)
    end.

-file("src/gftp/stream.gleam", 118).
?DOC(" Upgrades a TCP stream to an SSL stream using the provided options. If the stream is already an SSL stream, it returns it as is.\n").
-spec upgrade_to_ssl(data_stream(), kafein:wrap_options()) -> {ok,
        data_stream()} |
    {error, kafein:error()}.
upgrade_to_ssl(Stream, Options) ->
    case Stream of
        {ssl, _, _} ->
            {ok, Stream};

        {tcp, Tcp_socket} ->
            case kafein:wrap(stream_ffi:patch_wrap_options(Options), Tcp_socket) of
                {ok, Ssl_socket} ->
                    {ok, {ssl, Ssl_socket, Tcp_socket}};

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

-file("src/gftp/stream.gleam", 134).
?DOC(" Downgrades an SSL stream to a TCP stream. If the stream is already a TCP stream, it returns it as is.\n").
-spec downgrade_to_tcp(data_stream()) -> data_stream().
downgrade_to_tcp(Stream) ->
    case Stream of
        {ssl, _, Tcp_socket} ->
            {tcp, Tcp_socket};

        {tcp, _} ->
            Stream
    end.

-file("src/gftp/stream.gleam", 142).
?DOC(" Retrieves the local address (IP and port) of the underlying TCP socket.\n").
-spec local_address(data_stream()) -> {ok, {binary(), integer()}} |
    {error, mug:error()}.
local_address(Stream) ->
    case Stream of
        {ssl, _, Tcp_socket} ->
            stream_ffi:local_address(Tcp_socket);

        {tcp, Tcp_socket@1} ->
            stream_ffi:local_address(Tcp_socket@1)
    end.

-file("src/gftp/stream.gleam", 150).
?DOC(" Retrieves the peer address (IP and port) of the underlying TCP socket.\n").
-spec peer_address(data_stream()) -> {ok, {binary(), integer()}} |
    {error, mug:error()}.
peer_address(Stream) ->
    case Stream of
        {ssl, _, Tcp_socket} ->
            stream_ffi:peer_address(Tcp_socket);

        {tcp, Tcp_socket@1} ->
            stream_ffi:peer_address(Tcp_socket@1)
    end.

-file("src/gftp/stream.gleam", 159).
?DOC(
    " Set the socket to line-delimited packet mode ({packet, line}).\n"
    " In this mode, recv returns one complete line per call.\n"
).
-spec set_line_mode(data_stream()) -> nil.
set_line_mode(Stream) ->
    case Stream of
        {ssl, Ssl_socket, _} ->
            stream_ffi:set_ssl_packet_line(Ssl_socket);

        {tcp, Tcp_socket} ->
            stream_ffi:set_tcp_packet_line(Tcp_socket)
    end.

-file("src/gftp/stream.gleam", 168).
?DOC(
    " Set the socket back to raw packet mode ({packet, raw}).\n"
    " In this mode, recv returns whatever data is available.\n"
).
-spec set_raw_mode(data_stream()) -> nil.
set_raw_mode(Stream) ->
    case Stream of
        {ssl, Ssl_socket, _} ->
            stream_ffi:set_ssl_packet_raw(Ssl_socket);

        {tcp, Tcp_socket} ->
            stream_ffi:set_tcp_packet_raw(Tcp_socket)
    end.

-file("src/gftp/stream.gleam", 182).
?DOC(
    " Set `{active, once}` on the underlying socket so that the next packet is\n"
    " delivered as an Erlang message to the calling process instead of being\n"
    " buffered for a synchronous `receive` call.\n"
    "\n"
    " After each message arrives you must call this again to receive the next one.\n"
    " Use `select_stream_messages` to build a selector that maps the raw socket\n"
    " messages into `StreamMessage` values.\n"
).
-spec receive_next_packet_as_message(data_stream()) -> nil.
receive_next_packet_as_message(Stream) ->
    case Stream of
        {ssl, Ssl_socket, _} ->
            kafein:receive_next_packet_as_message(Ssl_socket);

        {tcp, Tcp_socket} ->
            mug:receive_next_packet_as_message(Tcp_socket)
    end.

-file("src/gftp/stream.gleam", 208).
?DOC(
    " Add handlers for both TCP and SSL socket messages to a `process.Selector`.\n"
    "\n"
    " The `mapper` function converts a `StreamMessage` into your actor's message\n"
    " type. Handlers for both TCP and SSL messages are registered; only the one\n"
    " matching the actual socket type will ever fire.\n"
    "\n"
    " ```gleam\n"
    " import gleam/erlang/process\n"
    " import gftp/stream.{type StreamMessage}\n"
    "\n"
    " type MyMessage {\n"
    "   GotStream(StreamMessage)\n"
    "   // ... other variants\n"
    " }\n"
    "\n"
    " let selector =\n"
    "   process.new_selector()\n"
    "   |> stream.select_stream_messages(GotStream)\n"
    " ```\n"
).
-spec select_stream_messages(
    gleam@erlang@process:selector(JXB),
    fun((stream_message()) -> JXB)
) -> gleam@erlang@process:selector(JXB).
select_stream_messages(Selector, Mapper) ->
    _pipe = Selector,
    _pipe@1 = mug:select_tcp_messages(
        _pipe,
        fun(Tcp_msg) -> Mapper(case Tcp_msg of
                    {packet, _, Data} ->
                        {packet, Data};

                    {socket_closed, _} ->
                        stream_closed;

                    {tcp_error, _, Err} ->
                        {stream_error, {socket, Err}}
                end) end
    ),
    kafein:select_ssl_messages(_pipe@1, fun(Ssl_msg) -> Mapper(case Ssl_msg of
                    {packet, _, Data@1} ->
                        {packet, Data@1};

                    {socket_closed, _} ->
                        stream_closed;

                    {ssl_error, _, Err@1} ->
                        {stream_error, {tls, Err@1}}
                end) end).

-file("src/gftp/stream.gleam", 234).
?DOC(
    " Transfer socket ownership to a different process.\n"
    "\n"
    " This is necessary when a data stream is created in one process (e.g. an OTP actor)\n"
    " but message-based I/O (`receive_next_packet_as_message`) will be used from another.\n"
    " The receiving process must be the socket's controlling process to get `{active, once}` messages.\n"
).
-spec controlling_process(data_stream(), gleam@erlang@process:pid_()) -> {ok,
        nil} |
    {error, mug:error()}.
controlling_process(Stream, Pid) ->
    case Stream of
        {ssl, Ssl_socket, Tcp_socket} ->
            case stream_ffi:tcp_controlling_process(Tcp_socket, Pid) of
                {ok, _} ->
                    stream_ffi:ssl_controlling_process(Ssl_socket, Pid);

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

        {tcp, Tcp_socket@1} ->
            stream_ffi:tcp_controlling_process(Tcp_socket@1, Pid)
    end.

-file("src/gftp/stream.gleam", 251).
?DOC(
    " Shuts down the provided stream, whether it's an SSL or TCP stream.\n"
    " It abstracts away the differences between the two types of streams and provides a unified interface for shutting them down.\n"
).
-spec shutdown(data_stream()) -> {ok, nil} | {error, mug:error()}.
shutdown(Stream) ->
    case Stream of
        {ssl, Ssl_socket, _} ->
            kafein_ffi:shutdown(Ssl_socket);

        {tcp, Tcp_socket} ->
            mug_ffi:shutdown(Tcp_socket)
    end.