postgleam

-module(postgleam@replication).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/postgleam/replication.gleam").
-export([connect/1, 'query'/3, send_standby_status/6, disconnect/1, start_streaming/3, stop_streaming/2, receive_wal_message/2, encode_lsn/1, decode_lsn/1]).
-export_type([lsn/0, wal_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.

-type lsn() :: {lsn, integer()}.

-type wal_message() :: {x_log_data,
        integer(),
        integer(),
        integer(),
        bitstring()} |
    {primary_keepalive, integer(), integer(), boolean()}.

-file("src/postgleam/replication.gleam", 67).
?DOC(" Start a replication connection by connecting with the replication parameter\n").
-spec connect(postgleam@config:config()) -> {ok,
        postgleam@connection:connection_state()} |
    {error, postgleam@error:error()}.
connect(Cfg) ->
    Repl_config = {config,
        erlang:element(2, Cfg),
        erlang:element(3, Cfg),
        erlang:element(4, Cfg),
        erlang:element(5, Cfg),
        erlang:element(6, Cfg),
        erlang:element(7, Cfg),
        erlang:element(8, Cfg),
        [{<<"replication"/utf8>>, <<"database"/utf8>>} | erlang:element(9, Cfg)],
        erlang:element(10, Cfg),
        erlang:element(11, Cfg),
        erlang:element(12, Cfg),
        erlang:element(13, Cfg)},
    postgleam@connection:connect(Repl_config).

-file("src/postgleam/replication.gleam", 83).
?DOC(
    " Execute a simple query on a replication connection.\n"
    " Used for CREATE_REPLICATION_SLOT etc.\n"
).
-spec 'query'(postgleam@connection:connection_state(), binary(), integer()) -> {ok,
        {list(postgleam@connection:simple_query_result()),
            postgleam@connection:connection_state()}} |
    {error, postgleam@error:error()}.
'query'(State, Sql, Timeout) ->
    postgleam@connection:simple_query(State, Sql, Timeout).

-file("src/postgleam/replication.gleam", 136).
?DOC(
    " Send a standby status update to the server.\n"
    " All LSN values are 64-bit unsigned integers (microseconds since 2000-01-01).\n"
).
-spec send_standby_status(
    postgleam@connection:connection_state(),
    integer(),
    integer(),
    integer(),
    integer(),
    boolean()
) -> {ok, postgleam@connection:connection_state()} |
    {error, postgleam@error:error()}.
send_standby_status(
    State,
    Written,
    Flushed,
    Applied,
    Timestamp,
    Reply_requested
) ->
    Reply_byte = case Reply_requested of
        true ->
            1;

        false ->
            0
    end,
    Data = <<"r"/utf8,
        Written:64,
        Flushed:64,
        Applied:64,
        Timestamp:64,
        Reply_byte:8>>,
    postgleam@connection:send_message(State, {copy_data_msg, Data}).

-file("src/postgleam/replication.gleam", 172).
?DOC(" Disconnect a replication connection\n").
-spec disconnect(postgleam@connection:connection_state()) -> nil.
disconnect(State) ->
    postgleam@connection:disconnect(State).

-file("src/postgleam/replication.gleam", 178).
-spec recv_copy_both_response(
    postgleam@connection:connection_state(),
    integer()
) -> {ok, postgleam@connection:connection_state()} |
    {error, postgleam@error:error()}.
recv_copy_both_response(State, Timeout) ->
    gleam@result:'try'(
        postgleam@connection:receive_message(State, Timeout),
        fun(_use0) ->
            {Msg, State@1} = _use0,
            case Msg of
                {copy_both_response, _, _} ->
                    {ok, State@1};

                {error_response, Fields} ->
                    Pg_fields = postgleam@error:parse_error_fields(Fields),
                    {error,
                        {pg_error, Pg_fields, erlang:element(3, State@1), none}};

                {notice_response, _} ->
                    recv_copy_both_response(State@1, Timeout);

                _ ->
                    {error,
                        {protocol_error, <<"Expected CopyBothResponse"/utf8>>}}
            end
        end
    ).

-file("src/postgleam/replication.gleam", 93).
?DOC(
    " Start streaming replication.\n"
    " Sends the START_REPLICATION command and waits for CopyBothResponse.\n"
).
-spec start_streaming(
    postgleam@connection:connection_state(),
    binary(),
    integer()
) -> {ok, postgleam@connection:connection_state()} |
    {error, postgleam@error:error()}.
start_streaming(State, Sql, Timeout) ->
    gleam@result:'try'(
        postgleam@connection:send_message(State, {simple_query, Sql}),
        fun(State@1) -> recv_copy_both_response(State@1, Timeout) end
    ).

-file("src/postgleam/replication.gleam", 219).
-spec recv_ready_after_stream(
    postgleam@connection:connection_state(),
    integer()
) -> {ok, postgleam@connection:connection_state()} |
    {error, postgleam@error:error()}.
recv_ready_after_stream(State, Timeout) ->
    gleam@result:'try'(
        postgleam@connection:receive_message(State, Timeout),
        fun(_use0) ->
            {Msg, State@1} = _use0,
            case Msg of
                {ready_for_query, Status} ->
                    {ok,
                        {connection_state,
                            erlang:element(2, State@1),
                            erlang:element(3, State@1),
                            erlang:element(4, State@1),
                            erlang:element(5, State@1),
                            Status,
                            erlang:element(7, State@1)}};

                _ ->
                    recv_ready_after_stream(State@1, Timeout)
            end
        end
    ).

-file("src/postgleam/replication.gleam", 198).
-spec recv_stream_end(postgleam@connection:connection_state(), integer()) -> {ok,
        postgleam@connection:connection_state()} |
    {error, postgleam@error:error()}.
recv_stream_end(State, Timeout) ->
    gleam@result:'try'(
        postgleam@connection:receive_message(State, Timeout),
        fun(_use0) ->
            {Msg, State@1} = _use0,
            case Msg of
                {command_complete, _} ->
                    recv_ready_after_stream(State@1, Timeout);

                copy_done ->
                    recv_stream_end(State@1, Timeout);

                {copy_data, _} ->
                    recv_stream_end(State@1, Timeout);

                {error_response, Fields} ->
                    Pg_fields = postgleam@error:parse_error_fields(Fields),
                    {error,
                        {pg_error, Pg_fields, erlang:element(3, State@1), none}};

                _ ->
                    recv_stream_end(State@1, Timeout)
            end
        end
    ).

-file("src/postgleam/replication.gleam", 161).
?DOC(" Stop streaming by sending CopyDone, then receive CommandComplete + ReadyForQuery\n").
-spec stop_streaming(postgleam@connection:connection_state(), integer()) -> {ok,
        postgleam@connection:connection_state()} |
    {error, postgleam@error:error()}.
stop_streaming(State, Timeout) ->
    gleam@result:'try'(
        postgleam@connection:send_message(State, copy_done_msg),
        fun(State@1) -> recv_stream_end(State@1, Timeout) end
    ).

-file("src/postgleam/replication.gleam", 231).
-spec parse_wal_message(bitstring()) -> {ok, wal_message()} | {error, nil}.
parse_wal_message(Data) ->
    case Data of
        <<16#77, Wal_start:64, Wal_end:64, Server_time:64, Rest/binary>> ->
            {ok, {x_log_data, Wal_start, Wal_end, Server_time, Rest}};

        <<16#6B, Wal_end@1:64, Server_time@1:64, Reply:8>> ->
            {ok, {primary_keepalive, Wal_end@1, Server_time@1, Reply =:= 1}};

        _ ->
            {error, nil}
    end.

-file("src/postgleam/replication.gleam", 106).
?DOC(
    " Receive the next WAL message from a streaming replication connection.\n"
    " Returns either a parsed WAL message or an indication that streaming has ended.\n"
).
-spec receive_wal_message(postgleam@connection:connection_state(), integer()) -> {ok,
        {{ok, wal_message()} | {error, nil},
            postgleam@connection:connection_state()}} |
    {error, postgleam@error:error()}.
receive_wal_message(State, Timeout) ->
    gleam@result:'try'(
        postgleam@connection:receive_message(State, Timeout),
        fun(_use0) ->
            {Msg, State@1} = _use0,
            case Msg of
                {copy_data, Data} ->
                    case parse_wal_message(Data) of
                        {ok, Wal_msg} ->
                            {ok, {{ok, Wal_msg}, State@1}};

                        {error, nil} ->
                            {ok, {{error, nil}, State@1}}
                    end;

                copy_done ->
                    {ok, {{error, nil}, State@1}};

                {error_response, Fields} ->
                    Pg_fields = postgleam@error:parse_error_fields(Fields),
                    {error,
                        {pg_error, Pg_fields, erlang:element(3, State@1), none}};

                _ ->
                    {ok, {{error, nil}, State@1}}
            end
        end
    ).

-file("src/postgleam/replication.gleam", 32).
?DOC(" Encode an LSN integer to its string representation (e.g., \"1/F73E0220\")\n").
-spec encode_lsn(integer()) -> {ok, binary()} | {error, nil}.
encode_lsn(Lsn) ->
    Max_uint64 = 18446744073709551615,
    case (Lsn >= 0) andalso (Lsn =< Max_uint64) of
        true ->
            File_id = postgleam_ffi:int_shr(Lsn, 32),
            Offset = postgleam_ffi:int_band(Lsn, 16#FFFFFFFF),
            {ok,
                <<<<(postgleam_ffi:int_to_hex(File_id))/binary, "/"/utf8>>/binary,
                    (postgleam_ffi:int_to_hex(Offset))/binary>>};

        false ->
            {error, nil}
    end.

-file("src/postgleam/replication.gleam", 45).
?DOC(" Decode an LSN string to its integer representation\n").
-spec decode_lsn(binary()) -> {ok, integer()} | {error, nil}.
decode_lsn(Lsn) ->
    case gleam@string:split(Lsn, <<"/"/utf8>>) of
        [File_id_str, Offset_str] ->
            case (((string:length(File_id_str) > 0) andalso (string:length(
                File_id_str
            )
            =< 8))
            andalso (string:length(Offset_str) > 0))
            andalso (string:length(Offset_str) =< 8) of
                true ->
                    gleam@result:'try'(
                        postgleam_ffi:hex_to_int(File_id_str),
                        fun(File_id) ->
                            gleam@result:'try'(
                                postgleam_ffi:hex_to_int(Offset_str),
                                fun(Offset) ->
                                    {ok,
                                        postgleam_ffi:int_bor(
                                            postgleam_ffi:int_shl(File_id, 32),
                                            Offset
                                        )}
                                end
                            )
                        end
                    );

                false ->
                    {error, nil}
            end;

        _ ->
            {error, nil}
    end.