gleeth

-module(gleeth@watcher).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/gleeth/watcher.gleam").
-export([default_config/0, 'receive'/2, stop/1, start_with_subject/3, start_with_config/2, start/1]).
-export_type([block_event/0, watcher_config/0, watcher/0, watcher_message/0, watcher_state/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(
    " Block watcher using a BEAM actor (OTP gen_server).\n"
    "\n"
    " Spawns a supervised actor process that polls for new blocks and sends\n"
    " events to a caller-owned Subject. Uses the OTP actor pattern for proper\n"
    " lifecycle management and graceful shutdown.\n"
    "\n"
    " ## Examples\n"
    "\n"
    " ```gleam\n"
    " let assert Ok(w) = watcher.start(provider)\n"
    "\n"
    " // Block until a new block arrives (or timeout after 30s)\n"
    " case watcher.receive(w, 30_000) {\n"
    "   Ok(watcher.NewBlock(number, hash)) -> io.println(\"Block: \" <> number)\n"
    "   Error(Nil) -> io.println(\"Timeout\")\n"
    " }\n"
    "\n"
    " watcher.stop(w)\n"
    " ```\n"
).

-type block_event() :: {new_block, binary(), binary()}.

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

-opaque watcher() :: {watcher,
        gleam@erlang@process:subject(block_event()),
        gleam@erlang@process:subject(watcher_message())}.

-type watcher_message() :: {init,
        gleam@erlang@process:subject(watcher_message())} |
    poll |
    stop.

-type watcher_state() :: {watcher_state,
        gleeth@provider:provider(),
        gleam@erlang@process:subject(block_event()),
        gleam@erlang@process:subject(watcher_message()),
        watcher_config(),
        binary()}.

-file("src/gleeth/watcher.gleam", 42).
?DOC(" Default config: poll every 2 seconds.\n").
-spec default_config() -> watcher_config().
default_config() ->
    {watcher_config, 2000}.

-file("src/gleeth/watcher.gleam", 143).
?DOC(
    " Receive the next block event, blocking up to `timeout_ms` milliseconds.\n"
    " Returns `Error(Nil)` on timeout.\n"
    "\n"
    " ## Examples\n"
    "\n"
    " ```gleam\n"
    " case watcher.receive(w, 10_000) {\n"
    "   Ok(watcher.NewBlock(number, hash)) -> // new block\n"
    "   Error(Nil) -> // timeout\n"
    " }\n"
    " ```\n"
).
-spec 'receive'(watcher(), integer()) -> {ok, block_event()} | {error, nil}.
'receive'(Watcher, Timeout_ms) ->
    gleam@erlang@process:'receive'(erlang:element(2, Watcher), Timeout_ms).

-file("src/gleeth/watcher.gleam", 154).
?DOC(
    " Stop the watcher gracefully. The actor process exits cleanly.\n"
    "\n"
    " ## Examples\n"
    "\n"
    " ```gleam\n"
    " watcher.stop(w)\n"
    " ```\n"
).
-spec stop(watcher()) -> nil.
stop(Watcher) ->
    gleam@erlang@process:send(erlang:element(3, Watcher), stop).

-file("src/gleeth/watcher.gleam", 223).
-spec emit_blocks_range(
    gleeth@provider:provider(),
    gleam@erlang@process:subject(block_event()),
    integer(),
    integer()
) -> nil.
emit_blocks_range(Provider, Events, From, To) ->
    case From > To of
        true ->
            nil;

        false ->
            Block_hex = gleeth@utils@hex:from_int(From),
            Hash = case gleeth@rpc@methods:get_block_by_number(
                Provider,
                Block_hex
            ) of
                {ok, Block} ->
                    erlang:element(3, Block);

                {error, _} ->
                    <<""/utf8>>
            end,
            gleam@erlang@process:send(Events, {new_block, Block_hex, Hash}),
            emit_blocks_range(Provider, Events, From + 1, To)
    end.

-file("src/gleeth/watcher.gleam", 210).
-spec emit_new_blocks(
    gleeth@provider:provider(),
    gleam@erlang@process:subject(block_event()),
    binary(),
    binary()
) -> nil.
emit_new_blocks(Provider, Events, Last_block, Current_block) ->
    case {gleeth@utils@hex:to_int(Last_block),
        gleeth@utils@hex:to_int(Current_block)} of
        {{ok, Last_num}, {ok, Current_num}} ->
            emit_blocks_range(Provider, Events, Last_num + 1, Current_num);

        {_, _} ->
            nil
    end.

-file("src/gleeth/watcher.gleam", 243).
-spec is_newer(binary(), binary()) -> boolean().
is_newer(Current, Last) ->
    case {gleeth@utils@hex:to_int(Current), gleeth@utils@hex:to_int(Last)} of
        {{ok, C}, {ok, L}} ->
            C > L;

        {_, _} ->
            false
    end.

-file("src/gleeth/watcher.gleam", 162).
-spec handle_message(watcher_state(), watcher_message()) -> gleam@otp@actor:next(watcher_state(), watcher_message()).
handle_message(State, Message) ->
    case Message of
        stop ->
            gleam@otp@actor:stop();

        {init, Self} ->
            New_state = {watcher_state,
                erlang:element(2, State),
                erlang:element(3, State),
                Self,
                erlang:element(5, State),
                erlang:element(6, State)},
            gleam@erlang@process:send_after(
                Self,
                erlang:element(2, erlang:element(5, New_state)),
                poll
            ),
            gleam@otp@actor:continue(New_state);

        poll ->
            New_state@1 = case gleeth@rpc@methods:get_block_number(
                erlang:element(2, State)
            ) of
                {ok, Current_block} ->
                    case is_newer(Current_block, erlang:element(6, State)) of
                        true ->
                            emit_new_blocks(
                                erlang:element(2, State),
                                erlang:element(3, State),
                                erlang:element(6, State),
                                Current_block
                            ),
                            {watcher_state,
                                erlang:element(2, State),
                                erlang:element(3, State),
                                erlang:element(4, State),
                                erlang:element(5, State),
                                Current_block};

                        false ->
                            State
                    end;

                {error, _} ->
                    State
            end,
            gleam@erlang@process:send_after(
                erlang:element(4, New_state@1),
                erlang:element(2, erlang:element(5, New_state@1)),
                poll
            ),
            gleam@otp@actor:continue(New_state@1)
    end.

-file("src/gleeth/watcher.gleam", 98).
?DOC(
    " Start watching, sending events to a caller-provided Subject.\n"
    " Use this when another process needs to own the receiving Subject\n"
    " (e.g. the event watcher actor).\n"
).
-spec start_with_subject(
    gleeth@provider:provider(),
    watcher_config(),
    gleam@erlang@process:subject(block_event())
) -> {ok, watcher()} | {error, binary()}.
start_with_subject(Provider, Config, Events) ->
    case gleeth@rpc@methods:get_block_number(Provider) of
        {ok, Initial_block} ->
            State = {watcher_state,
                Provider,
                Events,
                gleam@erlang@process:new_subject(),
                Config,
                Initial_block},
            case begin
                _pipe = gleam@otp@actor:new(State),
                _pipe@1 = gleam@otp@actor:on_message(
                    _pipe,
                    fun handle_message/2
                ),
                gleam@otp@actor:start(_pipe@1)
            end of
                {ok, Started} ->
                    Actor_subject = erlang:element(3, Started),
                    gleam@erlang@process:send(
                        Actor_subject,
                        {init, Actor_subject}
                    ),
                    {ok, {watcher, Events, Actor_subject}};

                {error, _} ->
                    {error, <<"Failed to start watcher actor"/utf8>>}
            end;

        {error, _} ->
            {error, <<"Failed to fetch initial block number"/utf8>>}
    end.

-file("src/gleeth/watcher.gleam", 88).
?DOC(
    " Start watching with custom configuration.\n"
    "\n"
    " ## Examples\n"
    "\n"
    " ```gleam\n"
    " let config = watcher.WatcherConfig(poll_interval_ms: 500)\n"
    " let assert Ok(w) = watcher.start_with_config(provider, config)\n"
    " ```\n"
).
-spec start_with_config(gleeth@provider:provider(), watcher_config()) -> {ok,
        watcher()} |
    {error, binary()}.
start_with_config(Provider, Config) ->
    start_with_subject(Provider, Config, gleam@erlang@process:new_subject()).

-file("src/gleeth/watcher.gleam", 76).
?DOC(
    " Start watching for new blocks with default config (poll every 2s).\n"
    "\n"
    " ## Examples\n"
    "\n"
    " ```gleam\n"
    " let assert Ok(w) = watcher.start(provider)\n"
    " ```\n"
).
-spec start(gleeth@provider:provider()) -> {ok, watcher()} | {error, binary()}.
start(Provider) ->
    start_with_config(Provider, default_config()).